Redis队列自研组件

     背景

        年初的时候设计实践过一个课题：SpringBoot+Redis实现不重复消费的队列，并用它开发了一个年夜饭下单和制作的服务。不知道大家还有没有印象。完成这个课题后，我兴致勃勃的把它运用到了项目里面，可谁曾想，运行不久后运维人员就找到了我，说我的功能有问题，而且问题很大。具体表现在用户下单之后，很久都没有收到订单完成的消息，后台服务里的日志也是残缺的，只查到了开始制作年夜饭的日志，没有年夜饭制作过程和完成的日志。

        为此，我花了大量的时间和精力去分析程序，定位问题，最终发现，是运维人员在系统上线时将生产服务和UAT服务的redis服务地址给配成了同一个，这就导致生产上的订单进入Redis队列后，被UAT服务给消费了，而UAT和生产又是不同的数据库，自然导致UAT上通过队列中的主键在数据库中找不到相关数据，从而消费失败的结果。而这些失败的信息全都记录在了UAT的服务器上，生产服务器中自然很难分析和定位到问题。

        要解决这个问题其实很简单，只需要把channelTopic改为从配置文件中获取，然后生产、UAT环境配置不同的字符串即可。

        但实际上真的只做这些就够了吗？如果下次再发生不可预料的问题，我还要花那么多的时间吭哧吭哧的去看日志，调程序，定位问题吗？答案是否定的！

      原服务整改

        结合生产环境产生的问题，痛定思痛，我决定对原来的服务进行一轮大整改，优化服务的可维护性，可测试性。在我的构想中，新的服务需要有以下功能：

        1、保证原有的“年夜饭”功能稳定正常的运行。

        2、可以查询哪些订单还未开始处理

        3、可以查询哪些订单已经处理，以及处理结果。

        4、可以清空N天以前的处理成功的订单。

        5、可以清空待处理的订单

        6、对于已经处理但处理失败的订单，可以一键重新处理

        7、待处理订单插队

        我的设想是，通过redisTemplate.opsForZSet()方法创建两个新队列：待办队列和已办队列，在下单时，插入一条数据到待办队列，在处理任务时，从待办队列中删除该数据，在处理完成后，插入一条数据到已办队列，这样，通过查询待办队列，已办队列，就可以知道哪些任务还在排队，哪些任务已经完成了。

        以下是我的程序整改过程，老粉可以对比上一篇博客看看两者的不同之处。

#以下是application.yml配置server:port: 19200servlet:context-path: /leiximax-http-header-size: 102400
spring:redis:database: 9host: 127.0.0.1port: 6379password:jedis:pool:max-active: 8max-wait: -1max-idle: 8min-idle: 0
leixi:redis-queue-key: NEW_YEAR_DINNER_DEV

