使用redis做消息队列实现任务调度

旧城等待， 2022-12-23 15:22 143阅读 0赞

总括：

1，消息本身(使用的时候去掉消息本身的过期时间)  
2，队列和消息的关系（消息id）  
3，消费异常重新进入队列  
4，消费用进程锁，和过期时间  
5，结合Spring的定时任务 @Scheduled(cron = "\*/1 \* \* \* \* \*") 默认是串行化，此任务执行完才开启下一次定时任务，否则  
定时时间到了也不会触发定时任务  
这几点结合就实现了kafka队列的调度（顺序消费，避免高并发内存溢出）

示例：

**A:需求说明:**

1.  如果系统中需要用到定时执行计划的，又不想用到中间件，如果轮询数据库的话，会导致大量资源消耗，这样我们就可以使用Redis来实现类似功（需要使用rabbitMQ的请看这里：[https://blog.csdn.net/u010096717/article/details/82148681][https_blog.csdn.net_u010096717_article_details_82148681]）
2.  业务类型，如订单一些评论，如果48h用户未对商家评论，系统会自动产生一条默认评论，还有排队到时提醒等

**B:实现思路：**

1.  将整个Redis当做消息池，以kv形式存储消息，key为id，value为具体的消息body
2.  使用ZSET做优先队列，按照score维持优先级（用当前时间+需要延时的时间作为score）
3.  轮询ZSET，拿出score比当前时间戳大的数据（已过期的）
4.  根据id拿到消息池的具体消息进行消费
5.  消费成功，删除改队列和消息
6.  消费失败，让该消息重新回到队列

**C:代码实现**

1.      Message消息封装类
    
         
    
    1.  `@Data`
    2.  `public class Message {`
    3.   
    4.  `/**`
    5.  `* 消息id`
    6.  `*/`
    7.  `private String id;`
    8.  `/**`
    9.  `* 消息延迟/毫秒`
    10. `*/`
    11. `private long delay;`
    12.  
    13. `/**`
    14. `* 消息存活时间`
    15. `*/`
    16. `private int ttl;`
    17. `/**`
    18. `* 消息体，对应业务内容`
    19. `*/`
    20. `private String body;`
    21. `/**`
    22. `* 创建时间，如果只有优先级没有延迟，可以设置创建时间为0`
    23. `* 用来消除时间的影响`
    24. `*/`
    25. `private long createTime;`
    26.  
    27. `}`

2.基于redis的消息队列

1.  `@Component`
2.  `public class RedisMQ {`
3.   
4.  `/**`
5.  `* 消息池前缀，以此前缀加上传递的消息id作为key，以消息{@link Message}`
6.  `* 的消息体body作为值存储`
7.  `*/`
8.  `public static final String MSG_POOL = "Message:Pool:";`
9.  `/**`
10. `* zset队列 名称 queue`
11. `*/`
12. `public static final String QUEUE_NAME = "Message:Queue:";`
13.  
14. `private static final int SEMIH = 30*60;`
15.  
16.  
17.  
18. `@Autowired`
19. `private RedisService redisService;`
20.  
21. `/**`
22. `* 存入消息池`
23. `* @param message`
24. `* @return`
25. `*/`
26. `public boolean addMsgPool(Message message) {`
27.  
28. `if (null != message) {`
29. `return redisService.setExp(MSG_POOL + message.getId(), message.getBody(), Long.valueOf(message.getTtl() + SEMIH));`
30. `}`
31. `return false;`
32. `}`
33.  
34. `/**`
35. `* 从消息池中删除消息`
36. `* @param id`
37. `* @return`
38. `*/`
39. `public void deMsgPool(String id) {`
40. `redisService.remove(MSG_POOL + id);`
41. `}`
42.  
43. `/**`
44. `* 向队列中添加消息`
45. `* @param key`
46. `* @param score 优先级`
47. `* @param val`
48. `* @return 返回消息id`
49. `*/`
50. `public void enMessage(String key, long score, String val) {`
51. `redisService.zsset(key,val,score);`
52. `}`
53.  
54. `/**`
55. `* 从队列删除消息`
56. `* @param id`
57. `* @return`
58. `*/`
59. `public boolean deMessage(String key, String id) {`
60. `return redisService.zdel(key, id);`
61. `}`
62.  
63. `}`

3Redis操作工具类,这个工具类比较多方法，就不贴在这里了([https://blog.csdn.net/u010096717/article/details/83783865][https_blog.csdn.net_u010096717_article_details_83783865])

4.编写消息发送（生产者）

1.  `@Component`
2.  `public class MessageProvider {`
3.   
4.  `static Logger logger = LoggerFactory.getLogger(MessageProvider.class);`
5.   
6.   
7.  `private static int delay = 30;//30秒，可自己动态传入`
8.   
9.  `@Resource`
10. `private RedisMQ redisMQ;`
11.  
12. `SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");`
13.  
14. `//改造成redis`
15. `public void sendMessage(String messageContent) {`
16. `try {`
17. `if (messageContent != null){`
18. `String seqId = UUID.randomUUID().toString();`
19. `// 将有效信息放入消息队列和消息池中`
20. `Message message = new Message();`
21. `// 可以添加延迟配置`
22. `message.setDelay(delay*1000);`
23. `message.setCreateTime(System.currentTimeMillis());`
24. `message.setBody(messageContent);`
25. `message.setId(seqId);`
26. `// 设置消息池ttl，防止长期占用`
27. `message.setTtl(delay + 360);`
28. `redisMQ.addMsgPool(message);`
29. `//当前时间加上延时的时间，作为score`
30. `Long delayTime = message.getCreateTime() + message.getDelay();`
31. `String d = sdf.format(message.getCreateTime());`
32. `System.out.println("当前时间：" + d+",消费的时间：" + sdf.format(delayTime));`
33. `redisMQ.enMessage(RedisMQ.QUEUE_NAME,delayTime, message.getId());`
34. `}else {`
35. `logger.warn("消息内容为空！！！！！");`
36. `}`
37. `}catch (Exception e){`
38. `e.printStackTrace();`
39. `}`
40. `}`
41. `}`

5.消息消费者

1.  `@Component`
2.  `public class RedisMQConsumer {`
3.   
4.  `@Resource`
5.  `private RedisMQ redisMQ;`
6.   
7.  `@Autowired`
8.  `private RedisService redisService;`
9.   
10. `@Autowired`
11. `private MessageProvider provider;`
12.  
13. `SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");`
14.  
15.  
16. `/**`
17. `* 消息队列监听器<br>`
18. `*`
19. `*/`
20. `@Scheduled(cron = "*/1 * * * * *")`
21. `public void monitor() {`
22. `Set<String> set = redisService.rangeByScore(RedisMQ.QUEUE_NAME, 0, System.currentTimeMillis());`
23. `if (null != set) {`
24. `long current = System.currentTimeMillis();`
25. `for (String id : set) {`
26. `long score = redisService.getScore(RedisMQ.QUEUE_NAME, id).longValue();`
27. `if (current >= score) {`
28. `// 已超时的消息拿出来消费`
29. `String str = "";`
30. `try {`
31. `str = redisService.get(RedisMQ.MSG_POOL + id);`
32. `System.out.println("消费了:" + str+ ",消费的时间：" + sdf.format(System.currentTimeMillis()));`
33. `} catch (Exception e) {`
34. `e.printStackTrace();`
35. `//如果出了异常，则重新放回队列`
36. `System.out.println("消费异常，重新回到队列");`
37. `provider.sendMessage(str);`
38. `} finally {`
39. `redisMQ.deMessage(RedisMQ.QUEUE_NAME, id);`
40. `redisMQ.deMsgPool(id);`
41. `}`
42. `}`
43. `}`
44. `}`
45. `}`
46. `}`

6.配置信息

1.  ``
2.  `<dependency>`
3.  `<groupId>org.springframework.boot</groupId>`
4.  `<artifactId>spring-boot-starter-data-redis</artifactId>`
5.  `</dependency>`
6.   
7.   
8.  `2yml配置`
9.  `spring:`
10. `redis:`
11. `database: 1`
12. `host: 127.0.0.1`
13. `port: 6379`

以上代码已经实现了延迟消费功能，现在来测试一下，调用MessageProvider的sendMessage方法，我设定了30秒

可以看到结果

![20181106114754730.png][]

因为我们是用定时器去轮询的，会出现误差

参看：

附件

[https://blog.csdn.net/u010096717/article/details/83783865][https_blog.csdn.net_u010096717_article_details_83783865]

[https://blog.csdn.net/u010096717/article/details/83783039][https_blog.csdn.net_u010096717_article_details_83783039]

[https_blog.csdn.net_u010096717_article_details_82148681]: https://blog.csdn.net/u010096717/article/details/82148681
[https_blog.csdn.net_u010096717_article_details_83783865]: https://blog.csdn.net/u010096717/article/details/83783865
[20181106114754730.png]: /images/20221120/306b521fd164410997b7dbd0e6e921a3.png
[https_blog.csdn.net_u010096717_article_details_83783039]: https://blog.csdn.net/u010096717/article/details/83783039