这篇是关于我使用Spring Cloud Steam操作RabbitMQ采用ttl+死信队列的方式实现的延迟队列。
前言
在公司项目中遇到了需要延迟队列的需求,为了以后可维护性和扩展性要求必须要用Springcloud Stream组件来操作mq,而且公司的rabbit也不允许安装延迟插件,只能用最原始的ttl+死信来实现,在搭建过程中遇到很多问题,最终成功实现,下面是代码,采用的是spring cloud steam3.1后的函数式编程实现。
先科普下原理:生产者发送消息到普通交换机绑定了个设置ttl时间的队列,这个队列绑定了个死信交换机且没人消费,如果消息过期就会发送到死信队列里,消费者就监听这个死信队列实现延迟队列的效果。
yaml配置文件
function中definition就是定义你的函数名称(后面发送和接收方法的名字)
bindings中的xxx-in/out-0,分为三个部分。第一个部分就是方法名,第二部分是输入或输出代表生产者out还是消费者in的意思,第三部分的0在rabbitmq里面是固定的,他是为了兼容kafkaf提供的,我们死写0就行。destination里面配置交换机的名字,后面的变量是环境的意思,到时候会变成dev、test、uat这些。group分组要设置一下我这里就叫模块的名字了。
重点:下面的producer:required-groups一定要配置,就是那个分组名字。
然后就是配置消费者,绑定的交换机要是最下面为生产者配置的死信交换机,分组名字也要记得填上不然不会消费。
后面的content-type是类型,可以不指定,默认就是这个
最下面是rabbit的配置,bingdings里面我配置了为生产者邦迪一个死信交换机,然后设置生产者多久没消费就会到死信的ttl。dead-letter-exchange是指定交换机名字,你就设置为原本交换机名称_DLX就好了为了规范,然后配置死信队列的名字,就是在交换机后面加上那个分组,一定要和上面的分组一致。
因为这里死信队列出来的消息是direct的,消费者是监听死信队列已经过期的消息的,所以交换机类型也要设置为direct,不然会报:direct消息转化topic异常,但是这个异常只会第一个消息报,后面的消息都不会了,可能是springcloud stream底层帮我们处理了,但为了避免第一次报错还是设置一下交换机类型。
cloud:
function:
definition: memberAccountUpdateTaskDelayed;handleTaskDelayed;
stream:
bindings:
memberAccountUpdateTaskDelayed-out-0:
destination: MEMBER_TOPIC_${spring.profiles.active} #延迟exchange,交换模式是topic
content-type: application/json #设置消息的类型为json
group: ${spring.application.name}
producer:
required-groups: ${spring.application.name} #必须要制定生产者分组,不然发不过去
handleTaskDelayed-in-0:
destination: MEMBER_TOPIC_DLX_${spring.profiles.active}
content-type: application/json
group: ${spring.application.name}
rabbit:
bindings:
memberAccountUpdateTaskDelayed-out-0:
producer:
ttl: 10000 #延时队列的延时时间,单位毫秒
auto-bind-dlq: true #开启死信队列
dead-letter-exchange: MEMBER_TOPIC_DLX_${spring.profiles.active} #死信交换机
dead-letter-queueName: MEMBER_TOPIC_DLX_${spring.profiles.active}.${spring.application.name} #死信队列名称
handleTaskDelayed-in-0:
consumer:
exchange-type: direct #死信交换机必须是direct类型的才能接受
生产者发送消息
这里我发送的是对象,在TaskDelayMessage就是发送的对象,实体类里面的属性上记得加上JSON反序列化的注解,不然消费者监听的时候会报反序列化错误。
方法名要和配置文件你配置的那个方法名一样
@Repository
@Slf4j
public class MemberAccountUpdateTaskHandleTimeoutTaskRepositoryImpl implements MemberAccountUpdateTaskHandleTimeoutTaskRepository {
private final Sinks.Many> sinks =
Sinks.many().multicast().onBackpressureBuffer();
@Bean
public Supplier>> memberAccountUpdateTaskDelayed(){
return sinks::asFlux;
}
@Override
public void sendDelayMessage(TaskDelayMessage message) {
log.info("生产者准备发送消息:{}", message+" -"+System.currentTimeMillis());
Message msg = MessageBuilder.withPayload(message).build();
while (sinks.tryEmitNext(msg).isFailure()) {
LockSupport.parkNanos(10);
}
log.info("生产者成功发送消息:{}", message+" -"+System.currentTimeMillis());
}
}
消费者监听消息
方法名要和配置文件你配置的那个方法名一样
@Bean
Consumer handleTaskDelayed() {
return message-> {
log.info("消费者监听到了消息:{}", message);
memberAccountUpdateTaskService.handleTimeOutTasks(message);
};
}
总结
总结:其实就是配置麻烦,国内springcloud steam的教程和博客是真的很少,搭建过程中遇到了很多问题csdn和b站也都没找到解决办法。最后看了油管和springcloud官方文档还有追进去看源码才解决。如果想学习的stream这个组件推荐去油管看教程,或者直接看springcloud官网文档。
服务器托管,北京服务器托管,服务器租用 http://www.fwqtg.net
相关推荐: 绝地求生 压枪python版
仅做学习交流,非盈利,侵联删(狗头保命)一、概述
二、详解
三、最麻烦的部分
四、最后仅做学习交流,非盈利,侵联删(狗头保命) 一、概述 1.1 效果 总的来说,这种方式是通过图像识别来完成的,不侵入游戏,不读取内存,安全不被检测。 1.2 前置知识 游戏中有各种不同的枪械,不同的枪械后坐力不一样,射速也不同。相同的枪械,装上不同的配件后,后坐…