标签：异步 confirm publisher 通信 rabbitmq 发送消息 import correlationData

服务异步通信-高级篇

0.RabbitMQ的基础知识回顾

消息队列在使用过程中，面临着很多实际问题需要思考：

1.消息可靠性

消息从发送，到消费者接收，会经理多个过程：

其中的每一步都可能导致消息丢失，常见的丢失原因包括：

发送时丢失：
- 生产者发送的消息未送达exchange
- 消息到达exchange后未到达queue
MQ宕机，queue将消息丢失
consumer接收到消息后未消费就宕机

针对这些问题，RabbitMQ分别给出了解决方案：

生产者确认机制
mq持久化
消费者确认机制
失败重试机制

下面我们就通过案例来演示每一个步骤。

首先，导入demo工程：

项目结构如下：

1.1.生产者消息确认

RabbitMQ提供了publisher confirm机制来避免消息发送到MQ过程中丢失。这种机制必须给每个消息指定一个唯一ID。消息发送到MQ以后，会返回一个结果给发送者，表示消息是否处理成功。

返回结果有两种方式：

publisher-confirm，发送者确认
- 消息成功投递到交换机，返回ack
- 消息未投递到交换机，返回nack
publisher-return，发送者回执

消息投递到交换机了，但是没有路由到队列。返回ACK，及路由失败原因。

注意：

1.1.1.修改配置

首先，修改publisher服务中的application.yml文件，添加下面的内容：

spring:
  rabbitmq:
    publisher-confirm-type: correlated
    publisher-returns: true
    template:
      mandatory: true

说明：

publish-confirm-type：开启publisher-confirm，这里支持两种类型：
- simple：同步等待confirm结果，直到超时
- correlated：异步回调，定义ConfirmCallback，MQ返回结果时会回调这个ConfirmCallback
publish-returns：开启publish-return功能，同样是基于callback机制，不过是定义ReturnCallback
template.mandatory：定义消息路由失败时的策略。true，则调用ReturnCallback；false：则直接丢弃消息

1.1.2.定义Return回调

每个RabbitTemplate只能配置一个ReturnCallback，因此需要在项目加载时配置：

修改publisher服务，添加一个：

 1 package cn.itcast.mq.config;
 2 
 3 import lombok.extern.slf4j.Slf4j;
 4 import org.springframework.amqp.rabbit.core.RabbitTemplate;
 5 import org.springframework.beans.BeansException;
 6 import org.springframework.context.ApplicationContext;
 7 import org.springframework.context.ApplicationContextAware;
 8 import org.springframework.context.annotation.Configuration;
 9 
10 @Slf4j
11 @Configuration
12 public class CommonConfig implements ApplicationContextAware {
13     @Override
14     public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
15         // 获取RabbitTemplate
16         RabbitTemplate rabbitTemplate = applicationContext.getBean(RabbitTemplate.class);
17         // 设置ReturnCallback
18         rabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey) -> {
19             // 投递失败，记录日志
20             log.info("消息发送失败，应答码{}，原因{}，交换机{}，路由键{},消息{}",
21                      replyCode, replyText, exchange, routingKey, message.toString());
22             // 如果有业务需要，可以重发消息
23         });
24     }
25 }

1.1.3.定义ConfirmCallback

ConfirmCallback可以在发送消息时指定，因为每个业务处理confirm成功或失败的逻辑不一定相同。

在publisher服务的cn.itcast.mq.spring.SpringAmqpTest类中，定义一个单元测试方法：

public void testSendMessage2SimpleQueue() throws InterruptedException {
    // 1.消息体
    String message = "hello, spring amqp!";
    // 2.全局唯一的消息ID，需要封装到CorrelationData中
    CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
    // 3.添加callback
    correlationData.getFuture().addCallback(
        result -> {
            if(result.isAck()){
                // 3.1.ack，消息成功
                log.debug("消息发送成功, ID:{}", correlationData.getId());
            }else{
                // 3.2.nack，消息失败
                log.error("消息发送失败, ID:{}, 原因{}",correlationData.getId(), result.getReason());
            }
        },
        ex -> log.error("消息发送异常, ID:{}, 原因{}",correlationData.getId(),ex.getMessage())
    );
    // 4.发送消息
    rabbitTemplate.convertAndSend("task.direct", "task", message, correlationData);

    // 休眠一会儿，等待ack回执
    Thread.sleep(2000);
}

快速把内部类中的唯一方法，变成lambda的快捷键是 Alt+Enter

1.1.4.测试

正常发送到交换机的情况，执行了confirm中的success方法

没有发送到交换机的情况，执行了confirm中的fail方法

标签：异步,confirm,publisher,通信,rabbitmq,发送,消息,import,correlationData
From： https://www.cnblogs.com/Zz198/p/16717908.html

服务异步通信-（rabbitmq）高级篇