使用rabbitmq 实现延迟消费

我会带着你远行 2022-06-07 00:29 336阅读 0赞

## 使用场景介绍 ##

1、发版时经常需要不停机发版，遇到mq消费者，消费一半停机就会出现消息丢失（没有使用手动确认的情况）

2、例如支付场景，准时支付、超过未支付将执行不同的方案，其中超时未支付可以看做一个延时消息。

3、例如滴滴、淘宝的自动评价都是类似场景（不一定是用的什么技术）

我是发版的情况遇到了

可能有人会问了，mq支持手动确认啊，为什么不使用确认机制呢？  
1、由于用的是Spring 的RabbitListener注解，无法使用手动确认机制  
2、也是最主要的原因是我们使用mq的场景

（类似于假如我做一个事情需要两步完成，每一步完成都会收钱。那么就会出现，第一步完成后，停机了，假如使用消息重发就会造成浪费第一步完成的钱）

# 实现方式 #

开始想到的是发版时，可以将消息发到队列里，然后消费者不要马上消费，等一定的时间再来消费这一消息。

在网上找资料找到两种实现方式：

### 1、使用插件 ###

在rabbitmq 3.5.7及以上的版本提供了一个插件（rabbitmq-delayed-message-exchange）来实现延迟队列，而我们公司mq是单独管理的，所以这种方式直接pass

### 2、使用两个特性 ###

AMQP和RabbitMQ本身没有直接支持延迟队列功能，但是可以通过以下特性模拟出延迟队列的功能。但是我们可以通过RabbitMQ的两个特性来曲线实现延迟队列。

#### Time To Live(TTL) ####

RabbitMQ可以针对Queue和Message设置 x-message-tt，来控制消息的生存时间，如果超时，则消息变为dead letter  
RabbitMQ针对队列中的消息过期时间有两种方法可以设置。

A: 通过队列属性设置，队列中所有消息都有相同的过期时间。
    B: 对消息进行单独设置，每条消息TTL可以不同。

如果同时使用，则消息的过期时间以两者之间TTL较小的那个数值为准。消息在队列的生存时间一旦超过设置的TTL值，就成为dead letter

#### Dead Letter Exchanges（DLX） ####

RabbitMQ的Queue可以配置x-dead-letter-exchange 和x-dead-letter-routing-key（可选）两个参数，如果队列内出现了dead letter，则按照这两个参数重新路由。  
  
x-dead-letter-exchange：出现dead letter之后将dead letter重新发送到指定exchange  
x-dead-letter-routing-key：指定routing-key发送  
队列出现dead letter的情况有：

消息或者队列的TTL过期
    队列达到最大长度
    消息被消费端拒绝（basic.reject or basic.nack）并且requeue=false

利用DLX，当消息在一个队列中变成死信后，它能被重新publish到另一个Exchange。这时候消息就可以重新被消费。

## 示例代码 ##

### 初始化过期队列的代码 ###

@Bean
        public Queue signQueueStore() {
            Map<String, Object> arguments = new HashMap<>();
            arguments.put("x-dead-letter-exchange", exchange);
            arguments.put("x-dead-letter-routing-key", signKey);
            arguments.put("x-message-ttl",300000);
            Queue queue = new Queue(signQueueStore,true,false,false,arguments);
            System.out.println("arguments :" + queue.getArguments());
            return queue;
        }

### 绑定过期队列的代码 ###

@Bean
        public Binding  signStoreBinding() {
            return BindingBuilder.bind(signQueueStore()).to(defaultExchange()).with(signKeyStore);
        }

### 初始化正常队列的代码和绑定普通队列的代码 ###

@Bean
        public Queue signQueue() {
            return new Queue(signQueue);
        }
    
        @Bean
        public Binding signBinding(Queue signQueue, DirectExchange defaultExchange) {
            /** 将队列绑定到交换机 */
            return BindingBuilder.bind(signQueue).to(defaultExchange).with(signKey);
        }

其他发送与消费不变

在重启时将消息发送到过期队列，在重启完成后，消息发送到正常队列，过期队列消息过期后会自动路由到正常队列进行消费。（可以设置一个标志，判断该标志位重启状态，则发到过期队列，为正常状态发送到正常队列）

# 参考文献 #

[rabbitmq 实现延迟队列的两种方式][rabbitmq]

[rabbitmq]: http://blog.csdn.net/u014308482/article/details/53036770