RabbitMQDLX(DeadLetterExchanges)实现延迟队列

一、延迟队列设计

队列中的消息成为死信(Dead Letter)的几种情况&＃xff1a;

1) The message is rejected (basic.reject or basic.nack) with requeue&＃61;false,

2) The TTL for the message expires; or

3) The queue length limit is exceeded.

而通过配置队列的x-dead-letter-exchange及x-dead-letter-routing-key键值&＃xff0c;Dead Letter就会被重新发送到指定的DLX中。

 


如上图所示&＃xff0c;注意延迟队列并没有消费者&＃xff0c;并且设置参数&＃xff1a;

x-dead-letter-exchange:DLX

x-dead-letter-routing-key:DLK

x-max-length:1000000 // length &＃61; 1million

x-message-ttl:30000 // TTL &＃61; 30s

生产者发布的消息路由到该延迟队列&＃xff0c;30秒后消息成为死信并被重新发送到死信交换器DLX中&＃xff1b;消费者订阅死信队列DLQ&＃xff0c;消费消息。

二、
延迟队列实现

1. &＃64;Configuration
2. &＃64;PropertySource("classpath:rabbitmq-cfg.properties")
3. public class RabbitConfig {
4. /**
5. * 配置RabbitMQ连接工厂
6. *
7. * &＃64;param host
8. * &＃64;param port
9. * &＃64;param username
10. * &＃64;param password
11. * &＃64;param virtualHost
12. * &＃64;param connectionTimeout
13. * &＃64;return
14. */
15. &＃64;Bean
16. public ConnectionFactory connectionFactory(&＃64;Value("${spring.rabbitmq.host}") String host,
17. &＃64;Value("${spring.rabbitmq.port}") String port, &＃64;Value("${spring.rabbitmq.username}") String username,
18. &＃64;Value("${spring.rabbitmq.password}") String password,
19. &＃64;Value("${spring.rabbitmq.virtual-host}") String virtualHost,
20. &＃64;Value("${spring.rabbitmq.connection-timeout}") String connectionTimeout) {
21. CachingConnectionFactory ret &＃61; new CachingConnectionFactory();
22. ret.setHost(host);
23. ret.setPort(Integer.parseInt(port));
24. ret.setUsername(username);
25. ret.setPassword(password);
26. ret.setVirtualHost(virtualHost);
27. ret.setConnectionTimeout(Integer.parseInt(connectionTimeout));
28. return ret;
29. }
30. 
    
31. /**
32. * 声明30s延迟队列
33. *
34. * &＃64;param name
35. * &＃64;return
36. */
37. &＃64;Bean
38. public Queue queueDelay30s(&＃64;Value("${atf.queue.delay.30s}") String name) {
39. Map arguments &＃61; new HashMap();
40. arguments.put("x-message-ttl", 30 * 1000); // Message TTL &＃61; 30s
41. arguments.put("x-max-length", 1000000); // Max length &＃61; 1million
42. // 死信路由到死信交换器DLX
43. arguments.put("x-dead-letter-exchange", "atf.exchange.delay.dlx");
44. arguments.put("x-dead-letter-routing-key", "atf.routingkey.delay.dlk");
45. return new Queue(name, true, false, false, arguments);
46. }
47. 
    
48. /**
49. * 声明死信队列DLQ
50. *
51. * &＃64;param name
52. * &＃64;return
53. */
54. &＃64;Bean
55. public Queue queueDelayDlq(&＃64;Value("${atf.queue.delay.dlq}") String name) {
56. return new Queue(name, true, false, false);
57. }
58. 
    
59. /**
60. * 声明延迟交换器
61. *
62. * &＃64;param name
63. * &＃64;return
64. */
65. &＃64;Bean
66. public DirectExchange exchangeDelay(&＃64;Value("${atf.exchange.delay}") String name) {
67. DirectExchange ret &＃61; new DirectExchange(name, true, false);
68. ret.setInternal(false);
69. return ret;
70. }
71. 
    
72. /**
73. * 声明死信交换器DLX
74. *
75. * &＃64;param name
76. * &＃64;return
77. */
78. &＃64;Bean
79. public DirectExchange exchangeDelayDlx(&＃64;Value("${atf.exchange.delay.dlx}") String name) {
80. return new DirectExchange(name, true, false);
81. }
82. 
    
83. /**
84. * 绑定延迟交换器和30s延迟队列
85. *
86. * &＃64;param queue
87. * &＃64;param exchange
88. * &＃64;param routingKey
89. * &＃64;return
90. */
91. &＃64;Bean
92. public Binding bindingDelay(&＃64;Qualifier("queueDelay30s") Queue queue,
93. &＃64;Qualifier("exchangeDelay") DirectExchange exchange,
94. &＃64;Value("${atf.routingkey.delay.30s}") String routingKey) {
95. return BindingBuilder.bind(queue).to(exchange).with(routingKey);
96. }
97. 
    
98. /**
99. * 绑定死信交换器DLX和死信队列DLQ
100. *
101. * &＃64;param queue
102. * &＃64;param exchange
103. * &＃64;param routingKey
104. * &＃64;return
105. */
106. &＃64;Bean
107. public Binding bindingDelayConsume(&＃64;Qualifier("queueDelayDlq") Queue queue,
108. &＃64;Qualifier("exchangeDelayDlx") DirectExchange exchange,
109. &＃64;Value("${atf.routingkey.delay.dlk}") String routingKey) {
110. return BindingBuilder.bind(queue).to(exchange).with(routingKey);
111. }
112. }