//以下是Java程序代码：
/*** Redis配置** @author leixiyueqi* @since 2024/06/15 22:00*/
@Configuration
public class RedisConfig {@Value("${leixi.redis-queue-key}")private String REDIS_QUEUE_KEY ;@Beanpublic ChannelTopic topic() {return new ChannelTopic(REDIS_QUEUE_KEY+ Constant.WORKING_QUEUE_SUFFIX);}@Beanpublic MessageListenerAdapter messageListenerAdapter(DinnerListener listener) {return new MessageListenerAdapter(listener);}@Beanpublic RedisMessageListenerContainer redisContainer(RedisConnectionFactory redisConnectionFactory,MessageListenerAdapter messageListenerAdapter,ChannelTopic topic) {RedisMessageListenerContainer container = new RedisMessageListenerContainer();container.setConnectionFactory(redisConnectionFactory);container.addMessageListener(messageListenerAdapter, topic);return container;}
}/*** 订单处理控制层* * @author leixiyueqi* @since 2024/06/15 22:00*/
@RestController
public class DinnerController {private int i = 0;@Autowiredprivate DinnerService service;@Value("${leixi.redis-queue-key}")private String REDIS_QUEUE_KEY ;@GetMapping("/orderDinner")public Object orderDinner() {OrderEntity entity = new OrderEntity();entity.setOrderCode("Order" + (++i));entity.setCustomerName("第"+i+"位客户");return service.orderNewYearEveDinner(entity);}@Autowiredprivate RedisTemplate<String, String> redisTemplate;@GetMapping("/getPendingOrder")public Object getPendingOrder() {return redisTemplate.opsForZSet().range(REDIS_QUEUE_KEY+ Constant.PENDING_SUFFIX, 0, -1);}@GetMapping("/cleanPendingOrder")public Object cleanPendingOrder() {Set<String> set = redisTemplate.opsForZSet().range(REDIS_QUEUE_KEY+ Constant.PENDING_SUFFIX, 0, -1);if(set.size() > 0) {redisTemplate.opsForZSet().remove(REDIS_QUEUE_KEY + Constant.PENDING_SUFFIX, set.toArray());}return "待处理订单已被清空！";}@GetMapping("/getHandledOrder")public Object getHandledOrder() {return redisTemplate.opsForZSet().range(REDIS_QUEUE_KEY+Constant.HANDLED_SUFFIX, 0, -1);}@GetMapping("/cleanOldSucceedOrder")public Object cleanHandledOrder(@RequestParam("day") Integer day) {Set<String> set = redisTemplate.opsForZSet().rangeByScore(REDIS_QUEUE_KEY+Constant.HANDLED_SUFFIX, 0, Constant.getScoreByDate() - day);set.forEach(s -> {JSONObject obj = JSONObject.parseObject(s);if (obj.getString("result").equals("SUCCESS")) {redisTemplate.opsForZSet().remove(REDIS_QUEUE_KEY+Constant.HANDLED_SUFFIX, s);}});return redisTemplate.opsForZSet().range(REDIS_QUEUE_KEY+Constant.HANDLED_SUFFIX, 0, System.currentTimeMillis());}/*** 这里还有最后一个问题， 把已办里的错误的信息摘除出来，重新走请求。并且反馈哪些信息重新走了请求。*/@GetMapping("/restartFailedOrder")public Object restartFailedOrder() {Set<String> set = redisTemplate.opsForZSet().range(REDIS_QUEUE_KEY+Constant.HANDLED_SUFFIX, 0, -1);StringBuilder sb = new StringBuilder();set.forEach(s -> {JSONObject obj = JSONObject.parseObject(s);if (!obj.getString("result").equals("SUCCESS")) {redisTemplate.opsForZSet().remove(REDIS_QUEUE_KEY+Constant.HANDLED_SUFFIX, s);OrderEntity entity = JSON.parseObject(obj.getString("msg"), OrderEntity.class);service.orderNewYearEveDinner(entity);sb.append(entity.getOrderCode()).append(",");}});return "以下订单号被重启: "+ sb;}@GetMapping("/cutInLineJob")public Object cutInLineJob(@RequestParam("orderCode")  String orderCode) {Set<String> set = redisTemplate.opsForZSet().range(REDIS_QUEUE_KEY+ Constant.PENDING_SUFFIX, 0, -1);for (String s : set) {OrderEntity obj = JSONObject.parseObject(s, OrderEntity.class);if (obj.getOrderCode().equals(orderCode)) {CompletableFuture.runAsync(() -> {service.doListenerWork(s);});return "订单 " + orderCode + " 插队成功！";}}return " 插队失败，该订单已经在制作了！";}
}/**** @author leixiyueqi* @since 2024/06/15 22:00*/
@Data
public class OrderEntity implements Serializable {/*** 客户姓名*/private String customerName;/*** 订单号*/private String orderCode;/*** 菜单*/List<String> menus;/*** 出餐状态*/private String dinnerState;/*** 做饭开始时间*/private String dinnerStartTime;/*** 做饭结束时间*/private String dinnerEndTime;/*** 备注*/private String remark;
}/*** 监听类*  * @author leixiyueqi* @since 2024/06/15 22:00*/
@Component
public class DinnerListener implements MessageListener {@Autowiredprivate DinnerService service;private final Object lock = new Object();@Overridepublic void onMessage(Message message, byte[] pattern)  {synchronized (lock) {service.doListenerWork(message.toString());}}
}/**** @author leixiyueqi* @since 2024/06/15 22:00*/
@Slf4j
@Service
public class DinnerService {@Autowiredprivate RedisTemplate<String, String> redisTemplate;@Value("${leixi.redis-queue-key}")private String REDIS_QUEUE_KEY ;/*** 年夜饭下单** @param req 订单信息* @return*/public Object orderNewYearEveDinner(OrderEntity req) {// 存储订单信息saveOrder(req);// 异步开始做菜redisTemplate.delete(JSON.toJSONString(req));redisTemplate.opsForZSet().add(REDIS_QUEUE_KEY+ Constant.PENDING_SUFFIX, JSON.toJSONString(req), Constant.getScoreByDate());redisTemplate.convertAndSend(REDIS_QUEUE_KEY+ Constant.WORKING_QUEUE_SUFFIX, JSON.toJSONString(req));return "您已成功下单，订单号为"+ req.getOrderCode()+"，后厨正在准备预制菜!";}/*** 这里模拟的是做年夜饭的过程方法，该方法用时较长，整个过程需要10秒，但是，这个过程中存在多种意外，该方法可能失败** @param req 订单信息*/public void doNewYearEveDinner(OrderEntity req) throws Exception {System.out.println("开始做订单 " + req.getOrderCode() + " 的年夜饭");Thread.sleep(10000);// 这里写个方法模拟报错的场景int i = new Random().nextInt(6) + 1;if (i ==4) {throw new Exception("厨师跑了");}if (i ==5) {throw new Exception("食物跑了");}if (i ==6) {throw new Exception("厨房着火了");}System.out.println("订单 " + req.getOrderCode() + " 的年夜饭已经完成");}private void saveOrder(OrderEntity req) {//这里假设做的是订单入库操作System.out.println("订单 " + req.getOrderCode() + " 已经入库， 做饭开始时间为 "+ new Date());}/*** 根据订单编号修改订单信息** @param orderCode 订单编号* @param dinnerStatus* @param remark*/public void updateOrder(String orderCode, String dinnerStatus, String remark) {// 根据订单编号修改订单的出餐结束时间，出餐状态，备注等信息。System.out.println("更新订单 "+ orderCode +" 信息，做饭结束时间为 "+ new Date() + ", 出餐状态为"+ dinnerStatus +", 备注为 " +remark);}public void doListenerWork(String message) {Boolean flag = redisTemplate.opsForValue().setIfAbsent(message, "1", 1, TimeUnit.DAYS);// 加锁失败，已有消费端在此时对此消息进行处理，这里不再做处理if (!flag) {return;}redisTemplate.opsForZSet().remove(REDIS_QUEUE_KEY+ Constant.PENDING_SUFFIX, message);OrderEntity param = CastUtils.cast(JSON.parseObject(message, OrderEntity.class));JSONObject obj = new JSONObject();obj.put("msg", message);try {obj.put("server", InetAddress.getLocalHost().getHostAddress());this.doNewYearEveDinner(param);this.updateOrder(param.getOrderCode(), "SUCCESS", "成功");obj.put("result", "SUCCESS");}catch (Exception e) {e.printStackTrace();this.updateOrder(param.getOrderCode(), "FAIL", e.getMessage());obj.put("result", "FAIL");obj.put("desc", e.getMessage());}finally {obj.put("endTime", new Date());redisTemplate.opsForZSet().add(REDIS_QUEUE_KEY+Constant.HANDLED_SUFFIX, obj.toJSONString(), Constant.getScoreByDate());}}/*** 静态工具类* * @author leixiyueqi* @since 2024/06/15 22:00*/
public class Constant {// 工作队列后缀public static final String WORKING_QUEUE_SUFFIX = "_QUEUE";//待处理工作队列后缀public static final String PENDING_SUFFIX = "_PENDING";// 已处理工作后缀public static final String HANDLED_SUFFIX = "_HANDLED";//一天的毫秒数private static final Integer ONE_DAY_MINI = 86400000;/*** 根据当前日期计算队列的分数** @return*/public static Integer getScoreByDate() {return (int)System.currentTimeMillis()/ONE_DAY_MINI;}
}

