126 lines
4.0 KiB
Markdown
126 lines
4.0 KiB
Markdown
|
# sl-express-mq使用手册
|
|||
|
|
|
|||
|
|
## 1、说明
|
|||
|
|
|
|||
|
|
为了将MQ的使用相关的代码进行统一，所以将发送消息的代码、消费者的配置抽取到sl-express-mq工程中。
|
|||
|
|
|
|||
|
|
主要功能：
|
|||
|
|
|
|||
|
|
- 为RabbitTemplate设置了ReturnsCallback，如果消息发送到交换机但是没有到达队列，会进行日志的记录。
|
|||
|
|
- 统一了配置了消息的消费，消费者如果处理消息失败，会进行重试，如果依然是失败的话，会将错误消息发送到error.queue队列，后续需要人工进行处理。
|
|||
|
|
- 统一了发送消息代码，如果网络等异常情况导致发送消息失败会进行重试，如果依然失败的话将消息内容持久化到mysql数据库，后续通过xxl-job任务进行重新发送；如果其他情况导致失败，不会进行重试，直接存储消息到mysql数据库中。
|
|||
|
|
|
|||
|
|
## 2、SQL脚本
|
|||
|
|
|
|||
|
|
错误消息记录的SQL脚本：
|
|||
|
|
|
|||
|
|
~~~sql
|
|||
|
|
CREATE TABLE `sl_fail_msg` (
|
|||
|
|
`id` bigint NOT NULL,
|
|||
|
|
`msg_id` varchar(32) DEFAULT NULL COMMENT '消息id',
|
|||
|
|
`exchange` varchar(100) DEFAULT NULL COMMENT '交换机',
|
|||
|
|
`routing_key` varchar(50) DEFAULT NULL COMMENT '路由key',
|
|||
|
|
`msg` text COMMENT '消息内容',
|
|||
|
|
`reason` text CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci COMMENT '失败原因',
|
|||
|
|
`created` datetime DEFAULT NULL COMMENT '创建时间',
|
|||
|
|
`updated` datetime DEFAULT NULL COMMENT '更新时间',
|
|||
|
|
PRIMARY KEY (`id`),
|
|||
|
|
KEY `created` (`created`)
|
|||
|
|
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci COMMENT='失败消息记录表';
|
|||
|
|
~~~
|
|||
|
|
|
|||
|
|
## 3、使用
|
|||
|
|
|
|||
|
|
### 3.1、导入依赖
|
|||
|
|
|
|||
|
|
~~~xml
|
|||
|
|
<dependency>
|
|||
|
|
<groupId>com.sl-express.mq</groupId>
|
|||
|
|
<artifactId>sl-express-mq</artifactId>
|
|||
|
|
<version>1.0-SNAPSHOT</version>
|
|||
|
|
</dependency>
|
|||
|
|
~~~
|
|||
|
|
|
|||
|
|
### 3.2、启动类增加@EnableRetry
|
|||
|
|
|
|||
|
|
例如：
|
|||
|
|
|
|||
|
|
~~~java
|
|||
|
|
package com.sl;
|
|||
|
|
|
|||
|
|
import org.springframework.boot.SpringApplication;
|
|||
|
|
import org.springframework.boot.autoconfigure.SpringBootApplication;
|
|||
|
|
import org.springframework.cloud.openfeign.EnableFeignClients;
|
|||
|
|
import org.springframework.retry.annotation.EnableRetry;
|
|||
|
|
|
|||
|
|
@EnableRetry //开启重试机制
|
|||
|
|
@EnableFeignClients
|
|||
|
|
@SpringBootApplication
|
|||
|
|
public class DispatchApplication {
|
|||
|
|
|
|||
|
|
public static void main(String[] args) {
|
|||
|
|
SpringApplication.run(DispatchApplication.class, args);
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
~~~
|
|||
|
|
|
|||
|
|
### 3.3、编写配置
|
|||
|
|
|
|||
|
|
在springboot的配置文件中bootstrap-*.yml修改配置：
|
|||
|
|
|
|||
|
|
~~~yaml
|
|||
|
|
spring:
|
|||
|
|
rabbitmq: #mq的配置
|
|||
|
|
host: ${rabbitmq.host}
|
|||
|
|
port: ${rabbitmq.port}
|
|||
|
|
username: ${rabbitmq.username}
|
|||
|
|
password: ${rabbitmq.password}
|
|||
|
|
virtual-host: ${rabbitmq.virtual-host}
|
|||
|
|
publisher-confirm-type: correlated #发送消息的异步回调，记录消息是否发送成功
|
|||
|
|
publisher-returns: true #开启publish-return功能，消息到达交换机，但是没有到达对列表
|
|||
|
|
template:
|
|||
|
|
mandatory: true #消息路由失败时的策略, true: 调用ReturnCallback, false：丢弃消息
|
|||
|
|
listener:
|
|||
|
|
simple:
|
|||
|
|
acknowledge-mode: auto #，出现异常时返回nack，消息回滚到mq；没有异常，返回ack
|
|||
|
|
retry:
|
|||
|
|
enabled: true # 开启消费者失败重试
|
|||
|
|
initial-interval: 1000 # 初识的失败等待时长为1秒
|
|||
|
|
multiplier: 1 # 失败的等待时长倍数，下次等待时长 = multiplier * last-interval
|
|||
|
|
max-attempts: 3 # 最大重试次数
|
|||
|
|
stateless: true # true无状态；false有状态。如果业务中包含事务，这里改为false
|
|||
|
|
~~~
|
|||
|
|
|
|||
|
|
### 3.4、使用
|
|||
|
|
|
|||
|
|
~~~java
|
|||
|
|
//注入Service
|
|||
|
|
@Resource
|
|||
|
|
private MQService mqService;
|
|||
|
|
|
|||
|
|
....................
|
|||
|
|
|
|||
|
|
//发送消息
|
|||
|
|
this.mqService.sendMsg(exchange, null, msg);
|
|||
|
|
~~~
|
|||
|
|
|
|||
|
|
> **消息的消费，依然是SpringBoot的写法。例如：**
|
|||
|
|
>
|
|||
|
|
> ~~~java
|
|||
|
|
> @Component
|
|||
|
|
> public class OrderMQListener {
|
|||
|
|
>
|
|||
|
|
> @RabbitListener(bindings = @QueueBinding(
|
|||
|
|
> value = @Queue(name = "${rabbitmq.order.queue}"),
|
|||
|
|
> exchange = @Exchange(name = "${rabbitmq.order.exchange}", type = ExchangeTypes.TOPIC),
|
|||
|
|
> key = "#"
|
|||
|
|
> ))
|
|||
|
|
> public void listenOrderMsg(String msg) {
|
|||
|
|
>
|
|||
|
|
> //此处编写业务逻辑
|
|||
|
|
>
|
|||
|
|
> }
|
|||
|
|
> }
|
|||
|
|
> ~~~
|