背景
年初的时候设计实践过一个课题：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: 19200
servlet:
context-path: /leixi
max-http-header-size: 102400
spring:
redis:
database: 9
host: 127.0.0.1
port: 6379
password:
jedis:
pool:
max-active: 8
max-wait: -1
max-idle: 8
min-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 ;
@Bean
public ChannelTopic topic() {
return new ChannelTopic(REDIS_QUEUE_KEY+ Constant.WORKING_QUEUE_SUFFIX);
}
@Bean
public MessageListenerAdapter messageListenerAdapter(DinnerListener listener) {
return new MessageListenerAdapter(listener);
}
@Bean
public 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;
@Autowired
private 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);
}
@Autowired
private 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 {
@Autowired
private DinnerService service;
private final Object lock = new Object();
@Override
public 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 {
@Autowired
private 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 {
@Autowired
private RedisTemplate<String, String> redisTemplate;
@Autowired
private 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 {
@Autowired
private 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
@SneakyThrows
public void handle(Object obj) {
OrderEntity entity = JSON.parseObject(JSON.toJSONString(obj), OrderEntity.class);
doNewYearEveDinner(entity);
updateOrder(entity.getOrderCode(), Constant.SUCCESS, "出餐成功");
}
@Override
public 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 {
@Autowired
private QueueCommonService service;
private final Object lock = new Object();
@Override
public 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 {
@Autowired
private 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 {
@Autowired
private 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的方式长时间没对消息进行消费时，消息会丢失，因此才额外花费了这么多的功夫来打补丁。果然方向选错了，工作量会成倍的增加，希望看到本文的读者能引以为戒，不要自误啊。