      接口测试

        1、年夜饭下单

        

        2、查询待处理订单

        

        3、查询已处理订单

        

        4、清空已处理且成功的订单

        

        5、清空待处理订单

        

        6、一键重启处理失败的订单

        

        7、订单插队

        

      组件化封装

        完成了以上测试，基本上我想要的功能都已经实现了。但是仔细想了下，上述的功能里除了第一个下单接口是跟业务相关的，剩下的所有接口都是业务无关的。如果我们公司主营业务变了，从年夜饭变成中秋做月饼，端午包棕子，本服务中的大部分代码都可以在调整之后复用。那么，为什么我不整理出一个与业务无关的Redis队列工具出来呢，这样可以极大的提升代码的可复用性。后面有新的业务时，直接引入这个工具包，完善业务部分即可。

        以下是我在反思之后，对代码的整改（只包含有整改或新增的代码）

/*** 消息承载类** @author leixiyueqi* @since 2024/06/15 22:00*/
@Data
public class RedisQueueMsg<T> implements Serializable {/*** 消息Id*/private String id;/*** 服务名*/private String serverName;/*** 数据体*/private T data;
}package com.leixi.queue.pojo;import lombok.Data;import java.io.Serializable;
import java.util.Date;/*** 任务处理结果封装类** @author leixiyueqi* @since 2024/06/15 22:00*/
@Data
public class RedisResultVo implements Serializable {private String status;private Object data;private String desc;private Date startTime;private Date endTime;private String server;public RedisResultVo() {this.startTime = new Date();}public RedisResultVo(Object data) {this.data = data;this.startTime = new Date();}
}/*** 抽象的业务处理服务** @author leixiyueqi* @since 2024/06/15 22:00*/
public abstract class QueueBusiBasicService {/*** 处理任务的方法** @param obj 业务类*/public abstract void handle(Object obj);/*** 处理失败的回调方法** @param obj 业务类* @param e*/public abstract void callBack(Object obj, Exception e);
}/*** Redis队列的服务层** @author leixiyueqi* @since 2024/06/15 22:00*/
@Slf4j
@Service
public class QueueCommonService {@Autowiredprivate RedisTemplate<String, String> redisTemplate;@Autowiredprivate Map<String, QueueBusiBasicService> serviceMap;@Value("${leixi.redis-queue-key}")private String REDIS_QUEUE_KEY ;/*** 插入消息到队列** @param obj 业务对象* @param serverName  服务名* @return*/public RedisQueueMsg sendMessage(Object obj, String serverName) {RedisQueueMsg msg = new RedisQueueMsg();msg.setId(IdUtil.fastSimpleUUID());msg.setServerName(serverName);msg.setData(obj);redisTemplate.delete(JSON.toJSONString(msg));redisTemplate.opsForZSet().add(REDIS_QUEUE_KEY+ Constant.PENDING_SUFFIX, JSON.toJSONString(msg), Constant.getScoreByDate());redisTemplate.convertAndSend(REDIS_QUEUE_KEY+ Constant.WORKING_QUEUE_SUFFIX, JSON.toJSONString(msg));return msg;}/*** 处理队列中的工作** @param message*/public void handle(String message) {Boolean flag = redisTemplate.opsForValue().setIfAbsent(message, "1", 1, TimeUnit.DAYS);// 加锁失败，已有消费端在此时对此消息进行处理，这里不再做处理if (!flag) {return;}redisTemplate.opsForZSet().remove(REDIS_QUEUE_KEY+ Constant.PENDING_SUFFIX, message);RedisQueueMsg param = CastUtils.cast(JSON.parseObject(message, RedisQueueMsg.class));RedisResultVo result = new RedisResultVo(param);try {result.setServer(InetAddress.getLocalHost().getHostAddress());serviceMap.get(param.getServerName()).handle(param.getData());result.setStatus(Constant.SUCCESS);}catch (Exception e) {e.printStackTrace();serviceMap.get(param.getServerName()).callBack(param.getData(), e);result.setStatus(Constant.FAIL);result.setDesc(e.getMessage());}finally {result.setEndTime(new Date());redisTemplate.opsForZSet().add(REDIS_QUEUE_KEY+Constant.HANDLED_SUFFIX, JSON.toJSONString(result), Constant.getScoreByDate());}}/*** 查询待处理任务** @return*/public Object getPendingTask() {return redisTemplate.opsForZSet().range(REDIS_QUEUE_KEY+ Constant.PENDING_SUFFIX, 0, -1);}/*** 清理待处理任务** @return*/public Object cleanPendingTask() {Set<String> set = redisTemplate.opsForZSet().range(REDIS_QUEUE_KEY+ Constant.PENDING_SUFFIX, 0, -1);if(set.size() > 0) {redisTemplate.opsForZSet().remove(REDIS_QUEUE_KEY + Constant.PENDING_SUFFIX, set.toArray());}return "待处理任务已被清空！";}/*** 查询已处理任务** @return*/public Object getHandledTask() {return redisTemplate.opsForZSet().range(REDIS_QUEUE_KEY+Constant.HANDLED_SUFFIX, 0, -1);}/*** 清理某天前的处理任务** @param day 天数* @return*/public Object cleanHandledTask(Integer day) {Set<String> set = redisTemplate.opsForZSet().rangeByScore(REDIS_QUEUE_KEY+Constant.HANDLED_SUFFIX, 0, Constant.getScoreByDate() - day);set.forEach(s -> {RedisResultVo obj = JSONObject.parseObject(s, RedisResultVo.class);if (obj.getStatus().equals(Constant.SUCCESS)) {redisTemplate.opsForZSet().remove(REDIS_QUEUE_KEY+Constant.HANDLED_SUFFIX, s);}});return redisTemplate.opsForZSet().range(REDIS_QUEUE_KEY+Constant.HANDLED_SUFFIX, 0, System.currentTimeMillis());}/*** 重新处理已处理任务*/public String restartFailedTask() {Set<String> set = redisTemplate.opsForZSet().range(REDIS_QUEUE_KEY+Constant.HANDLED_SUFFIX, 0, -1);StringBuilder sb = new StringBuilder();set.forEach(s -> {RedisResultVo obj = JSONObject.parseObject(s, RedisResultVo.class);if (!obj.getStatus().equals(Constant.SUCCESS)) {redisTemplate.opsForZSet().remove(REDIS_QUEUE_KEY+Constant.HANDLED_SUFFIX, s);RedisQueueMsg msg = JSON.parseObject(JSON.toJSONString(obj.getData()), RedisQueueMsg.class); ;sendMessage(msg.getData(), msg.getServerName());sb.append(msg.getId()).append(",");}});return "以下任务被重启: "+ sb;}/*** 任务插队** @param msgId 要插队的消息ID*/public RedisQueueMsg cutInLineTask(String msgId) {Set<String> set = redisTemplate.opsForZSet().range(REDIS_QUEUE_KEY+ Constant.PENDING_SUFFIX, 0, -1);for (String s : set) {RedisQueueMsg msg = JSONObject.parseObject(s, RedisQueueMsg.class);if (msg.getId().equals(msgId)) {CompletableFuture.runAsync(() -> {this.handle(s);});return msg;}}throw new RuntimeException("未找到相关任务，该项任务已经在执行了！");}
}/*** 业务服务类，继承抽象类务类，实现业务逻辑** @author leixiyueqi* @since 2024/06/15 22:00*/
@Service(Constant.DINNER_SERVER)
public class QueueDinnerService extends QueueBusiBasicService {@Autowiredprivate QueueCommonService queueCommonService;/*** 年夜饭下单** @param entity 订单信息* @return*/public Object orderNewYearEveDinner(OrderEntity entity) {// 存储订单信息saveOrder(entity);queueCommonService.sendMessage(entity, Constant.DINNER_SERVER);// 异步开始做菜return "您已成功下单，订单号为"+ entity.getOrderCode()+"，后厨正在准备预制菜!";}/*** 这里模拟的是做年夜饭的过程方法，该方法用时较长，整个过程需要10秒，但是，这个过程中存在多种意外，该方法可能失败** @param req 订单信息*/private void doNewYearEveDinner(OrderEntity req) throws Exception {System.out.println("开始做订单 " + req.getOrderCode() + " 的年夜饭");Thread.sleep(10000);int i = new Random().nextInt(6) + 1;if (i ==4) {throw new Exception("厨师跑了");}if (i ==5) {throw new Exception("食物跑了");}if (i ==6) {throw new Exception("厨房着火了");}System.out.println("订单 " + req.getOrderCode() + " 的年夜饭已经完成");}/*** 保存订单信息** @param req*/private void saveOrder(OrderEntity req) {//这里假设做的是订单入库操作System.out.println("订单 " + req.getOrderCode() + " 已经入库， 做饭开始时间为 "+ new Date());}/*** 根据订单编号修改订单信息** @param orderCode 订单编号* @param dinnerStatus* @param remark*/private void updateOrder(String orderCode, String dinnerStatus, String remark) {// 根据订单编号修改订单的出餐结束时间，出餐状态，备注等信息。System.out.println("更新订单 "+ orderCode +" 信息，做饭结束时间为 "+ new Date() + ", 出餐状态为"+ dinnerStatus +", 备注为 " +remark);}/*** 处理订单** @param obj 业务类*/@Override@SneakyThrowspublic void handle(Object obj) {OrderEntity entity = JSON.parseObject(JSON.toJSONString(obj), OrderEntity.class);doNewYearEveDinner(entity);updateOrder(entity.getOrderCode(), Constant.SUCCESS, "出餐成功");}@Overridepublic void callBack(Object obj, Exception e) {OrderEntity entity = JSON.parseObject(JSON.toJSONString(obj), OrderEntity.class);System.out.println("更新订单 "+ entity.getOrderCode() +" 信息，做饭结束时间为 "+ new Date() + ", 出餐状态为FAIL, 原因为 " +e.getMessage());}
}@Component
public class RedisQueueListener implements MessageListener {@Autowiredprivate QueueCommonService service;private final Object lock = new Object();@Overridepublic void onMessage(Message message, byte[] pattern)  {synchronized (lock) {service.handle(message.toString());}}
}/*** 业务控制层** @author leixiyueqi* @since 2024/06/15 22:00*/
@RestController
@RequestMapping("/dinner")
public class DinnerController {@Autowiredprivate QueueDinnerService dinnerService;private int i = 0;@GetMapping("/orderDinner")public Object orderDinner() {OrderEntity entity = new OrderEntity();entity.setOrderCode("Order" + (++i));entity.setCustomerName("第"+i+"位客户");return dinnerService.orderNewYearEveDinner(entity);}}/*** Redis工具控制器** @author leixiyueqi* @since 2024/06/15 22:00*/
@RestController
@RequestMapping("/redisQueue")
public class RedisQueueController {@Autowiredprivate QueueCommonService service;@GetMapping("/getPendingTask")public Object getPendingTask() {return service.getPendingTask();}@GetMapping("/cleanPendingTask")public Object cleanPendingTask() {return service.cleanPendingTask();}@GetMapping("/getHandledTask")public Object getHandledTask() {return service.getHandledTask();}@GetMapping("/cleanOldSucceedTask")public Object cleanHandledTask(@RequestParam("day") Integer day) {return service.cleanHandledTask(day);}/*** 这里还有最后一个问题， 把已办里的错误的信息摘除出来，重新走请求。并且反馈哪些信息重新走了请求。*/@GetMapping("/restartFailedTask")public Object restartFailedTask() {service.restartFailedTask();return "重启失败的服务成功";}@GetMapping("/cutInLineTask")public Object cutInLineTask(@RequestParam("msgId")  String msgId) {RedisQueueMsg msg = service.cutInLineTask(msgId);return "任务 "+ JSON.toJSONString(msg) + "插队成功！";}}

        组件化的调整就是把属于Redis队列的操作与业务类操作完全分开，这样，以后有别的业务需要引入组件处理时，只需要写个业务服务继承QueueBusiBasicService即可，最大限度的复用了队列的这套机制和代码。

        注意，本组件有它特定的适用场景：处理任务的频度不高，每次处理任务用时较长，而且任务有一定的小概率失败，失败之后重新处理不会影响最终处理结果。

        完成这个工具研发后，我结合之前在网上查到的Redis队列的一些案例，发现用别的方案可以更简单的去实现我要的效果，比如直接用Redis队列详解（springboot实战）里的方案，仅仅是因为不想在代码里写 while(true) 这种不是优雅的代码，再加上用Listener的方式长时间没对消息进行消费时，消息会丢失，因此才额外花费了这么多的功夫来打补丁。果然方向选错了，工作量会成倍的增加，希望看到本文的读者能引以为戒，不要自误啊。