springboot~kafka中延时消息的实现

应用场景

• 用户下单5分钟后,给他发短信
  
• 用户下单30分钟后，如果用户不付款就自动取消订单
  

kafka无死信队列

kafka本身没有这种延时队列的机制，像rabbitmq有自己的死信队列，当一些消息在一定时间不消费时会发到死信队列，由死信队列来处理它们，上面的两个需求如果是rabbitmq可以通过死信队列实现的。
kafka有生产者拦截器

通过对生产者拦截器实现一个TTL的检查，然后再通过类似netty里的延时队列组件来实现消息的延时发送，发到咱们的死信队列里

• ProducerInterceptorTTL源码
  

1. public class ProducerInterceptorTTL implements ProducerInterceptor<Integer, String>, ApplicationContextAware {
3.         // 消息延时,单位秒
4.         public static String TTL = "ttl";
6.         // 死信队列,延时后发送到的队列，我们称为死信队列
7.         public static String DEAD_TOPIC = "dead_topic";
9.         // 静态化的上下文，用于获取bean，因为ConsumerInterceptor是通过反射创建的，所以无法通过注入的方式获取bean
10.         private static ApplicationContext applicationContext;
12.         // 时间轮，用于延时发送消息
13.         private static LindTimeWheel timeWheel = new LindTimeWheel(1000, 8);
15.         @Override
16.         public ProducerRecord onSend(ProducerRecord<Integer, String> record) {
17.                 final String topic = record.topic();
18.                 final Integer partition = record.partition();
19.                 final Integer key = record.key();
20.                 final String value = record.value();
21.                 final Long timestamp = record.timestamp();
22.                 final Headers headers = record.headers();
23.                 long ttl = -1;
24.                 String deadTopic = null;
25.                 for (Header header : headers) {
26.                         if (header.key().equals(TTL)) {
27.                                 ttl = toLong(header.value());
28.                         }
29.                         if (header.key().equals(DEAD_TOPIC)) {
30.                                 deadTopic = new String(header.value());
31.                         }
32.                 }
33.                 // 消息超时判定
34.                 if (deadTopic != null && ttl > 0) {
35.                         // 可以放在死信队列中
36.                         String finalDeadTopic = deadTopic;
37.                         long finalTtl = ttl * 1000;
38.                         timeWheel.addTask(() -> {
39.                                 System.out.println("消息超时了," + finalTtl + "需要发到topic:" + record.key());
40.                                 KafkaTemplate kafkaTemplate = applicationContext.getBean(KafkaTemplate.class);
41.                                 kafkaTemplate.send(finalDeadTopic, record.value());
43.                         }, finalTtl);
44.                 }
45.                 // 拦截器拦下来之后改变原来的消息内容
46.                 ProducerRecord<Integer, String> newRecord = new ProducerRecord<Integer, String>(topic, partition, timestamp,
47.                                 key, value, headers);
48.                 // 传递新的消息
49.                 return newRecord;
51.         }
53.         @Override
54.         public void onAcknowledgement(RecordMetadata recordMetadata, Exception e) {
56.         }
58.         @Override
59.         public void close() {
61.         }
63.         @Override
64.         public void configure(Map<String, ?> map) {
66.         }
68.         @Override
69.         public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
70.                 this.applicationContext = applicationContext;
71.         }
73. }

复制代码

• 注册拦截器
  

1. spring:
2.   kafka:
3.     producer:
4.       properties:
5.         interceptor.classes: com.ruoyi.lawyer.delay.ProducerInterceptorTTL

复制代码

• 延时消息在某个时间段之后会送出
  
  

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！