1. public void sendDelay(AtfEventPayload payload) {
2. rabbitTemplate.convertAndSend("atf.exchange.delay", "atf.routingkey.delay.30s", payload);
3. }

1. &＃64;RabbitListener(queues &＃61; "atf.queue.delay.dlq", containerFactory &＃61; "rabbitListenerContainerFactory")
2. public void consumeDelay(AtfEventPayload payload) {
3. try {
4. AtWorkflow workflow &＃61; workflowSvs.find(payload.getWorkflowName());
5. workflowSvs.execute(workflow, null);
6. } catch (Exception e) {
7. e.printStackTrace();
8. }
9. }

像支付宝&＃xff0c;微信支付后的回调机制&＃xff0c;可以使用延迟队列实现。可是创建延时不同时间的队列&＃xff0c;死信后被消费&＃xff0c;消费异常后&＃xff0c;消息进入不同的队列。

/**
* 默认及时消息交换机
* &＃64;return
*/
&＃64;Bean()
public DirectExchange defaultDirectExchange() {return new DirectExchange(MessageQueueConstants.DEFAULT_DIRECT_EXCHANGE_NAME, true, false);
}/**
* 默认延迟消息死信队列
* &＃64;return
*/
&＃64;Bean
public Queue defaultDeadLetterQueue10() {Map, Object> arguments &＃61; new HashMap<>();
arguments.put("x-dead-letter-exchange",MessageQueueConstants.DEFAULT_DIRECT_EXCHANGE_NAME);//设置交换机路由
arguments.put("x-dead-letter-routing-key", MessageQueueConstants.DEFAULT_REPEAT_TRADE_QUEUE_NAME);//设置转发队列名称
arguments.put("x-message-ttl", 10 * 1000); // Message TTL &＃61; 10s
Queue queue &＃61; new Queue(MessageQueueConstants.DEFAULT_DEAD_LETTER_QUEUE_NAME10,true,false,false,arguments);
return queue;
}/**
* 默认延迟消息死信队列
* &＃64;return
*/
&＃64;Bean
public Queue defaultDeadLetterQueue30() {Map, Object> arguments &＃61; new HashMap<>();
arguments.put("x-dead-letter-exchange",MessageQueueConstants.DEFAULT_DIRECT_EXCHANGE_NAME);//设置交换机路由
arguments.put("x-dead-letter-routing-key", MessageQueueConstants.DEFAULT_REPEAT_TRADE_QUEUE_NAME);//设置转发队列名称
arguments.put("x-message-ttl", 30 * 1000); // Message TTL &＃61; 30s
Queue queue &＃61; new Queue(MessageQueueConstants.DEFAULT_DEAD_LETTER_QUEUE_NAME30,true,false,false,arguments);
return queue;
}/**
* 默认延迟消息死信队列
* &＃64;return
*/
&＃64;Bean
public Queue defaultDeadLetterQueue180() {Map, Object> arguments &＃61; new HashMap<>();
arguments.put("x-dead-letter-exchange",MessageQueueConstants.DEFAULT_DIRECT_EXCHANGE_NAME);//设置交换机路由
arguments.put("x-dead-letter-routing-key", MessageQueueConstants.DEFAULT_REPEAT_TRADE_QUEUE_NAME);//设置转发队列名称
arguments.put("x-message-ttl", 1000*60*3); // Message TTL &＃61; 180s
Queue queue &＃61; new Queue(MessageQueueConstants.DEFAULT_DEAD_LETTER_QUEUE_NAME180,true,false,false,arguments);
return queue;
}/**
* 默认延迟消息死信队列
* &＃64;return
*/
&＃64;Bean
public Queue defaultDeadLetterQueue1200() {Map, Object> arguments &＃61; new HashMap<>();
arguments.put("x-dead-letter-exchange",MessageQueueConstants.DEFAULT_DIRECT_EXCHANGE_NAME);//设置交换机路由
arguments.put("x-dead-letter-routing-key", MessageQueueConstants.DEFAULT_REPEAT_TRADE_QUEUE_NAME);//设置转发队列名称
arguments.put("x-message-ttl", 1000*60*20); // Message TTL &＃61; 1200s
Queue queue &＃61; new Queue(MessageQueueConstants.DEFAULT_DEAD_LETTER_QUEUE_NAME1200,true,false,false,arguments);
return queue;
}/**
* 默认延迟消息死信队列
* &＃64;return
*/
&＃64;Bean
public Queue defaultDeadLetterQueue3600() {Map, Object> arguments &＃61; new HashMap<>();
arguments.put("x-dead-letter-exchange",MessageQueueConstants.DEFAULT_DIRECT_EXCHANGE_NAME);//设置交换机路由
arguments.put("x-dead-letter-routing-key", MessageQueueConstants.DEFAULT_REPEAT_TRADE_QUEUE_NAME);//设置转发队列名称
arguments.put("x-message-ttl", 1000*60*60); // Message TTL &＃61; 3600s
Queue queue &＃61; new Queue(MessageQueueConstants.DEFAULT_DEAD_LETTER_QUEUE_NAME3600,true,false,false,arguments);
return queue;
}&＃64;Bean
public Binding defaultDeadLetterBinding10() {Binding bind &＃61; BindingBuilder.bind(defaultDeadLetterQueue10()).to(defaultDirectExchange()).with(MessageQueueConstants.DEFAULT_DEAD_LETTER_QUEUE_NAME10);
return bind;
}
&＃64;Bean
public Binding defaultDeadLetterBinding30() {Binding bind &＃61; BindingBuilder.bind(defaultDeadLetterQueue30()).to(defaultDirectExchange()).with(MessageQueueConstants.DEFAULT_DEAD_LETTER_QUEUE_NAME30);
return bind;
}
&＃64;Bean
public Binding defaultDeadLetterBinding180() {Binding bind &＃61; BindingBuilder.bind(defaultDeadLetterQueue180()).to(defaultDirectExchange()).with(MessageQueueConstants.DEFAULT_DEAD_LETTER_QUEUE_NAME180);
return bind;
}
&＃64;Bean
public Binding defaultDeadLetterBinding1200() {Binding bind &＃61; BindingBuilder.bind(defaultDeadLetterQueue1200()).to(defaultDirectExchange()).with(MessageQueueConstants.DEFAULT_DEAD_LETTER_QUEUE_NAME1200);
return bind;
}
&＃64;Bean
public Binding defaultDeadLetterBinding3600() {Binding bind &＃61; BindingBuilder.bind(defaultDeadLetterQueue3600()).to(defaultDirectExchange()).with(MessageQueueConstants.DEFAULT_DEAD_LETTER_QUEUE_NAME3600);
return bind;
}/**
* 默认延迟消息死信接受转发消息队列
* &＃64;return
*/
&＃64;Bean
public Queue defaultRepeatTradeQueue() {Queue queue &＃61; new Queue(MessageQueueConstants.DEFAULT_REPEAT_TRADE_QUEUE_NAME,true,false,false);
return queue;
}&＃64;Bean
public Binding defaultRepeatTradeBinding() {return BindingBuilder.bind(defaultRepeatTradeQueue()).to(defaultDirectExchange()).with(MessageQueueConstants.DEFAULT_REPEAT_TRADE_QUEUE_NAME);
}