文章目录
- Springboot项目中快速引入Rabbit MQ通用做法
- 由来
- 引入MQ目的
- 步骤
- 总结
- 1.引入MQ的目的
- 2.配置要点
- 3.实现过程
- 4.扩展功能
Springboot项目中快速引入Rabbit MQ通用做法
由来
本篇是在做BI项目时尝试引入MQ来优化项目时发现,MQ引入到项目中做法比较类似。变的只是谁发消息给谁,谁去监听消息。至于MQ的可靠性(比如生产者可靠性、消费者可靠性、消息可靠性)都是通过固定参数进行配置。
因此将引入MQ到BI项目的过程抽象出来,变成通用的方法~
引入MQ目的
砍掉耗时久的业务,缩短单个业务时间,监听者异步执行耗时久任务
做完之后,你可以完成springboot项目中MQ基本配置,并且MQ具有一定可靠性~
步骤
- 在Application中配置MQ
- 生产者确认机制,重连机制、消费者能者多劳、确认机制、失败消息处理策略
- 保证可靠性(配备了生产者确认,重连、消费者能者多劳,确认机制,失败消息处理策略)
yml配置
失败者消息处理策略实现 - 消息发到error交换机spring:rabbitmq:# 连接信息host: xxx # 你的IP地址port: 5672# 用户相关信息建议在rabbitMQ中设置好virtual-host: /bi # 可在UI界面创建独属项目的虚拟机与用户名username: xxx # 用户名password: xxx # 密码connection-timeout: 200ms # max waited time# 生产者(消息发送者)# 生产者确认机制 - 默认取消,消耗性能publisher-confirm-type: nonepublisher-returns: falsetemplate:# 生产者重连机制retry:enabled: trueinitial-interval: 1000msmultiplier: 1max-attempts: 3# 消费者(监听者)listener:simple:prefetch: 1 # (能者多劳)每次只能获取一条信息,处理完才能获取下一条acknowledge-mode: auto # 消费者确认 - 自动模式retry:enabled: true # 失败消息处理策略
/*** 失败者消息处理策略实现*/@Beanpublic MessageRecoverer messageRecoverer(RabbitTemplate rabbitTemplate){return new RepublishMessageRecoverer(rabbitTemplate, BI_ERROR_EXCHANGE, BI_ERROR_ROUTING_KEY);}
- 统一确定并定义业务的有关MQ常量 MQConstant.java
- 交换机、队列、Key
- error交换机,队列,Key(按需设置)
- 开始初始化
a. 业务MQ 与 error交换机MQ,并绑定关系(选一个就行,推荐第二种)
用Bean方法创建
直接在@RabbitListener注解中定义并绑定@Configuration public class ErrorConfiguration {@Beanpublic Queue errorQueue() {return QueueBuilder.durable(BI_ERROR_QUEUE).build();}@Beanpublic DirectExchange errorExchange() {return ExchangeBuilder.directExchange(BI_ERROR_EXCHANGE).build();}@Beanpublic Binding errorBinding() {return BindingBuilder.bind(errorQueue()).to(errorExchange()).with(BI_ERROR_ROUTING_KEY);}}
b. JSON消息转换器(替换掉原生的JDK)@RabbitListener(bindings = @QueueBinding(// 队列:// name - 队列名字// durable - 队列持久化,不会随着MQ关闭而消失// arguments:使队列为Lazy queue将消息尽快写入磁盘value = @Queue(name = BI_QUEUE_NAME,durable = "true",arguments = @Argument(name = "x-queue-mode", value = "lazy")),// 交换机:指定交换机的名字与类型(默认direct)exchange = @Exchange(name = BI_EXCHANGE_NAME, type = ExchangeTypes.DIRECT),// 按交换机类型(Direct、Topic),设置Keykey = BI_ROUTING_KEY )) public void receiveMessage(String msg) {
/*** 消息转换器* @return*/@Beanpublic MessageConverter messageConverter(){return new Jackson2JsonMessageConverter();}
- 编写业务代码(根据实际业务)
a. 发送信息:砍掉耗时久的业务,变成发送消息
ⅰ. 选用唯一性的信息,如id
ⅱ. 确定好消息的数据类型
b. 监听信息:添加监听者,执行耗时久的业务。
ⅰ. 需要根据实际情况修改代码
ⅱ. 可以根据业务实际情况使:业务幂等性
当然,还可以对其进行拓展,比如对error队列进行监听,针对错误消息进行特殊业务处理等等~@RabbitListener(bindings = @QueueBinding(// 队列:// name - 队列名字// durable - 队列持久化,不会随着MQ关闭而消失// arguments:使队列为Lazy queue将消息尽快写入磁盘value = @Queue(name = BI_QUEUE_NAME,durable = "true",arguments = @Argument(name = "x-queue-mode", value ="lazy")),// 交换机:指定交换机的名字与类型(默认direct)exchange = @Exchange(name = BI_EXCHANGE_NAME, type = ExchangeTypes.DIRECT),// 按交换机类型(Direct、Topic),设置Keykey = BI_ROUTING_KEY )) public void receiveMessage(Long chatId) {// 0. 业务幂等性判断 - 基于乐观锁改造boolean update = chartService.lambdaUpdate().set(Chart::getStatus, RUNNING_STATUS).eq(Chart::getId, chatId).eq(Chart::getStatus, WAIT_STATUS).update();if (!update) {handleChartUpdateError(chatId, "该图表正在生成中!请耐心等待");return;}
至此之后,MQ基本操作以及配置完毕~并且MQ可靠性相对高。 - 编写业务代码(根据实际业务)
总结
在Spring Boot项目中快速集成RabbitMQ以增强系统性能和可靠性,主要涉及以下几个关键步骤:
1.引入MQ的目的
- 业务优化:通过异步处理长时间运行的任务,减少单个业务请求的响应时间。
- 可靠性提升:确保消息传递的稳定性和准确性,即使在网络不稳定或系统故障的情况下。
2.配置要点
- 连接信息:在
application.yml
中配置RabbitMQ服务器的连接细节,包括主机、端口、虚拟主机、用户名、密码及连接超时时间。 - 生产者确认机制:启用确认机制,配置重试策略,确保消息成功发布。
- 消费者策略:实施“能者多劳”原则,即每次仅处理一条消息,以及自动确认和失败消息重试策略。
- 失败消息处理:定义失败消息的重发逻辑,例如重定向至错误交换机。
3.实现过程
- 配置MQ:在
application.yml
中设置RabbitMQ相关参数,包括连接信息、生产者和消费者的配置。 - 定义MQ常量:统一定义交换机、队列和路由键等MQ相关常量,便于维护和扩展。
- 初始化MQ组件:
- 创建错误队列、交换机和绑定关系。
- 使用Bean方式或直接在
@RabbitListener
注解中定义队列、交换机和绑定关系。 - 替换默认的消息转换器为JSON消息转换器。
- 编写业务代码:
- 发送端:使用唯一标识符发送消息,并确保数据类型的正确性。
- 接收端:监听队列,处理耗时任务,实现业务幂等性确保数据一致性。
4.扩展功能
- 监听错误队列,对错误消息进行特殊处理,进一步完善系统健壮性。
通过上述步骤,可以实现在Spring Boot项目中高效、可靠地使用RabbitMQ作为消息中间件,优化业务流程并提升系统整体性能。