在日常工作中我们常常需要使用到分布式延时队列，本文浅谈目前笔者使用过的几种实现方式。
一、rabbitMQ（TTL+死信队列）
利用rabbitMQ的高级特性——TTL（Time To Live），对延时队列设置TTL，当消息过期后自动路由到对应的死信队列，消费者监控死信队列即可，如图。
@Configuration
public class RabbitMQConfig {
/**
* pay延时exchange
*/
public static final String PAY_DELAY_EXCHANGE_NAME = "delay.queue.pay.business.exchange";
/**
* pay死信Exchange
*/
private static final String PAY_DEAD_LETTER_EXCHANGE = "delay.queue.pay.deadletter.exchange";
/**
* pay 延时队列 1 2 3
*/
private static final String PAY_DELAY_QUEUE_ONE_NAME = "delay.queue.pay.business.queueOne";
private static final String PAY_DELAY_QUEUE_TWO_NAME = "delay.queue.pay.business.queueTwo";
private static final String PAY_DELAY_QUEUE_THREE_NAME = "delay.queue.pay.business.queueThree";
private static final String PAY_DELAY_QUEUE_FOUR_NAME = "delay.queue.pay.business.queueFour";
private static final String PAY_DELAY_QUEUE_FIVE_NAME = "delay.queue.pay.business.queueFive";
/**
* pay 死信队列
*/
public static final String PAY_DEAD_LETTER_QUEUE_NAME = "delay.queue.pay.deadletter.queue";
/**
* pay 死信队列 路由键
*/
private static final String PAY_DEAD_LETTER_QUEUE_ROUTING_KEY = "delay.queue.pay.deadletter.delay.routingkey";
/**
* pay延时队列 路由键
*/
public static final String PAY_DELAY_QUEUE_ONE_ROUTING_KEY = "delay.queue.pay.business.queueone.routingkey";
public static final String PAY_DELAY_QUEUE_TWO_ROUTING_KEY = "delay.queue.pay.business.queuetwo.routingkey";
public static final String PAY_DELAY_QUEUE_THREE_ROUTING_KEY = "delay.queue.pay.business.queuethree.routingkey";
public static final String PAY_DELAY_QUEUE_FOUR_ROUTING_KEY = "delay.queue.pay.business.queuefour.routingkey";
public static final String PAY_DELAY_QUEUE_FIVE_ROUTING_KEY = "delay.queue.pay.business.queuefive.routingkey";
/**
* pay 延时exchange到死信队列 路由键
*/
public static final String PAY_DELAY_QUEUE_DEAD_ROUTING_KEY = "delay.queue.pay.business.queuedead.routingkey";
/**
* 付款 声明延时Exchange
*
* @return
*/
@Bean("delayExchange")
public DirectExchange delayExchange() {
return new DirectExchange(PAY_DELAY_EXCHANGE_NAME);
}
/**
* 付款 声明死信Exchange
*
* @return
*/
@Bean("deadLetterExchange")
public DirectExchange deadLetterExchange() {
return new DirectExchange(PAY_DEAD_LETTER_EXCHANGE);
}
/**
* 声明延时队列A 延时1m
* 并绑定到对应的死信交换机
*
* @return
*/
@Bean("delayQueueOne")
public Queue delayQueueOne() {
Map<String, Object> args = new HashMap<>(2);
// x-dead-letter-exchange 这里声明当前队列绑定的死信交换机
args.put("x-dead-letter-exchange", PAY_DEAD_LETTER_EXCHANGE);
// x-dead-letter-routing-key 这里声明当前队列的死信路由key
args.put("x-dead-letter-routing-key", PAY_DEAD_LETTER_QUEUE_ROUTING_KEY);
// x-message-ttl 声明队列的TTL
args.put("x-message-ttl", 60000);
return QueueBuilder.durable(PAY_DELAY_QUEUE_ONE_NAME).withArguments(args).build();
}
/**
* 声明延时队列B 延时 2m
* 并绑定到对应的死信交换机
*
* @return
*/
@Bean("delayQueueTwo")
public Queue delayQueueTwo() {
Map<String, Object> args = new HashMap<>(2);
// x-dead-letter-exchange 这里声明当前队列绑定的死信交换机
args.put("x-dead-letter-exchange", PAY_DEAD_LETTER_EXCHANGE);
// x-dead-letter-routing-key 这里声明当前队列的死信路由key
args.put("x-dead-letter-routing-key", PAY_DEAD_LETTER_QUEUE_ROUTING_KEY);
// x-message-ttl 声明队列的TTL
args.put("x-message-ttl", 120000);
return QueueBuilder.durable(PAY_DELAY_QUEUE_TWO_NAME).withArguments(args).build();
}
/**
* 声明延时队列C 延时 3m
* 并绑定到对应的死信交换机
*
* @return
*/
@Bean("delayQueueThree")
public Queue delayQueueThree() {
Map<String, Object> args = new HashMap<>(2);
// x-dead-letter-exchange 这里声明当前队列绑定的死信交换机
args.put("x-dead-letter-exchange", PAY_DEAD_LETTER_EXCHANGE);
// x-dead-letter-routing-key 这里声明当前队列的死信路由key
args.put("x-dead-letter-routing-key", PAY_DEAD_LETTER_QUEUE_ROUTING_KEY);
// x-message-ttl 声明队列的TTL
args.put("x-message-ttl", 180000);
return QueueBuilder.durable(PAY_DELAY_QUEUE_THREE_NAME).withArguments(args).build();
}
/**
* 声明延时队列D 延时 4m
* 并绑定到对应的死信交换机
*
* @return
*/
@Bean("delayQueueFour")
public Queue delayQueueFour() {
Map<String, Object> args = new HashMap<>(2);
// x-dead-letter-exchange 这里声明当前队列绑定的死信交换机
args.put("x-dead-letter-exchange", PAY_DEAD_LETTER_EXCHANGE);
// x-dead-letter-routing-key 这里声明当前队列的死信路由key
args.put("x-dead-letter-routing-key", PAY_DEAD_LETTER_QUEUE_ROUTING_KEY);
// x-message-ttl 声明队列的TTL
args.put("x-message-ttl", 240000);
return QueueBuilder.durable(PAY_DELAY_QUEUE_FOUR_NAME).withArguments(args).build();
}
/**
* 声明延时队列F 延时 5m
* 并绑定到对应的死信交换机
*
* @return
*/
@Bean("delayQueueFive")
public Queue delayQueueFive() {
Map<String, Object> args = new HashMap<>(2);
// x-dead-letter-exchange 这里声明当前队列绑定的死信交换机
args.put("x-dead-letter-exchange", PAY_DEAD_LETTER_EXCHANGE);
// x-dead-letter-routing-key 这里声明当前队列的死信路由key
args.put("x-dead-letter-routing-key", PAY_DEAD_LETTER_QUEUE_ROUTING_KEY);
// x-message-ttl 声明队列的TTL
args.put("x-message-ttl", 300000);
return QueueBuilder.durable(PAY_DELAY_QUEUE_FIVE_NAME).withArguments(args).build();
}
/**
* 声明死信队列 用于接收延时处理的消息
*
* @return
*/
@Bean("deadLetterQueue")
public Queue deadLetterQueue() {
return new Queue(PAY_DEAD_LETTER_QUEUE_NAME);
}
/**
* 声明延时队列one绑定关系
*
* @param queue
* @param exchange
* @return
*/
@Bean
public Binding delayBindingOne(@Qualifier("delayQueueOne") Queue queue,
@Qualifier("delayExchange") DirectExchange exchange) {
return BindingBuilder.bind(queue).to(exchange).with(PAY_DELAY_QUEUE_ONE_ROUTING_KEY);
}
/**
* 声明业务队列two绑定关系
*
* @param queue
* @param exchange
* @return
*/
@Bean
public Binding delayBindingTwo(@Qualifier("delayQueueTwo") Queue queue,
@Qualifier("delayExchange") DirectExchange exchange) {
return BindingBuilder.bind(queue).to(exchange).with(PAY_DELAY_QUEUE_TWO_ROUTING_KEY);
}
/**
* 声明业务队列three绑定关系
*
* @param queue
* @param exchange
* @return
*/
@Bean
public Binding delayBindingThree(@Qualifier("delayQueueThree") Queue queue,
@Qualifier("delayExchange") DirectExchange exchange) {
return BindingBuilder.bind(queue).to(exchange).with(PAY_DELAY_QUEUE_THREE_ROUTING_KEY);
}
/**
* 声明业务队列four绑定关系
*
* @param queue
* @param exchange
* @return
*/
@Bean
public Binding delayBindingFour(@Qualifier("delayQueueFour") Queue queue,
@Qualifier("delayExchange") DirectExchange exchange) {
return BindingBuilder.bind(queue).to(exchange).with(PAY_DELAY_QUEUE_FOUR_ROUTING_KEY);
}
/**
* 声明业务队列five绑定关系
*
* @param queue
* @param exchange
* @return
*/
@Bean
public Binding delayBindingFive(@Qualifier("delayQueueFive") Queue queue,
@Qualifier("delayExchange") DirectExchange exchange) {
return BindingBuilder.bind(queue).to(exchange).with(PAY_DELAY_QUEUE_FIVE_ROUTING_KEY);
}
/**
* 延时exchange到死信队列
*
* @param queue
* @param exchange
* @return
*/
@Bean
public Binding delayBindingDead(@Qualifier("deadLetterQueue") Queue queue,
@Qualifier("delayExchange") DirectExchange exchange) {
return BindingBuilder.bind(queue).to(exchange).with(PAY_DELAY_QUEUE_DEAD_ROUTING_KEY);
}
/**
* 声明死信队列绑定关系
*
* @param queue
* @param exchange
* @return
*/
@Bean
public Binding deadLetterBindingQueue(@Qualifier("deadLetterQueue") Queue queue,
@Qualifier("deadLetterExchange") DirectExchange exchange) {
return BindingBuilder.bind(queue).to(exchange).with(PAY_DEAD_LETTER_QUEUE_ROUTING_KEY);
}
}
