SpringCloudStream集成rabbitMQ（延时队列）

上节咱们讲了SpringCloud Stream集成rabbitMQ&＃xff0c;本章节咱们将实现延时队列功能。在实际开发中我们有些场景用延时队列实现非常方便。下面列举延时队列适合使用的场景&＃xff1a;

1. 用户下单30分钟后未付款自动关闭订单
2. 用户下单外卖以后&＃xff0c;距离超时时间还有 10 分钟时提醒外卖小哥即将超时
3. 邮箱编辑好邮件定时发送等等

上节已经讲过SpringCloud Stream集成rabbitMQ步骤&＃xff0c;下面就不在重复赘述了&＃xff0c;不明白的同学可以去看上节的内容。

1.主要的变化是配置文件

spring:application:name: rabbitmq-domecloud:function:definition: source;sink; #函数名称&＃xff0c;对应服务中的注入的Beanstream:binders: #需要绑定的rabbitmq的服务信息default-binder: #定义的名称&＃xff0c;用于bidding整合type: rabbit #消息组件类型environment: #配置rabbimq连接环境spring:rabbitmq:addresses: 10.0.1.141:5672 #服务器的地址和端口username: xt3dev #用户名password: 4V9prcFbRoYxLCMd #密码bindings:source-out-0: #自定义消息通道的名称destination: QUEUE_DOME #exchange名称,交换模式默认是topic&＃xff0c;创建时同时会创建QUEUE_DOME.${spring.application.name}队列content-type: application/json #设置消息的类型为jsongroup: ${spring.application.name} #分组binder: default-binder #绑定的binder名称sink-in-0:destination: QUEUE_DOMEcontent-type: application/jsongroup: ${spring.application.name}binder: default-binderrabbit:bindings:source-out-0:producer:ttl: 5000 #延时队列的延时时间&＃xff0c;单位毫秒auto-bind-dlq: true #为true是开启死信队列dead-letter-exchange: QUEUE_DOME_IN #死信队列的交换机dead-letter-queueName: QUEUE_DOME_IN.${spring.application.name} #死信队列名称

2.启动项目测试

浏览器访问http://localhost:8080/sendMessage?message&＃61;hello rabbitMQ&＃xff0c;从控制台我看可以看出生产者发送消息到消费者接收到消息是有5秒钟的延时&＃xff0c;我们配置文件配置的也是5000毫秒&＃xff0c;说明延时队列功能是可以的。

3.延时队列原理分析

RabbitMQ本身没有直接支持延迟队列功能&＃xff0c;两种方式可以实现延迟队列&＃xff0c;一种是利用对的TTL特性来实现&＃xff0c;另外一种是使用RabbitMQ延迟插件来实现。本文主要讲述的是TTL的方式来实现的

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

• A: 通过队列属性设置&＃xff0c;队列中所有消息都有相同的过期时间。
• B: 对消息进行单独设置&＃xff0c;每条消息TTL可以不同。
  如果同时使用&＃xff0c;则消息的过期时间以两者之间TTL较小的那个数值为准。消息在队列的生存时间一旦超过设置的TTL值&＃xff0c;就成为dead letter

延迟队列原理分析图

 4.设置单条消息过期时间

使用spring cloud stream的方式是没办法设置单条消息过期时间的&＃xff0c;只能利用RabbitTemplate来设置单条消息过期时间。

4.1 创建RabbitConfig类

&＃64;Configuration
&＃64;RequiredArgsConstructor
public class RabbitConfig {private final BinderFactory binderFactory;private final static String binderName &＃61; "default-binder";&＃64;Beanpublic RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) throws NoSuchFieldException, IllegalAccessException {// 获取目标binderBinder binder &＃61;binderFactory.getBinder(binderName, BinderConfiguration.class);Assert.notNull(binder, binderName &＃43; " is null");// 获取binder的connectionFactoryField field &＃61; binder.getClass().getDeclaredField("connectionFactory");field.setAccessible(true);connectionFactory &＃61; (ConnectionFactory) field.get(binder);// newreturn new RabbitTemplate(connectionFactory);}
}

4.2 改造生产者实现类

&＃64;Slf4j
&＃64;Service
&＃64;RequiredArgsConstructor
public class ProducerServiceImpl implements ProducerService {private final RabbitTemplate rabbitTemplate;private final Sinks.Many> sinks &＃61;Sinks.many().multicast().onBackpressureBuffer();&＃64;Beanpublic Supplier>> source(){return sinks::asFlux;}&＃64;Overridepublic void sendMessage(String message, String expiration) {log.info("发送时间:{}, 生产者发送消息:{}", new Date(), message);rabbitTemplate.convertAndSend("QUEUE_DOME_OUT", "#", message, msg -> {MessageProperties messageProperties &＃61; msg.getMessageProperties();// 设置过期时间&＃xff0c;单位&＃xff1a;毫秒messageProperties.setExpiration(expiration);return msg;});}
}


4.3 改造MessageController类

&＃64;RestController
&＃64;RequiredArgsConstructor
public class MessageController {private final ProducerService producerService;&＃64;GetMapping("/sendMessage")public void sendMessage(&＃64;RequestParam(value &＃61; "message", defaultValue &＃61; "hello world") String message,&＃64;RequestParam(value &＃61; "expiration", defaultValue &＃61; "5000") String expiration) {producerService.sendMessage(message, expiration);}
}

4.4 启动项目测试

浏览器访问http://localhost:8080/sendMessage?message&＃61;hello%20rabbitMQ&expiration&＃61;3000&＃xff0c;设置消息过期时间为3秒

 结果&＃xff1a;

浏览器访问http://localhost:8080/sendMessage?message&＃61;hello%20rabbitMQ&expiration&＃61;3000&＃xff0c;设置消息过期时间为5秒 

结果&＃xff1a;