Springboot整合RabbitMQ消息中间件

spring-boot-rabbitmq–消息中间件整合

前言：RabbitMQ的各种交换机说明

1、直连交换机

• 生产者发布消息时必须带着routing-key，队列绑定到交换机时必须指定binding-key ,且routing-key和binding-key必须完全相同，如此才能将消息路由到队列中
• 直连交换机通常用来循环分发任务给多个workers，例如在一个日志处理系统中，一个worker处理error级别日志，另外一个worker用来处理info级别的日志，此时生产者只需要在发送时指定特定的routing-key即可，绑定队列时binding-key只需要和routing-key保持一致即可接收到特定的消息。

2、扇形交换机

• 相对于直连交换机，扇形交换机没有路由设置

3、主题交换机

• routing-key必须由多个单词或者通配符组成，单词或者通配符之间使用.隔开，上限为255个字节;
• 通配符只能匹配一个单词；
• 通配符可以匹配零个或者多个单词；
  队列绑定交换机时的binding-key要能够匹配发送消息时的routing-key才能将消息路由到对应的队列；
• 根据routing-key和binding-key的匹配情况，消息可能进入单个队列，也可能进入多个队列，也可能丢失
• 主题队列的routing-key设置为#时，表示所有所有的队列都可以接收到消息，相当于fanout交换机；
• 主题队列的routing-key中不包含#或者*时，表示指定队列可以接收到消息，相当于direct交换机；

4、RabbitMQ有五种消息模式

• 无交换机模式
  • 简单模式【点对点模式】
  • 工作模式【一对多，资源争抢模式】
• 有交换机模式
  • 直连交换机【通过路由Key进行分配到不同队列】
  • 扇形交换机【发布订阅模式，即生产者将消息发布到订阅的队列上】
  • 主题交换机【通过主题标识分配，属于直连交换机的升级】

5、消息确认机制

• 不确认
• 自动确认
• 手动确认：

6、RabbitMQ可以作为RPC异步调用

一、基础快速入门

• 一个系统有消息发送也有消息接收，本示例采用发送和接收放到一个项目中
• 本示例采用简易配置，附录中有详细配置参数

1、添加依赖

<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

2、公共配置项整理

• listener.simple的concurrency和max-concurrency 为并发线程处理，
• rabbitmq.template.retry下的是生产者重试配置，listener.retry 为消费者重试配置
• publisher-confirms 是生产消息到达exchange后回调，publisher-returns 是exhcange路由到queue的回调

spring:application:name: rabbitmq-provider# 配置rabbitmqrabbitmq:# 连接地址，多个地址之间用都好隔开address: 192.168.1.82:5762, 192.168.1.82:5763,192.168.1.82:5764connection-timeout: 15000 # 连接超时时间virtual-host: /	# 虚拟主机的方式username: guestpassword: guest# 生产者配置publisher-confirm-type: simple  # 设置回调方式为simple，correlatedpublisher-confirms: true #开启消息到达exchange的回调，发送成功失败都会触发回调publisher-returns: true #开启消息从exhcange路由到queue的回调，只有路由失败时才会触发回调template:#为true时，如果exchange根据routingKey将消息路由到queue时找不到匹配的queue，# 触发return回调，为false时，exchange直接丢弃消息。mandatory: true# 配置重试机制retry:enabled: true # 开启发送机制max-attempts: 3 # 最大重试次数。默认为 3 。initial-interval: 1000 # 重试间隔，单位为毫秒。默认为 1000 。# 消费者监听器配置listener:simple:concurrency: 2 # 每个 @ListenerContainer 的并发消费的线程数max-concurrency: 10 # 每个 @ListenerCon 允许的并发消费的线程数acknowledge-mode: manual   #auto 自动确认，manual 手动确认，none 没有消息确认prefetch: 5# 重试机制retry:  # 配置重试机制enabled: true # 开启消费重试机制max-attempts: 3 # 最大重试次数。默认为 3 。initial-interval: 1000 # 重试间隔，单位为毫秒。默认为 1000 。

3、公用配置类整理

队列初始化工作可以在RabbitMQ的界面上创建，也可以采用代码的方式初始化，一般建议在管理平台上创建

一般需要主机交换机名称，队列名称，路由名称的使用

• DirectExchange(String name, boolean durable, boolean autoDelete) 直连交换机
• FanoutExchange(String name, boolean durable, boolean autoDelete) 扇形交换机
• TopicExchange(String name, boolean durable, boolean autoDelete) 主题交换机
• Queue(String name, boolean durable, boolean exclusive, boolean autoDelete) 队列
• Binding 绑定

@Configuration
public class RabbitConfig {//JSON序列化@Beanpublic MessageConverter messageConverter() {return new Jackson2JsonMessageConverter();}/*** 直连交换机配置*/public static class DirectExchangeDemoConfiguration {// 创建 Queue@Beanpublic Queue demo01Queue() {return new Queue(Demo01Message.QUEUE, // Queue 名字true, // durable: 是否持久化false, // exclusive: 是否排它false); // autoDelete: 是否自动删除}.build();}// 创建 Direct Exchange@Beanpublic DirectExchange demo01Exchange() {return new DirectExchange(Demo01Message.EXCHANGE,true,  // durable: 是否持久化false);  // exclusive: 是否排它}// 创建 Binding// Exchange：Demo01Message.EXCHANGE// Routing key：Demo01Message.ROUTING_KEY// Queue：Demo01Message.QUEUE@Beanpublic Binding demo01Binding() {return BindingBuilder.bind(demo01Queue()).to(demo01Exchange()).with(Demo01Message.ROUTING_KEY);}}/*** 主题交换机 */public static class TopicExchangeDemoConfiguration {// 创建 Queue@Beanpublic Queue demo02Queue() {return new Queue(Demo02Message.QUEUE, // Queue 名字true, // durable: 是否持久化false, // exclusive: 是否排它false); // autoDelete: 是否自动删除}// 创建 Topic Exchange@Beanpublic TopicExchange demo02Exchange() {return new TopicExchange(Demo02Message.EXCHANGE,true,  // durable: 是否持久化false);  // exclusive: 是否排它}// 创建 Binding// Exchange：Demo02Message.EXCHANGE// Routing key：Demo02Message.ROUTING_KEY// Queue：Demo02Message.QUEUE@Beanpublic Binding demo02Binding() {return BindingBuilder.bind(demo02Queue()).to(demo02Exchange()).with(Demo02Message.ROUTING_KEY);}}/*** 扇形交换机*/public static class FanoutExchangeDemoConfiguration {// 创建 Queue A@Beanpublic Queue demo03QueueA() {return new Queue(Demo03Message.QUEUE_A, // Queue 名字true, // durable: 是否持久化false, // exclusive: 是否排它false); // autoDelete: 是否自动删除}// 创建 Queue B@Beanpublic Queue demo03QueueB() {return new Queue(Demo03Message.QUEUE_B, // Queue 名字true, // durable: 是否持久化false, // exclusive: 是否排它false); // autoDelete: 是否自动删除}// 创建 Fanout Exchange@Beanpublic FanoutExchange demo03Exchange() {return new FanoutExchange(Demo03Message.EXCHANGE,true,  // durable: 是否持久化false);  // exclusive: 是否排它}// 创建 Binding A// Exchange：Demo03Message.EXCHANGE// Queue：Demo03Message.QUEUE_A@Beanpublic Binding demo03BindingA() {return BindingBuilder.bind(demo03QueueA()).to(demo03Exchange());}// 创建 Binding B// Exchange：Demo03Message.EXCHANGE// Queue：Demo03Message.QUEUE_B@Beanpublic Binding demo03BindingB() {return BindingBuilder.bind(demo03QueueB()).to(demo03Exchange());}}/*** HeadersExchange 示例的配置类*/public static class HeadersExchangeDemoConfiguration {// 创建 Queue@Beanpublic Queue demo04Queue() {return new Queue(Demo04Message.QUEUE, // Queue 名字true, // durable: 是否持久化false, // exclusive: 是否排它false); // autoDelete: 是否自动删除}// 创建 Headers Exchange@Beanpublic HeadersExchange demo04Exchange() {return new HeadersExchange(Demo04Message.EXCHANGE,true,  // durable: 是否持久化false);  // exclusive: 是否排它}// 创建 Binding// Exchange：Demo04Message.EXCHANGE// Queue：Demo04Message.QUEUE// Headers: Demo04Message.HEADER_KEY + Demo04Message.HEADER_VALUE@Beanpublic Binding demo4Binding() {return BindingBuilder.bind(demo04Queue()).to(demo04Exchange()).where(Demo04Message.HEADER_KEY).matches(Demo04Message.HEADER_VALUE); // 配置 Headers 匹配}}}

4、消息生产者和消费者类

一般建议，一类消息生产采用一个类，做到职责单一。

交换机，路由，和主题的名称最好采用枚举或者常量的方式定义

注意下面的消息生产者和消息消费者的交换机方式是一一对应的

二、消息生产者

• rabbitTemplate#convertAndSend 方法实现所有形式的消息发放
• rabbitTemplate.setConfirmCallback 设置消息回调
• 直连交换机是常用的方式

RabbitMq消息生产者实例

public class rabbitMqProducer{//点对点模式public void send(String msg){amqpTemplate.convertAndSend(SIMPLE_QUEUE_NAME, msg );}//工作模式public void work() throws InterruptedException {String msg = "这是一个work模式";for (int i = 0; i < 10; i++) {amqpTemplate.convertAndSend(WORKER_QUEUE_NAME, msg + i);Thread.sleep(5000);}}//发送同步消息@Overridepublic void sendMessage(Object message) {rabbitTemplate.convertAndSend(Constants.SAVE_USER_EXCHANGE_NAME,Constants.SAVE_USER_QUEUE_ROUTE_KEY,message, correlationData);log.info("发送消息到RabbitMQ, 消息内容: " + message);}//发送异步消息@Asyncpublic ListenableFuture<Void> asyncSend(Integer id) {try {//设置消息属性MessageProperties messageProperties = new MessageProperties();messageProperties.setMessageId(UUID.randomUUID().toString());messageProperties.setContentType("text/plain");messageProperties.setContentEncoding("utf-8");// 发送消息Message message = new Message(messageStr.getBytes(), messageProperties);message.setId(id);// 同步发送消息rabbitTemplate.convertAndSend(Demo01Message.EXCHANGE, Demo01Message.ROUTING_KEY, message);// 返回成功的 Futurereturn AsyncResult.forValue(null);} catch (Throwable ex) {// 返回异常的 Futurereturn AsyncResult.forExecutionException(ex);}}//扇形交换机public void sendMessage(Object message) {log.info("【消息发送者】发送消息到fanout交换机，消息内容为: {}", message);rabbitTemplate.convertAndSend(Constants.FANOUT_EXCHANGE_NAME, "", message);}//带消息序列的消息public void sendMessage() {//采用内部类方式for (int i = 1; i <= 3; i++) {CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());logger.info("【Producer】发送的消费ID = {}", correlationData.getId());String msg = "hello confirm message" + i;logger.info("【Producer】发送的消息 = {}", msg);rabbitTemplate.convertAndSend(EXCHANGE_NAME, ROUTING_KEY, msg, correlationData);}}//主题交换机//确认回调接口final ConfirmCallback confirmCallback = new RabbitTemplate.ConfirmCallback() {/*** 	@param ack broker 是否落盘成功* 	@param cause 失败的一些异常信息*/@Overridepublic void confirm(CorrelationData correlationData,boolean ack, String cause) {System.err.println("消息ACK结果:" + ack+ ", correlationData: " + correlationData.getId());}};//  简单的主题消息public void sendMessage1(Object message,Map<String, Object> properties) throws Exception {//创建消息MessageHeaders mhs = new MessageHeaders(properties);Message<?> msg = MessageBuilder.createMessage(message, mhs);rabbitTemplate.setConfirmCallback(confirmCallback);// 	指定业务唯一的iDCorrelationData correlationData =new CorrelationData(UUID.randomUUID().toString());//消息处理器MessagePostProcessor mpp = new MessagePostProcessor() {@Overridepublic Message postProcessMessage(Message message) throws AmqpException {System.err.println("---> post to do: " + message);return message;}};//消息发送，交换机，路由rabbitTemplate.convertAndSend("exchange-1",   "route.01",  msg,mpp,  correlationData);}//消息发送public void syncSend(Integer id, String headerValue) {// 创建 MessageProperties 属性MessageProperties messageProperties = new MessageProperties();messageProperties.setHeader(Demo04Message.HEADER_KEY, headerValue); // 设置 header// 创建 Message 消息Message message = rabbitTemplate.getMessageConverter().toMessage(new Demo04Message().setId(id), messageProperties);// 同步发送消息rabbitTemplate.send(Demo04Message.EXCHANGE, null, message);}
}
@Data
public class Demo07Message implements Serializable {public static final String QUEUE = "QUEUE_DEMO_07"; // 正常队列public static final String DEAD_QUEUE = "DEAD_QUEUE_DEMO_07"; // 死信队列public static final String EXCHANGE = "EXCHANGE_DEMO_07";public static final String ROUTING_KEY = "ROUTING_KEY_07"; // 正常路由键public static final String DEAD_ROUTING_KEY = "DEAD_ROUTING_KEY_07"; // 死信路由键private Integer id;private String body;
}

6、Headers交换机

Headers Exchange 不依赖于 routing key 与 binding key 的匹配规则来路由消息，而是根据发送的消息内容中的 headers 属性进行匹配

//消息体重定义HEADER_KEY
public class Demo04Message implements Serializable {public static final String QUEUE = "QUEUE_DEMO_04_A";public static final String EXCHANGE = "EXCHANGE_DEMO_04";public static final String HEADER_KEY = "color";public static final String HEADER_VALUE = "red";/*** 编号*/private Integer id;// ... 省略 set/get/toString 方法}

三、消息消费者

• @RabbitHandler 只能放在方法上，表明这是一个消息处理方法
• @RabbitListener 可以放在方法上和类上，一般使用在方法上
  • queues 指定队列名称
  • concurrency 并发消费

消息消费者示例

public class RabbitMQConsumerDemo{// 通过注解自动创建 spring.simple.queue 队列@RabbitListener(queuesToDeclare = @Queue("spring.simple.queue"))public void listen(String msg) {System.out.println("简单队列 接收到消息：" + msg);}// 通过注解自动创建 spring.work.queue 队列@RabbitListener(queuesToDeclare = @Queue("spring.work.queue"))public void listen(String msg) {System.out.println("work模式 接收到消息：" + msg);}// 创建两个队列共同消费@RabbitListener(queuesToDeclare = @Queue("spring.work.queue"))public void listen2(String msg) {System.out.println("work模式二 接收到消息：" + msg);}//直连交换机示例@RabbitHandler@RabbitListener(bindings = @QueueBinding(value = @Queue(value = "customer.code.wechatCallback",autoDelete = "false"),exchange = @Exchange(value = "code.wechatCallback",type = ExchangeTypes.FANOUT)),concurrency = "2")public void receiveMessage01(String msg, Channel channel, Message message) throws IOException {try {// 这里模拟一个空指针异常，String string = null;string.length();//消费端限流channel.basicQos(3,false);log.info("【Consumer01成功接收到消息】>>> {}", msg);// 确认收到消息，只确认当前消费者的一个消息收到channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);} catch (Exception e) {if (message.getMessageProperties().getRedelivered()) {log.info("【Consumer01】消息已经回滚过，拒绝接收消息 ： {}", msg);// 拒绝消息，并且不再重新进入队列channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);} else {log.info("【Consumer01】消息即将返回队列重新处理 ：{}", msg);//设置消息重新回到队列处理// requeue表示是否重新回到队列，true重新入队channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);}}}//扇形交换机示例@RabbitHandler@RabbitListener(bindings = @QueueBinding(value = @Queue(value = "customer.code.wechatCallback",autoDelete = "false"),exchange = @Exchange(value = "code.wechatCallback",type = ExchangeTypes.FANOUT)),concurrency = "1")public void receiveMessage(Object message) {logger.info("消息接收者接收到来自【队列一】的消息，消息内容: {}", message);}//主题交换机@RabbitHandler@RabbitListener(bindings =  @QueueBinding(value= @Queue(value = "queue-1",durable = "true",autoDelete = "false"),exchange =  @Exchange(name = "exchange-1",durable = "true",type = ExchangeTypes.TOPICignoreDeclarationExceptions = "true"),key = "route.*"))public void onMessage(Message message, Channel channel,String body) throws Exception {//	1. 收到消息以后进行业务端消费处理System.err.println("消费消息:" + message.getPayload());//设置手工牵手Long deliveryTag = (Long)message.getHeaders().get(AmqpHeaders.DELIVERY_TAG);//手动确认消息channel.basicAck(deliveryTag, false);}//header交换机@Component@RabbitListener(queues = Demo04Message.QUEUE)public class Demo04Consumer {private Logger logger = LoggerFactory.getLogger(getClass());@RabbitHandlerpublic void onMessage(Demo04Message message) {logger.info("[onMessage][线程编号:{} 消息内容：{}]", Thread.currentThread().getId(), message);}}
}

2、重试机制【消息消费者】

• 方式一：通过配置项的方式，详细看公用配置项的设置，改方式的优点是简单，缺点是不容易控制重试的时间
• 方式二：结合MySQL或Redis等持久化方式控制重试次数，该方式的优点是可以自由控制重试的间隔时间，缺点是比较复杂

五、回调消息【消息生产者】

（1）方式一：自定义消息回调处理类

@Slf4j
@Component
public class CustomConfirmAndReturnCallback implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnCallback {@Autowiredprivate RabbitTemplate rabbitTemplate;/*** PostConstruct: 用于在依赖关系注入完成之后需要执行的方法上，以执行任何初始化.*/@PostConstructpublic void init() {//指定 ConfirmCallbackrabbitTemplate.setConfirmCallback(this);//指定 ReturnCallbackrabbitTemplate.setReturnCallback(this);}/*** 消息从交换机成功到达队列，则returnedMessage方法不会执行;* 消息从交换机未能成功到达队列，则returnedMessage方法会执行;*/@Overridepublic void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {log.info("returnedMessage回调方法>>>" + new String(message.getBody(), StandardCharsets.UTF_8) + ",replyCode:" + replyCode + ",replyText:" + replyText + ",exchange:" + exchange + ",routingKey:" + routingKey);}/*** 如果消息没有到达交换机,则该方法中isSendSuccess = false,error为错误信息;* 如果消息正确到达交换机,则该方法中isSendSuccess = true;*/@Overridepublic void confirm(CorrelationData correlationData, boolean isSendSuccess, String error) {log.info("confirm回调方法>>>回调消息ID为: " + correlationData.getId());if (isSendSuccess) {logger.info("confirm回调方法>>>消息发送到交换机成功！");} else {logger.info("confirm回调方法>>>消息发送到交换机失败！，原因 : [{}]", error);}}}

public class callbackConfirmDemo{@AutoWiredprivate CustomConfirmAndReturnCallback  confirmCallback;//  简单的主题消息public void sendMessage1(Object message,Map<String, Object> properties) throws Exception {//创建消息MessageHeaders mhs = new MessageHeaders(properties);Message<?> msg = MessageBuilder.createMessage(message, mhs);// 使用自定义的确认回调rabbitTemplate.setConfirmCallback(confirmCallback);// 	指定业务唯一的iDCorrelationData correlationData =new CorrelationData(UUID.randomUUID().toString());//消息处理器MessagePostProcessor mpp = new MessagePostProcessor() {@Overridepublic Message postProcessMessage(Message message) throws AmqpException {System.err.println("---> post to do: " + message);return message;}};//消息发送rabbitTemplate.convertAndSend("exchange-1",   "springboot.rabbit",  msg,mpp,  correlationData);}//设置回调public void syncSend(Integer id) {// 创建 Demo13Message 消息Demo13Message message = new Demo13Message();message.setId(id);// 同步发送消息rabbitTemplate.invoke(new RabbitOperations.OperationsCallback<Object>() {@Overridepublic Object doInRabbit(RabbitOperations operations) {// 同步发送消息operations.convertAndSend(Demo13Message.EXCHANGE, Demo13Message.ROUTING_KEY, message);logger.info("[doInRabbit][发送消息完成]");// 等待确认operations.waitForConfirms(0); // timeout 参数，如果传递 0 ，表示无限等待logger.info("[doInRabbit][等待 Confirm 完成]");return null;}}, new ConfirmCallback() {@Overridepublic void handle(long deliveryTag, boolean multiple) throws IOException {logger.info("[handle][Confirm 成功]");}}, new ConfirmCallback() {@Overridepublic void handle(long deliveryTag, boolean multiple) throws IOException {logger.info("[handle][Confirm 失败]");}});}//设置消息回调public void sendMessage(){//设置消息消费端确认回调rabbitTemplate.setConfirmCallback((correlationData, ack, cause)->{if(ack){log.info("消息{}接收成功",correlationData.getId());}else{log.info("消息{}接收失败，原因{}",correlationData.getId(),cause);}});//设置消费端返回回调rabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey)->{log.info("消息{}发送失败，应答码{}，原因{}，交换机{}，路由键{}",message.toString(),replyCode,replyText,exchange,routingKey);});rabbitTemplate.convertAndSend(Constants.FANOUT_EXCHANGE_NAME, "", message);}
}

（2）方式二：匿名内部类方式处理

六、批量消息生产和消费

方式一：通过application,yaml配置项中添加批量生产和消费相关配置可以实现

方式二：通过手动的方式，设置BatchingRabbitTemplate和SimpleRabbitListenerContainerFactory实现

方式三： 批量消费中，通过注解配置 concurrency = "2"形式类配置并发批量消费

注意，批量消费和批量生产不一定需要严格搭配，按照各自需求来即可

1、在配置类中添加

• BatchingRabbitTemplate
• SimpleRabbitListenerContainerFactory

@Configuration
public class RabbitConfig {//手动设置批量处理@Beanpublic BatchingRabbitTemplate batchRabbitTemplate(ConnectionFactory connectionFactory) {// 创建 BatchingStrategy 对象，代表批量策略int batchSize = 16384; // 超过收集的消息数量的最大条数。int bufferLimit = 33554432; // 每次批量发送消息的最大内存int timeout = 30000; // 超过收集的时间的最大等待时长，单位：毫秒BatchingStrategy batchingStrategy = new SimpleBatchingStrategy(batchSize, bufferLimit, timeout);// 创建 TaskScheduler 对象，用于实现超时发送的定时器TaskScheduler taskScheduler = new ConcurrentTaskScheduler();// 创建 BatchingRabbitTemplate 对象BatchingRabbitTemplate batchTemplate = new BatchingRabbitTemplate(batchingStrategy, taskScheduler);batchTemplate.setConnectionFactory(connectionFactory);return batchTemplate;}// 批量消息监听@Bean(name = "consumerBatchContainerFactory")public SimpleRabbitListenerContainerFactory consumerBatchContainerFactory(SimpleRabbitListenerContainerFactoryConfigurer configurer, ConnectionFactory connectionFactory) {// 创建 SimpleRabbitListenerContainerFactory 对象SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();configurer.configure(factory, connectionFactory);// 额外添加批量消费的属性factory.setBatchListener(true);factory.setBatchSize(10);factory.setReceiveTimeout(30 * 1000L);factory.setConsumerBatchEnabled(true);return factory;}
}

2、批量生产

• BatchingRabbitTemplate

@Component
public class Demo06Producer {@Autowiredprivate BatchingRabbitTemplate batchingRabbitTemplate;public void syncSend(Integer id) {// 创建 Demo05Message 消息Demo05Message message = new Demo05Message();message.setId(id);// 同步发送消息batchingRabbitTemplate.convertAndSend(Demo05Message.EXCHANGE, Demo05Message.ROUTING_KEY, message);}}

2、批量消费

• containerFactory

@Component
@RabbitListener(queues = Demo05Message.QUEUE, concurrency = "2",containerFactory = "consumerBatchContainerFactory")
public class Demo05Consumer {private Logger logger = LoggerFactory.getLogger(getClass());@RabbitHandlerpublic void onMessage(List<Demo05Message> messages) {logger.info("[onMessage][线程编号:{} 消息数量：{}]", Thread.currentThread().getId(), messages.size());}}

七、延迟队列

方式一：采用RabbitMQ的控制台方式添加延迟队列

方式二：采用程序的方式设置延迟队列

1、配置延迟队列

// 创建 Queue
public class delayQueueDemo{@Beanpublic Queue demo08Queue() {return QueueBuilder.durable(Demo08Message.QUEUE) // durable: 是否持久化.exclusive() // exclusive: 是否排它.autoDelete() // autoDelete: 是否自动删除.ttl(10 * 1000) // 设置队列里的默认过期时间为 10 秒.deadLetterExchange(Demo08Message.EXCHANGE).deadLetterRoutingKey(Demo08Message.DELAY_ROUTING_KEY).build();}
}public class delayProducerDemo{public void syncSend(Integer id, Integer delay) {// 创建 Demo07Message 消息Demo08Message message = new Demo08Message();message.setId(id);// 同步发送消息rabbitTemplate.convertAndSend(Demo08Message.EXCHANGE, Demo08Message.ROUTING_KEY, message,new MessagePostProcessor() {@Overridepublic Message postProcessMessage(Message message) throws AmqpException {// 设置消息的 TTL 过期时间if (delay != null && delay > 0) {message.getMessageProperties().setExpiration(String.valueOf(delay)); // Spring-AMQP API 设计有问题，所以传入了 String = =}return message;}});}
}

八、消费异常处理

1、自定义异常处理

@Component("rabbitListenerErrorHandler")
public class RabbitListenerErrorHandlerImpl implements RabbitListenerErrorHandler {private Logger logger = LoggerFactory.getLogger(getClass());@Overridepublic Object handleError(Message amqpMessage, org.springframework.messaging.Message<?> message,ListenerExecutionFailedException exception) {// 打印异常日志logger.error("[handleError][amqpMessage:[{}] message:[{}]]", amqpMessage, message, exception);// 直接继续抛出异常throw exception;}}
@Component
public class RabbitLoggingErrorHandler implements ErrorHandler {private Logger logger = LoggerFactory.getLogger(getClass());public RabbitLoggingErrorHandler(SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory) {rabbitListenerContainerFactory.setErrorHandler(this);}@Overridepublic void handleError(Throwable t) {logger.error("[handleError][发生异常]]", t);}}

2、消费者添加监听处理异常

@Component
@RabbitListener(queues = Demo16Message.QUEUE,errorHandler = "rabbitListenerErrorHandler")
//@RabbitListener(queues = Demo15Message.QUEUE)
public class Demo16Consumer {private Logger logger = LoggerFactory.getLogger(getClass());@RabbitHandlerpublic void onMessage(Demo16Message message) {logger.info("[onMessage][线程编号:{} 消息内容：{}]", Thread.currentThread().getId(), message);// 模拟消费异常throw new RuntimeException("你猜");}}

九、广播消费和集群消费

 在 RabbitMQ 中，如果多个 Consumer 订阅相同的 Queue ，那么每一条消息有且仅会被一个 Consumer 所消费。这个特性，就为我们实现集群消费提供了基础。在广播消费概念中，如果多个 Consumer 订阅相同的 Queue ，我们可以通过给每个 Consumer 创建一个其独有 Queue ，从而保证都能接收到全量的消息。同时，RabbitMQ 支持队列的自动删除，所以我们可以在 Consumer 关闭的时候，通过该功能删除其独有的 Queue如何实现广播消费，

@Component
@RabbitListener(bindings = @QueueBinding(value = @Queue(name = BroadcastMessage.QUEUE + "-" + "#{T(java.util.UUID).randomUUID()}",autoDelete = "true"),exchange = @Exchange(name = BroadcastMessage.EXCHANGE,type = ExchangeTypes.TOPIC,declare = "false"))
)
public class BroadcastConsumer {private Logger logger = LoggerFactory.getLogger(getClass());@RabbitHandlerpublic void onMessage(BroadcastMessage message) {logger.info("[onMessage][线程编号:{} 消息内容：{}]", Thread.currentThread().getId(), message);}} 

十、关于死信队列的处理方式

• 首先，死信队列的绑定，参考公用配置类

• 方式一：通过配置类自动创建死信交换机，队列并绑定，同时在正常的队列中通过Map的形式添加死信队列的配置

• 方式二：在RabbitMQ的工作台上手动给队列添加死信队列，

• 如果死信队列积累较多，可以通过死信队列监听器来处理死信队列数据

1、定义死信队列的绑定关系

public  class DirectExchangeDemoConfiguration {@Beanpublic DirectExchange dlxExchange(){//死信交换机return new DirectExchange(dlxExchangeName);}@Beanpublic Queue dlxQueue(){//死信队列return new Queue(dlxQueueName);}@Beanpublic Binding dlcBinding(Queue dlxQueue, DirectExchange dlxExchange){//死信队列绑定交换机return BindingBuilder.bind(dlxQueue).to(dlxExchange).with(dlxRoutingKey);}//正常队列@Beanpublic Queue queue(){// 正常队列中通过参数绑定死信队列Map<String,Object> params = new HashMap<>();params.put("x-dead-letter-exchange",dlxExchangeName);//声明当前队列绑定的死信交换机params.put("x-dead-letter-routing-key",dlxRoutingKey);//声明当前队列的死信路由键return QueueBuilder.durable(queueName).withArguments(params).build();}// 创建 Direct Exchange@Beanpublic DirectExchange demo01Exchange() {return new DirectExchange(Demo01Message.EXCHANGE,true,  // durable: 是否持久化false);  // exclusive: 是否排它}// 创建 Binding// Exchange：Demo01Message.EXCHANGE// Routing key：Demo01Message.ROUTING_KEY// Queue：Demo01Message.QUEUE@Beanpublic Binding demo01Binding() {return BindingBuilder.bind(demo01Queue()).to(demo01Exchange()).with(Demo01Message.ROUTING_KEY);}
}

2、定义死信队列的消费者

@Component
@RabbitListener(queues = Demo07Message.DEAD_QUEUE)
public class Demo07DeadConsumer {private Logger logger = LoggerFactory.getLogger(getClass());@RabbitHandlerpublic void onMessage(Demo07Message message) {logger.info("[onMessage][【死信队列】线程编号:{} 消息内容：{}]", Thread.currentThread().getId(), message);}}

十一、顺序队列

• 普通顺序消息 ：Producer 将相关联的消息发送到相同的消息队列。
• 完全严格顺序 ：在【普通顺序消息】的基础上，Consumer 严格顺序消费。

1、定义多个队列

// Demo10Message.javapublic class Demo10Message implements Serializable {private static final String QUEUE_BASE = "QUEUE_DEMO_10-";public static final String QUEUE_0 = QUEUE_BASE + "0";public static final String QUEUE_1 = QUEUE_BASE + "1";public static final String QUEUE_2 = QUEUE_BASE + "2";public static final String QUEUE_3 = QUEUE_BASE + "3";public static final int QUEUE_COUNT = 4;public static final String EXCHANGE = "EXCHANGE_DEMO_10";/*** 编号*/private Integer id;// ... 省略 set/get/toString 方法}

2、实现顺序队列的绑定关系

   
public  class DirectExchangeDemoConfiguration {// 创建 Queue@Beanpublic Queue demo10Queue0() {return new Queue(Demo10Message.QUEUE_0);}@Beanpublic Queue demo10Queue1() {return new Queue(Demo10Message.QUEUE_1);}@Beanpublic Queue demo10Queue2() {return new Queue(Demo10Message.QUEUE_2);}@Beanpublic Queue demo10Queue3() {return new Queue(Demo10Message.QUEUE_3);}// 创建 Direct Exchange@Beanpublic DirectExchange demo10Exchange() {return new DirectExchange(Demo10Message.EXCHANGE,true,  // durable: 是否持久化false);  // exclusive: 是否排它}// 创建 Binding@Beanpublic Binding demo10Binding0() {return BindingBuilder.bind(demo10Queue0()).to(demo10Exchange()).with("0");}@Beanpublic Binding demo10Binding1() {return BindingBuilder.bind(demo10Queue1()).to(demo10Exchange()).with("1");}@Beanpublic Binding demo10Binding2() {return BindingBuilder.bind(demo10Queue2()).to(demo10Exchange()).with("2");}@Beanpublic Binding demo10Binding3() {return BindingBuilder.bind(demo10Queue3()).to(demo10Exchange()).with("3");}}

2、发送顺序队列消息

// Demo10Producer.java@Component
public class Demo10Producer {@Autowiredprivate RabbitTemplate rabbitTemplate;public void syncSend(Integer id) {// 创建 Demo10Message 消息Demo10Message message = new Demo10Message();message.setId(id);// 同步发送消息rabbitTemplate.convertAndSend(Demo10Message.EXCHANGE, this.getRoutingKey(id), message);}private String getRoutingKey(Integer id) {return String.valueOf(id % Demo10Message.QUEUE_COUNT);}}

3、消费顺序队列消息

// Demo10Consumer.java@Component
@RabbitListener(queues = Demo10Message.QUEUE_0)
@RabbitListener(queues = Demo10Message.QUEUE_1)
@RabbitListener(queues = Demo10Message.QUEUE_2)
@RabbitListener(queues = Demo10Message.QUEUE_3)
public class Demo10Consumer {private Logger logger = LoggerFactory.getLogger(getClass());@RabbitHandler(isDefault = true)public void onMessage(Message<Demo10Message> message) {logger.info("[onMessage][线程编号:{} Queue:{} 消息编号：{}]", Thread.currentThread().getId(), getQueue(message),message.getPayload().getId());}private static String getQueue(Message<Demo10Message> message) {return message.getHeaders().get("amqp_consumerQueue", String.class);}}

十二、事务消息

1、创建事务消息配置

// RabbitConfig.java@Configuration
@EnableTransactionManagement
public class RabbitConfig {/*** Direct Exchange 示例的配置类*/public static class DirectExchangeDemoConfiguration {// 创建 Queue@Beanpublic Queue demo11Queue() {return new Queue(Demo11Message.QUEUE, // Queue 名字true, // durable: 是否持久化false, // exclusive: 是否排它false); // autoDelete: 是否自动删除}// 创建 Direct Exchange@Beanpublic DirectExchange demo11Exchange() {return new DirectExchange(Demo11Message.EXCHANGE,true,  // durable: 是否持久化false);  // exclusive: 是否排它}// 创建 Binding// Exchange：Demo11Message.EXCHANGE// Routing key：Demo11Message.ROUTING_KEY// Queue：Demo11Message.QUEUE@Beanpublic Binding demo11Binding() {return BindingBuilder.bind(demo11Queue()).to(demo11Exchange()).with(Demo11Message.ROUTING_KEY);}}@Beanpublic RabbitTransactionManager rabbitTransactionManager(ConnectionFactory connectionFactory, RabbitTemplate rabbitTemplate) {// <Y> 设置 RabbitTemplate 支持事务rabbitTemplate.setChannelTransacted(true);// 创建 RabbitTransactionManager 对象return new RabbitTransactionManager(connectionFactory);}}

2、生产事务消息

• 注意@Transactional注解

@Component
public class Demo11Producer {private Logger logger = LoggerFactory.getLogger(getClass());@Autowiredprivate RabbitTemplate rabbitTemplate;@Transactionalpublic void syncSend(Integer id) throws InterruptedException {// 创建 Demo11Message 消息Demo11Message message = new Demo11Message();message.setId(id);// 同步发送消息rabbitTemplate.convertAndSend(Demo11Message.EXCHANGE, Demo11Message.ROUTING_KEY, message);logger.info("[syncSend][发送编号：[{}] 发送成功]", id);// <X> 等待Thread.sleep(10 * 1000L);}}

3、消费事务消息

// Demo11ProducerTest.java@RunWith(SpringRunner.class)
@SpringBootTest(classes = Application.class)
public class Demo11ProducerTest {@Autowiredprivate Demo11Producer producer;@Testpublic void testSyncSend() throws InterruptedException {int id = (int) (System.currentTimeMillis() / 1000);producer.syncSend(id);// 阻塞等待，保证消费new CountDownLatch(1).await();}}

十三、消息确认

1、修改配置文件

# 同步确认
spring.rabbitmq.listener.simple. acknowledge-mode=manual
spring.rabbitmq.publisher-confirm-type=simple# 异步确认
spring.rabbitmq.publisher-confirm-type=correlated

2、消费确认

• channel.basicAck

@Component
@RabbitListener(queues = Demo12Message.QUEUE)
public class Demo12Consumer {private Logger logger = LoggerFactory.getLogger(getClass());@RabbitHandlerpublic void onMessage(Demo12Message message, Channel channel,@Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag) throws IOException {logger.info("[onMessage][线程编号:{} 消息内容：{}]", Thread.currentThread().getId(), message);// 提交消费进度if (message.getId() % 2 == 1) {// ack 确认消息// 第二个参数 multiple ，用于批量确认消息，为了减少网络流量，手动确认可以被批处。// 1. 当 multiple 为 true 时，则可以一次性确认 deliveryTag 小于等于传入值的所有消息// 2. 当 multiple 为 false 时，则只确认当前 deliveryTag 对应的消息channel.basicAck(deliveryTag, false);}}}

3、发送确认

// Demo13Producer.java@Component
public class Demo13Producer {private Logger logger = LoggerFactory.getLogger(getClass());@Autowiredprivate RabbitTemplate rabbitTemplate;public void syncSend(Integer id) {// 创建 Demo13Message 消息Demo13Message message = new Demo13Message();message.setId(id);// 同步发送消息rabbitTemplate.invoke(new RabbitOperations.OperationsCallback<Object>() {@Overridepublic Object doInRabbit(RabbitOperations operations) {// 同步发送消息operations.convertAndSend(Demo13Message.EXCHANGE, Demo13Message.ROUTING_KEY, message);logger.info("[doInRabbit][发送消息完成]");// 等待确认operations.waitForConfirms(0); // timeout 参数，如果传递 0 ，表示无限等待logger.info("[doInRabbit][等待 Confirm 完成]");return null;}}, new ConfirmCallback() {@Overridepublic void handle(long deliveryTag, boolean multiple) throws IOException {logger.info("[handle][Confirm 成功]");}}, new ConfirmCallback() {@Overridepublic void handle(long deliveryTag, boolean multiple) throws IOException {logger.info("[handle][Confirm 失败]");}});}}

4、定义异步确认的回调类setConfirmCallback

• rabbitTemplate.setConfirmCallback 配置了全局的异步确认

@Component
public class RabbitProducerConfirmCallback implements RabbitTemplate.ConfirmCallback {private Logger logger = LoggerFactory.getLogger(getClass());public RabbitProducerConfirmCallback(RabbitTemplate rabbitTemplate) {rabbitTemplate.setConfirmCallback(this);}@Overridepublic void confirm(CorrelationData correlationData, boolean ack, String cause) {if (ack) {logger.info("[confirm][Confirm 成功 correlationData: {}]", correlationData);} else {logger.error("[confirm][Confirm 失败 correlationData: {} cause: {}]", correlationData, cause);}}}

5、返回确认ReturnCallback

// RabbitProducerReturnCallback.java@Component
public class RabbitProducerReturnCallback implements RabbitTemplate.ReturnCallback {private Logger logger = LoggerFactory.getLogger(getClass());public RabbitProducerReturnCallback(RabbitTemplate rabbitTemplate) {rabbitTemplate.setReturnCallback(this);}@Overridepublic void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {logger.error("[returnedMessage][message: [{}] replyCode: [{}] replyText: [{}] exchange: [{}] routingKey: [{}]]",message, replyCode, replyText, exchange, routingKey);}}

RabbitMQ的消息高可用+重试方案落地

• 1、通过MQ消息失败记录表记录失败记录
• 2、通过xxl-job定时扫描数据表，然后对于超过重试次数的进行重试。

一、消息消费

1、创建一个抽象的模板类

@Slf4j
public abstract class AbstractConsumer {/*** 定义静态内部类指定监听队列名称*/static class queueName{public static  final  String ORDER_CANCEL_QUEUE = "queue.assets_center.unification.order_cancel";public static  final  String HOUR_EXCHANGE_REFUND_QUEUE = "queue.assets_center.unification.convert_in_wandou_packet_revert";}@Resourceprivate IMqConsumeService mqConsumeService;@Autowiredprivate RetryMessageMapper retryMessageMapper;@Autowiredprivate DingDingHelper dingDingHelper;private static  final String MQ_HEADER_REF = "spring_returned_message_correlation";/*** 消息订阅入口方法*/@SneakyThrowsprotected void doConsume(Channel channel, Message message, String body) {String messageId = (String) message.getMessageProperties().getHeaders().get(MQ_HEADER_REF);try {//1、添加traceIdMDC.put("X-TRACE-ID", messageId);log.info("[{}] 消费消息: messageId:{}, mq参数:{}", getConsumeType().getMessage(), messageId, body);//2、保存消息日志以及幂等性判断boolean check = this.saveConsumeLogAndJudgeIdempotent(channel, message, body);if(check){//3、业务处理this.bizHandle(body);//4、手工ackchannel.basicAck(message.getMessageProperties().getDeliveryTag(), false);}} catch (Exception exception) {log.error("消息消费异常, messageId = " + messageId, exception);//1、异常通知dingDingHelper.sendAlert(getConsumeType().getMessage() + "- 消息消费异常","消息内容:" + body);//2、异常记录,并启动重试机制RetryMessage retryMessage = new RetryMessage(getQueueEnum(),getTimedMaxRetryTimes(),getTimedRecompenseBizMethod(),messageId,body,exception);retryMessageMapper.insert(retryMessage);//3、响应消费失败channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, false);} finally {MDC.remove("X-TRACE-ID");}}/*** 保存消费日志 & 消息幂等判断*/@SneakyThrowsprotected boolean saveConsumeLogAndJudgeIdempotent(Channel channel, Message message, String body) {String messageId = (String) message.getMessageProperties().getHeaders().get(MQ_HEADER_REF);//幂等校验,是否重复MqConsume mqConsume = new MqConsume(getConsumeType(),getQueueEnum().getQueueName(),messageId,body);Boolean repeat = mqConsumeService.insert(mqConsume);//重复消费控制if (repeat) {// 手工ackchannel.basicAck(message.getMessageProperties().getDeliveryTag(), false);return false;}return true;}//定义的处理业务代码，由子类实现处理protected abstract void bizHandle(String body);/*** 获取订阅的队列key*/protected abstract ListennedQueueEnum getQueueEnum();/*** 获取消费业务类型*/protected abstract MqConsumeTypeEnums getConsumeType();/*** 定时补偿方法*/public void timedRecompense(String body) {bizHandle(body);}/*** 获取本地定时补偿次数 默认3次*/protected Integer getTimedMaxRetryTimes() {return 3;}/*** 定时补偿业务方法* 格式：springbean的实例+需要执行补偿的方法* @return*/protected String getTimedRecompenseBizMethod() {Component component = AnnotationUtils.findAnnotation(this.getClass(), Component.class);String beanName = StrUtil.isBlank(component.value())? StrUtil.lowerFirst(this.getClass().getSimpleName()) : component.value();return beanName.concat(".").concat("timedRecompense");}
}

2、消息监听和消费过程

• 继承AbstractConsumer 并实现其中的抽象方法

@Slf4j
@Component
public class AssertsConvertOrderCancelConsumer extends AbstractConsumer {@Value("${klzz.internal.config.appId}")private String localAppId;/*** 监听订单取消*/@RabbitListener(queues = queueName.ORDER_CANCEL_QUEUE)public void messageListener(Channel channel, Message message, String body) {//调用父类的doConsume入口进行处理super.doConsume(channel, message, body);}/*** 子类实现的业务处理方法* @param body*/@Overridepublic void bizHandle(String body) {//  TODO 真正的消息处理}@Overrideprotected ListennedQueueEnum getQueueEnum() {return ListennedQueueEnum.ORDER_CANCEL;}@Overrideprotected MqConsumeTypeEnums getConsumeType() {return MqConsumeTypeEnums.ASSETS_CONVERT_ORDER_CANCEL;}
}

3、定义消息的队列配置枚举

@Getter
@AllArgsConstructor
public enum ListennedQueueEnum {ORDER_CANCEL("order_cancel","exchange.trade_order","queue.assets_center.unification.order_cancel"),HOUR_EXCHANGE_REFUND("hour_exchange_refund","exchange.order","queue.assets_center.unification.convert_in_wandou_packet_revert");private  String routeKey;private  String exchangeKey;private  String queueName;}

二、消息消息持久化

• 接收消息持久化，是为了保证MQ消息的高可用，
• 同时在处理MQ消息的过程中一定要确保mq_body的长度能够保存，否则容易出现因为保存数据失败导致MQ消息处理问题

1、消息日志表

• 注意：msg_id设置为唯一索引，进行唯一性校验

CREATE TABLE `mq_consume` (`id` bigint(20) NOT NULL AUTO_INCREMENT COMMENT '自增ID',`type` int(11) NOT NULL DEFAULT '0' COMMENT '消费类型',`msg_id` varchar(64) NOT NULL DEFAULT '' COMMENT '消息id',`title` varchar(32) NOT NULL DEFAULT '' COMMENT '标题',`queue_name` varchar(64) NOT NULL DEFAULT '' COMMENT '队列名称',`mq_body` varchar(3000) NOT NULL DEFAULT '' COMMENT '消息内容',`remark` varchar(256) NOT NULL DEFAULT '' COMMENT '备注',`app_id` varchar(64) NOT NULL DEFAULT '' COMMENT 'appId',`is_deleted` tinyint(1) NOT NULL DEFAULT '1' COMMENT '删除状态 1:正常 2:已删除',`create_time` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',`update_time` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',`version` int(5) NOT NULL DEFAULT '1' COMMENT '版本号',`status` tinyint(1) NOT NULL DEFAULT '1' COMMENT '1:成功,2=失败',PRIMARY KEY (`id`),UNIQUE KEY `unix_msg_id` (`msg_id`) USING BTREE
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='mq消息消费表';

2、消息持久化Bean类

@Data
@EqualsAndHashCode(callSuper = false)
@Accessors(chain = true)
public class MqConsume implements Serializable {private static final long serialVersionUID = 1L;@TableId(value = "id", type = IdType.AUTO)private Long id;private Integer type;private String msgId;private String title;private String queueName;private String mqBody;private String remark;private String appId;private Integer isDeleted; //删除状态 1:正常 2:已删除private LocalDateTime createTime;private LocalDateTime updateTime;private Integer version;private Integer status; // 1:成功,2=失败public MqConsume(MqConsumeTypeEnums consumeType,String queueName, String messageId, String body){this.type = consumeType.getCode();this.title = consumeType.getMessage();this.msgId = messageId;this.mqBody = body;this.queueName = queueName;}
}

3、MQ消息的Mapper

//MQ消息接口
public interface IMqConsumeService extends IService<MqConsume> {Boolean insert(MqConsume mqConsume);
}
//MQ业务实现类
@Slf4j
@Service("mqConsumeService")
public class MqConsumeServiceImpl extends ServiceImpl<MqConsumeMapper, MqConsume> implements IMqConsumeService {@Resourceprivate MqConsumeMapper mqConsumeMapper;@Overridepublic Boolean insert(MqConsume mqConsume) {Boolean repeat=Boolean.FALSE;try {mqConsumeMapper.insert(mqConsume);} catch (DuplicateKeyException e) {repeat = Boolean.TRUE;log.error("重复key插入失败={}",e.getMessage());}return repeat;}public MqConsume getByMessageId(String messageId) {LambdaQueryWrapper<MqConsume> queryWrapper = new QueryWrapper<MqConsume>().lambda().eq(MqConsume::getMsgId, messageId).eq(MqConsume::getIsDeleted, 1);return baseMapper.selectOne(queryWrapper);}
}
//DAO类
public interface MqConsumeMapper extends BaseMapper<MqConsume> {}

三、消息的重试机制

1、消息重试持久化Bean

@Data
@NoArgsConstructor
@EqualsAndHashCode(callSuper = false)
@ApiModel(value="retryMessage对象", description="用户消息重试表")
public class RetryMessage implements Serializable {private static final long serialVersionUID = 1L;private Long id;@ApiModelProperty(value = "消息id")private String messageId;@ApiModelProperty(value = "1-消费失败，2-消费成功，3-重试次数达到上限，4：执行中")private Integer consumerStatus;@ApiModelProperty(value = "交换机名字")private String exchange;@ApiModelProperty(value = "路由key")private String routingKey;@ApiModelProperty(value = "队列名字")private String queue;@ApiModelProperty(value = "消息实体，json格式")private String msgBody;@ApiModelProperty(value = "已重试次数")private Integer retryNum;@ApiModelProperty(value = "最大重试次数")private Integer maxRetryNum;@ApiModelProperty(value = "处理器名称")private String handleName;@ApiModelProperty(value = "错误信息")private String errorMsg;private Date createTime;private Date updateTime;/*** 构造方法*/public RetryMessage(ListennedQueueEnum queueEnum,Integer maxRetryNum,String handleName,String messageId,String body,Exception e){this.messageId = messageId;this.consumerStatus = 1;this.routingKey = queueEnum.getRouteKey();this.exchange = queueEnum.getExchangeKey();this.queue = queueEnum.getQueueName();this.msgBody = body;this.retryNum = 0;this.maxRetryNum = maxRetryNum;this.handleName = handleName;String errorMessage = Optional.ofNullable(e.getMessage()).orElse("");errorMessage = errorMessage.length() > 800 ? errorMessage.substring(0, 800) + "..." : errorMessage;this.errorMsg = errorMessage;this.createTime = new Date();this.updateTime = new Date();}}

2、消息重试Mapper

public interface RetryMessageMapper extends BaseMapper<RetryMessage> {@Select("SELECT * from retry_message WHERE consumer_status = #{consumerStatus} order by id asc limit  #{beginSize}, #{size}")List<RetryMessage> getConsumerList(@Param("consumerStatus") Integer consumerStatus, @Param("beginSize") Integer beginSize, @Param("size") Integer size);}

3、创建消息重试定时任务

@Slf4j
@Component
@JobHandler(value = "retryMessageJobHandler")
public class RetryMessageJobHandler extends IJobHandler {@Resourceprivate RetryMessageMapper retryMessageMapper;@Resourceprivate SpringReflectionUtil springReflectionUtil;@Resourceprivate DingDingHelper dingDingHelper;@Trace@Overridepublic ReturnT<String> execute(String s) throws Exception {String[] split = s.split(",");//起始值Integer beginSize = Integer.valueOf(split[0]);//页大小Integer size = Integer.valueOf(split[1]);List<RetryMessage> retryMessages = retryMessageMapper.getConsumerList(1, beginSize, size);if (!CollectionUtils.isEmpty(retryMessages)) {for (RetryMessage retryMessage : retryMessages) {try {//更新状态,改为执行中，保证幂等UpdateWrapper updateWrapper = new UpdateWrapper<>();updateWrapper.eq("id", retryMessage.getId());updateWrapper.eq("consumer_status", ConsumerSatusEnums.CONSUME_FAIL.getCode());retryMessage.setConsumerStatus(ConsumerSatusEnums.CONSUME_EXEC.getCode());int updateResult = retryMessageMapper.update(retryMessage, updateWrapper);//没更新成功，证明已经在执行中，直接返回if (updateResult < 1) {continue;}//超过最大次数,不处理if (retryMessage.getRetryNum()>=retryMessage.getMaxRetryNum()) {continue;}//处理器为dealUserPayStatusMsg的业务String handleName = retryMessage.getHandleName();//.号分开String[] handleNameSplit = handleName.split("\\.");String serviceName = handleNameSplit[0];String methodName = handleNameSplit[1];Object[] params = {retryMessage.getMsgBody()};springReflectionUtil.springInvokeMethod(serviceName, methodName, params);//成功更新重试次数sucessUpdate(retryMessage);} catch (Exception e) {//失败更新重试次数failUpdate(retryMessage);log.error("RetryMessageJobHandler处理异常={},retryMessage实体= {}", e.getMessage(),JSONUtil.toJsonStr(retryMessage));}}}return ReturnT.SUCCESS;}/*** 失败更新* @param retryMessage*/private void failUpdate(RetryMessage retryMessage) {RetryMessage retryMessageUpadte = new RetryMessage();retryMessageUpadte.setRetryNum(retryMessage.getRetryNum()+1);retryMessageUpadte.setId(retryMessage.getId());//由消费中，改回消费失败retryMessageUpadte.setConsumerStatus(ConsumerSatusEnums.CONSUME_FAIL.getCode());//判断是否达到最大重试次数if (retryMessageUpadte.getRetryNum()>=retryMessage.getMaxRetryNum()) {retryMessageUpadte.setConsumerStatus(ConsumerSatusEnums.CONSUME_MAX_LIMIT.getCode());//钉钉通知人工处理String title = "补偿最大次数用完，转人工干涉处理";String context = "消息实体=" + JSONUtil.toJsonStr(retryMessage);dingDingHelper.sendAlert(title,context);}retryMessageMapper.updateById(retryMessageUpadte);}/*** 成功更新* @param retryMessage*/private void sucessUpdate(RetryMessage retryMessage) {RetryMessage retryMessageUpadte = new RetryMessage();retryMessageUpadte.setRetryNum(retryMessage.getRetryNum()+1);retryMessageUpadte.setId(retryMessage.getId());//处理消费成功retryMessageUpadte.setConsumerStatus(ConsumerSatusEnums.CONSUME_SUCCESS.getCode());retryMessageMapper.updateById(retryMessageUpadte);}}

Spring反射工具类

@Component
public class SpringReflectionUtil {@Resourceprivate ApplicationContext applicationContext;public Object springInvokeMethod(String serviceName, String methodName, Object[] params){Object service = applicationContext.getBean(serviceName);Class<? extends Object>[] paramClass = null;if (params != null) {int paramsLength = params.length;paramClass = new Class[paramsLength];for (int index = 0; index < paramsLength; index++) {paramClass[index] = params[index].getClass();}}// 找到需要执行的方法Method method = ReflectionUtils.findMethod(service.getClass(), methodName, paramClass);// 执行方法return ReflectionUtils.invokeMethod(method, service, params);}}

四、RabbitMQ消息发送

1、MQ消息管理器

@Slf4j
@Component
public class RabbitMQManager {@Autowiredprivate RabbitTemplate rabbitTemplate;public void sendMessage(QueueEnum queueEnum, String jsonStr) {String exchange = queueEnum.getExchange();String routeKey = queueEnum.getRouteKey();String error = "success";long timeMillis = System.currentTimeMillis();try {rabbitTemplate.convertAndSend(exchange, routeKey, jsonStr);} catch (Exception e) {log.error(e.getMessage(), e);error = e.getMessage();} finally {String stringBuilder = "Rabbit MQ 发送消息:\n" +"交换机\t: " + exchange + "\n" +"路由键\t: " + routeKey + "\n" +"消息内容\t: " + jsonStr + "\n" +"错误信息\t: " + error + "\n" +"消耗时间\t: " + (System.currentTimeMillis() - timeMillis) + "ms\n";log.info(stringBuilder);}}/*** 发送消息,用object，cn.vipthink.assets.config.MqConfig，已经做了json的配置，直接传object就行，* 如果传string，等同于序列化了两次，序列化出来的body会多一些转义字符，* MqConfig中有配置消费接收body去除转义，所以我们自身的项目可以正常接收* 但是业务方，没有一般不会配置消费接收body去除转义* @param queueEnum* @param object*/public void sendMessage(QueueEnum queueEnum, Object object) {String exchange = queueEnum.getExchange();String routeKey = queueEnum.getRouteKey();String error = "success";long timeMillis = System.currentTimeMillis();try {rabbitTemplate.convertAndSend(exchange, routeKey, object);} catch (Exception e) {log.error(e.getMessage(), e);error = e.getMessage();} finally {String stringBuilder = "Rabbit MQ 发送消息:\n" +"交换机\t: " + exchange + "\n" +"路由键\t: " + routeKey + "\n" +"消息内容\t: " + JSONUtil.toJsonStr(object) + "\n" +"错误信息\t: " + error + "\n" +"消耗时间\t: " + (System.currentTimeMillis() - timeMillis) + "ms\n";log.info(stringBuilder);}}
}

2、使用消息发送消息

//生命rabbitMQManager
@Resource
private RabbitMQManager rabbitMQManager;//在方法中调用sendMessage实现发送消息rabbitMQManager.sendMessage(QueueEnum.OPEN_HOUR_NOTIFY, messageNotifyRequestDTO.getMessageBody().toString());

3、定制Queue的枚举，注意需要区别于监听的枚举

@Getter
@AllArgsConstructor
public enum QueueEnum {/*** 开包通知mq*/OPEN_HOUR_NOTIFY("exchange.unification_assets", "open_hour_notify","queue.yx-poster.unification_assets.direct.open_hour_notify"),;private final String exchange;private final String routeKey;private final String name;}

五、可靠性消息的投递的通用方案–starter

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-SALisY9a-1692633241334)(…/…/bak/md_img/image-20220316140719460.png)]

1、添加依赖

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><groupId>cn.vipthink.infra</groupId><artifactId>mq-client-starter</artifactId><version>1.1.3</version><properties><maven.compiler.source>8</maven.compiler.source><maven.compiler.target>8</maven.compiler.target><spring.boot.version>2.1.17.RELEASE</spring.boot.version></properties><dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-dependencies</artifactId><version>${spring.boot.version}</version><type>pom</type><scope>import</scope></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-autoconfigure</artifactId><version>2.5.0</version></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId><version>2.5.2</version></dependency><dependency><groupId>org.springframework</groupId><artifactId>spring-jdbc</artifactId><version>5.3.3</version></dependency><dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId><version>1.18.16</version></dependency><dependency><groupId>org.springframework.data</groupId><artifactId>spring-data-redis</artifactId><version>2.1.20.RELEASE</version></dependency><dependency><groupId>com.squareup.okhttp3</groupId><artifactId>okhttp</artifactId><version>3.8.1</version></dependency><dependency><groupId>com.alibaba</groupId><artifactId>fastjson</artifactId><version>1.2.76</version></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-aop</artifactId><version>2.3.12.RELEASE</version></dependency></dependencies><distributionManagement><repository><id>nexus-releases</id><name>Nexus Nexus Repository</name><url>http://nexus-op.vipthink.cn/repository/maven-releases</url></repository><snapshotRepository><id>nexus-snapshots</id><name>local Nexus Repository</name><url>http://nexus-op.vipthink.cn/repository/maven-snapshots</url></snapshotRepository></distributionManagement></project>

2、配置项管理

（1）项目的配置项

spring.rabbitmq.addresses=xxx
spring.rabbitmq.username=xxx
spring.rabbitmq.password=xxx
spring.rabbitmq.port=5672
spring.rabbitmq.publisher-confirms=true
spring.rabbitmq.publisher-confirm-type=correlated

（2）配置项类

@Data
@NoArgsConstructor
@ConfigurationProperties(prefix = "tx.message")
public class AppProperties {private Integer taskInterval = 60;private Integer backoffDelay = 30;private Integer backoffMultiplier = 2;private Integer maxRetry = 3;private Integer querySize = 100;private String dingTalkTitle;private String dingTalkWebhook;
}    

（3）设置配置管理类

@Configuration
@EnableConfigurationProperties({AppProperties.class})
public class AppConfiguration {@Autowiredprivate AppProperties appProperties;public AppConfiguration() { }@Beanpublic MessageService messageService(JdbcTemplate jdbcTemplate) {MessageService messageService = new MessageService();messageService.setJdbcTemplate(jdbcTemplate);return messageService;}@Beanpublic MessageCallback messageCallback(MessageService messageService, RabbitTemplate rabbitTemplate) {MessageCallback messageCallback = new MessageCallback();messageCallback.setMessageService(messageService);messageCallback.setDingTalkUtil(this.dingTalkUtil());rabbitTemplate.setConfirmCallback(messageCallback);return messageCallback;}@Beanpublic MessageSender messageSender(MessageService messageService, RabbitTemplate rabbitTemplate) {MessageSenderImpl messageSenderImpl = new MessageSenderImpl();messageSenderImpl.setMessageService(messageService);messageSenderImpl.setRabbitTemplate(rabbitTemplate);messageSenderImpl.setBackoffDelay(this.appProperties.getBackoffDelay());messageSenderImpl.setBackoffMultiplier(this.appProperties.getBackoffMultiplier());messageSenderImpl.setMaxRetry(this.appProperties.getMaxRetry());return messageSenderImpl;}@Bean(initMethod = "execute")public MessageTask messageTask(MessageService messageService,RabbitTemplate rabbitTemplate, RedisTemplate redisTemplate) {MessageTask messageTask = new MessageTask();messageTask.setMessageService(messageService);messageTask.setRabbitTemplate(rabbitTemplate);messageTask.setRedisTemplate(redisTemplate);messageTask.setTaskInterval(this.appProperties.getTaskInterval());messageTask.setQuerySize(this.appProperties.getQuerySize());return messageTask;}@Beanpublic OkHttpClient okHttpClient() {OkHttpClient client = (new Builder()).build();return client;}//加载DingTalkUtil@Beanpublic DingTalkUtil dingTalkUtil() {DingTalkUtil dingTalkUtil = new DingTalkUtil();dingTalkUtil.setClient(this.okHttpClient());dingTalkUtil.setUrl(this.appProperties.getDingTalkWebhook());dingTalkUtil.setTitle(this.appProperties.getDingTalkTitle());return dingTalkUtil;}
}

（4）在META-INF/spring.factories文件

org.springframework.boot.autoconfigure.EnableAutoConfiguration=cn.vipthink.infra.mq.AppConfiguration

3、持久化消息部分

（1）TxMessageDO持久化类以及SQL

CREATE TABLE `tx_message`(id BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY,message_id BIGINT UNSIGNED NOT NULL COMMENT '消息ID',send_times TINYINT NOT NULL DEFAULT 0 COMMENT '已发送次数',max_retry_times TINYINT NOT NULL DEFAULT 5 COMMENT '最大重试次数',exchange_name VARCHAR(255) NOT NULL COMMENT '交换器名',routing_key VARCHAR(255) COMMENT '路由键',content TEXT COMMENT '消息内容'next_schedule_time DATETIME NOT NULL COMMENT '下一次调度时间',message_status TINYINT NOT NULL DEFAULT 'INIT' COMMENT '消息状态 INIT:处理中 SUCCESS：成功 ，FAIL：失败 ',backoff_delay BIGINT UNSIGNED NOT NULL DEFAULT 10 COMMENT '退避初始化值,单位为秒',backoff_multiplier TINYINT NOT NULL DEFAULT 2 COMMENT '退避因子',create_time DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,update_time DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,deleted TINYINT NOT NULL DEFAULT 0,INDEX idx_create_time (create_time),INDEX idx_next_schedule_time (next_schedule_time),unique INDEX idx_message_id(message_id)
) COMMENT '本地消息表';

@Data
@Accessors(chain=true)
@NoArgsConstructor
public class TxMessageDO {private Long id;private String messageId;private Integer sendTimes;private Integer maxRetryTimes;private String exchangeName;private String routingKey;private String content;private LocalDateTime nextScheduleTime;private String status;private Integer backoffDelay;private Integer backoffMultiplier;private LocalDateTime createTime;private LocalDateTime updateTime;private Integer deleted;
}    

（2）DAO和Service类

public class MessageService {public static final String TABLE_NAME = "tx_message";private JdbcTemplate jdbcTemplate;public MessageService() {}@Autowiredpublic void setJdbcTemplate(JdbcTemplate jdbcTemplate) {this.jdbcTemplate = jdbcTemplate;}//插入到消息表@Transactional( rollbackFor = {Exception.class})public int insert(final TxMessageDO txMessageDO) {String insertSql = "INSERT INTO tx_message(message_id,max_retry_times,exchange_name,routing_key,content,next_schedule_time,backoff_delay,backoff_multiplier) VALUES (?,?,?,?,?,?,?,?)";int i = this.jdbcTemplate.update(insertSql, new PreparedStatementSetter() {public void setValues(PreparedStatement pstmt) throws SQLException {pstmt.setString(1, txMessageDO.getMessageId());pstmt.setInt(2, txMessageDO.getMaxRetryTimes());pstmt.setString(3, txMessageDO.getExchangeName());pstmt.setString(4, txMessageDO.getRoutingKey());pstmt.setString(5, txMessageDO.getContent());pstmt.setObject(6, txMessageDO.getNextScheduleTime());pstmt.setInt(7, txMessageDO.getBackoffDelay());pstmt.setInt(8, txMessageDO.getBackoffMultiplier());}});return i;}//通过消息ID查询TxMessageDOpublic TxMessageDO getByMessageId(String messageId) {String sql = "SELECT id, message_id, send_times, max_retry_times, exchange_name, routing_key, next_schedule_time, status, backoff_delay, backoff_multiplier FROM tx_message WHERE message_id = ?";RowMapper<TxMessageDO> rowMapper = new BeanPropertyRowMapper(TxMessageDO.class);TxMessageDO txMessageDO = (TxMessageDO)this.jdbcTemplate.queryForObject(sql, rowMapper, new Object[]{messageId});return txMessageDO;}//查询所有的失败的TxMessageDOpublic List<TxMessageDO> listFail(String startTime, String endTime, Integer size) {String sql = "SELECT id, message_id, send_times, max_retry_times, exchange_name, routing_key, next_schedule_time, status, backoff_delay, backoff_multiplier,content FROM tx_message WHERE send_times<= max_retry_times  AND status IN( 'INIT','FAIL') AND  next_schedule_time BETWEEN  ? AND ? ORDER BY ID DESC LIMIT ? ";RowMapper<TxMessageDO> rowMapper = new BeanPropertyRowMapper(TxMessageDO.class);List<TxMessageDO> txMessageList = this.jdbcTemplate.query(sql, rowMapper, new Object[]{startTime, endTime, size});return txMessageList;}//更新消息IDpublic int updateById(final TxMessageDO txMessageDO) {String sql = "UPDATE tx_message SET send_times=? ,next_schedule_time=? , status =? WHERE id =? ";int i = this.jdbcTemplate.update(sql, new PreparedStatementSetter() {public void setValues(PreparedStatement pstmt) throws SQLException {pstmt.setInt(1, txMessageDO.getSendTimes());pstmt.setObject(2, txMessageDO.getNextScheduleTime());pstmt.setString(3, txMessageDO.getStatus());pstmt.setLong(4, txMessageDO.getId());}});return i;}
}

（3）消息状态枚举

public enum MessageStatus {INIT,SUCCESS,FAIL;private MessageStatus() {}
}

4、消息回调

public class MessageCallback implements ConfirmCallback {private static final Logger log = LoggerFactory.getLogger(MessageCallback.class);private MessageService messageService;private DingTalkUtil dingTalkUtil;public MessageCallback() {}public void setMessageService(MessageService messageService) {this.messageService = messageService;}public void setDingTalkUtil(DingTalkUtil dingTalkUtil) {this.dingTalkUtil = dingTalkUtil;}//消息确认public void confirm(CorrelationData correlationData, boolean ack, String cause) {if (correlationData != null && StringUtils.hasText(correlationData.getId())) {log.info("MessageId：{},ack：{}，cause：{}", new Object[]{correlationData.getId(), ack, cause});TxMessageDO messageDO = this.messageService.getByMessageId(correlationData.getId());if (messageDO != null) {TxMessageDO txMessageDO = TxMessageDO.builder().build();txMessageDO.setId(messageDO.getId());txMessageDO.setStatus(ack ? MessageStatus.SUCCESS.name() : MessageStatus.FAIL.name());txMessageDO.setSendTimes(messageDO.getSendTimes() + 1);txMessageDO.setNextScheduleTime(messageDO.getNextScheduleTime().plusSeconds((long)(txMessageDO.getSendTimes()* messageDO.getBackoffDelay() * messageDO.getBackoffDelay())));this.messageService.updateById(txMessageDO);if (txMessageDO.getSendTimes() > messageDO.getMaxRetryTimes() && !ack) {this.dingTalkUtil.send("消息ID:" + txMessageDO.getMessageId() + " 达到最大发送次数:" + messageDO.getMaxRetryTimes());}}}}
}

5、消息发送

（1）消息发送接口

public interface MessageSender {//消息发送void send(Message message);//发送延迟消息void sendDelay(Message message, LocalDateTime delayTime);
}

（2）消息发送实现类

public class MessageSenderImpl implements MessageSender {private RabbitTemplate rabbitTemplate;private MessageService messageService;//最大重试次数private Integer maxRetry;//延迟消息private Integer backoffDelay;private Integer backoffMultiplier;public MessageSenderImpl() {}public void setRabbitTemplate(RabbitTemplate rabbitTemplate) {this.rabbitTemplate = rabbitTemplate;}public void setMessageService(MessageService messageService) {this.messageService = messageService;}public void setMaxRetry(Integer maxRetry) {this.maxRetry = maxRetry;}public void setBackoffDelay(Integer backoffDelay) {this.backoffDelay = backoffDelay;}public void setBackoffMultiplier(Integer backoffMultiplier) {this.backoffMultiplier = backoffMultiplier;}//消息发送public void send(final Message message) {this.check(message);TxMessageDO txMessageDO = this.convert(message);this.messageService.insert(txMessageDO);boolean isSynchronizationActive = TransactionSynchronizationManager.isSynchronizationActive();if (!isSynchronizationActive) {this.sendMsg(message);} else {TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronizationAdapter() {public void afterCommit() {MessageSenderImpl.this.sendMsg(message);}});}}//发送消息private void sendMsg(Message message) {CorrelationData correlationData = new CorrelationData();correlationData.setId(message.getMessageId());this.rabbitTemplate.convertAndSend(message.getExchange(), message.getRoutingKey(),message.getMessage(),correlationData);}//发送延迟消息public void sendDelay(Message message, LocalDateTime delayTime) {this.check(message);Assert.notNull(delayTime, "messageId can not be null");TxMessageDO txMessageDO = this.convert(message);txMessageDO.setNextScheduleTime(delayTime);this.messageService.insert(txMessageDO);}//消息转换成DOprivate TxMessageDO convert(Message message) {TxMessageDO txMessageDO = TxMessageDO.builder().build().setMessageId(message.getMessageId()).setContent(message.getMessage()).setExchangeName(message.getExchange()).setRoutingKey(message.getRoutingKey()).setBackoffDelay(this.backoffDelay).setBackoffMultiplier(this.backoffMultiplier).setMaxRetryTimes(this.maxRetry).setNextScheduleTime(LocalDateTime.now());return txMessageDO;}//检查private void check(Message message) {Assert.notNull(message, "message can not be null");Assert.notNull(message.getMessageId(), "messageId can not be null");Assert.notNull(message.getMessage(), "message can not be null");Assert.notNull(message.getExchange(), "exchange can not be null");Assert.notNull(message.getRoutingKey(), "routingKey can not be null");}
}

6、消息监听AOP类

@Aspect
public class MessageListnerAspct {public MessageListnerAspct() {}@Pointcut("execution(* cn.vipthink.orderfeign.*Feign*.*(..)) ")public void oldOrderFeignAspect() {}@Around("oldOrderFeignAspect()")public Object oldOrderFeignAspect(ProceedingJoinPoint proceedingJoinPoint) throws Throwable {//获取渠道ObjObject channelObj = Arrays.stream(proceedingJoinPoint.getArgs()).filter((s) -> {return s instanceof Channel;}).findAny().orElse((Object)null);//获取消息ObjectObject messageObj = Arrays.stream(proceedingJoinPoint.getArgs()).filter((s) -> {return s instanceof Message;}).findAny().orElse((Object)null);if (channelObj != null) {Channel var4 = (Channel)channelObj;}if (channelObj != null && messageObj != null) {Message message = (Message)messageObj;String envIdHeader = (String)message.getMessageProperties().getHeader("envId");String envId = (String)System.getenv().get("envId");if (StringUtils.hasText(envIdHeader) && envIdHeader.equals(envId)) {return proceedingJoinPoint.proceed();} else if (message.getMessageProperties().getRedelivered()) {return proceedingJoinPoint.proceed();} else {Channel channel = (Channel)channelObj;channel.basicRecover(true);return null;}} else {return proceedingJoinPoint.proceed();}}
}

7、引入将starter打包成为POM

（1）引入POM包

<dependency><groupId>cn.vipthink.infra</groupId><artifactId>mq-client-starter</artifactId><version>1.1.3</version>
</dependency>

（2）添加配置

spring.rabbitmq.addresses=xxx
spring.rabbitmq.username=xxx
spring.rabbitmq.password=xxx
spring.rabbitmq.port=5672
spring.rabbitmq.publisher-confirms=true
spring.rabbitmq.publisher-confirm-type=correlated

（3）使用MessageSender发送

@Autowired
private  MessageSender messageSender;Message message = new Message();
messageSender.send(message)

代码参考：https://www.iocoder.cn/Spring-Boot/RabbitMQ/?github