进阶

Mr.Xmln2024年11月17日大约 6 分钟

消息 - Confirm

消息的Confirm其实就是一个确认机制，指的是生产者投递消息之后，消息到达了broker内的交换机上则会给生产者一个应答，生产者根据应答判断消息是否正常的发送到了broker内的交换机上，这确保消息可靠性投递的保障

yml配置：

spring:
  rabbitmq:
    publisher-confirm-type: correlated # 开启生产者确认机制

# 发布者类型确认使用  - publisher-confirm-type 值
# public enum ConfirmType {
#     在范围操作内使用RabbitTemplate#waitForConfirms() （或waitForConfirmsOrDie() 。
#     SIMPLE,
# 	  与CorrelationData一起使用以将确认与已发送的消息关联起来。
#     CORRELATED,
# 	  发布者确认被禁用（默认）
#     NONE
# }

写一个类实现 RabbitTemplate.ConfirmCallback 接口，判断成功失败 ack 结果，如果 ack 为 false 那么对消息进行重新投递或者其他操作

@Slf4j
@Component
public class MyConfirmCallBack implements RabbitTemplate.ConfirmCallback {
    /**
     * 回调方法，确认消息成功或者失败
     *
     * @param correlationData 回调的相关数据。
     * @param ack             true 表示 ack， false 表示 nack
     * @param cause           可选原因，适用于 nack（如果可用），否则为 null。
     */
    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        if (ack) {
            log.info("消息发送成功：{}", correlationData.getId());
            return;
        }
        // 执行消息发送失败的逻辑
        log.info("消息发送失败：{}", cause);
    }
}

在发送消息页面进行配置关联数据和初始化内容

@Resource
private RabbitTemplate rabbitTemplate;
@Resource
private MyConfirmCallBack confirmCallBack;

@PostConstruct  // 初始化
public void init() {
    rabbitTemplate.setConfirmCallback(confirmCallBack);
}

@GetMapping("/message") // 发送消息
public void sendMessage(String msg) {
    {
        Message message = MessageBuilder.withBody(msg.getBytes(StandardCharsets.UTF_8)).build();
        CorrelationData correlationData = new CorrelationData();  // 关联数据
        // 这个setId就是设置唯一的id，因为发送的消息可能一样，那就需要设置一个唯一id来跟踪提供者发送的消息
        correlationData.setId("order_1");
        // 采用直连交换机
        rabbitTemplate.convertAndSend(RabbitConfig.NORMAL_EXCHANGE_NAME, RabbitConfig.NORMAL_BINDING_NAME, message, correlationData);
        log.info("发送消息成功：{}", msg);
    }
}

除了单独创建一个类实现接口之外，还可以有以下的方式：

使用匿名内部类
让当前类实现接口
使用Lambda表达式（因为是函数时接口）

消息 - Return

RabbitMQ消息的传递流程为：producer —> exchange —> queue —> consumer

消息从 producer —> exchange 会返回一个 ConfirmCallback

消息从 exchange —> queue 投递失败会返回一个 ReturnCallback

可以通过这两个 callback 控制消息的可靠传递

开启配置：

spring:
  rabbitmq:
    publisher-returns: true # 开启生产者返回机制

实现接口

@Slf4j
@Component
public class MyReturnCollBack implements RabbitTemplate.ReturnsCallback {
    @Override // 该接口个上面的接口不同的是该接口是在发送消息失败的时候会调用
    public void returnedMessage(ReturnedMessage returned) {
        log.error("消息发送失败 - 交换机：{}", returned.getExchange());
        log.error("消息发送失败 - 路由KEY：{}", returned.getRoutingKey());
        log.error("消息发送失败 - 原因：{}", returned.getMessage());
    }
}

配置类

@Resource
private RabbitTemplate rabbitTemplate;
@Resource
private MyReturnCollBack returnCollBack;

@PostConstruct  // 初始化
public void init() {
    rabbitTemplate.setReturnsCallback(returnCollBack); // 设置回调函数
}

@GetMapping("/message") // 发送消息
public void sendMessage(String msg) {
    {
        Message message = MessageBuilder.withBody(msg.getBytes(StandardCharsets.UTF_8)).build();
        CorrelationData correlationData = new CorrelationData();  // 关联数据
        correlationData.setId("order_1");
        rabbitTemplate.convertAndSend(RabbitConfig.NORMAL_EXCHANGE_NAME, RabbitConfig.NORMAL_BINDING_NAME + "1", message, correlationData);
        log.info("发送消息成功：{}", msg);
    }
}

备用交换机

当消息发送失败时会走备用交换机，它和死信队列有点相像，不同的是死信队列主要用于处理消费端的问题，比如消息处理失败；而备用交换机则用于处理生产端的问题，比如消息无法被预期的交换机路由

配置：

public static final String NORMAL_EXCHANGE_NAME = "exchange.order"; // 正常交换机
public static final String NORMAL_QUEUE_NAME = "queue.order"; // 正常队列
public static final String NORMAL_BINDING_NAME = "order"; // 正常绑定
public static final String SPARE_EXCHANGE_NAME = NORMAL_EXCHANGE_NAME + ".spare"; // 备用交换机
public static final String SPARE_QUEUE_NAME = NORMAL_QUEUE_NAME + ".spare"; // 备用队列

@Bean   // 正常交换机
public DirectExchange directExchange() {
    return ExchangeBuilder
            .directExchange(NORMAL_EXCHANGE_NAME)
            .alternate(SPARE_EXCHANGE_NAME)  // 绑定备用交换机
            .build();
}

@Bean   // 正常队列
public Queue orderQueue() {
    return QueueBuilder.durable(NORMAL_QUEUE_NAME).build();
}

@Bean   // 正常绑定
public Binding orderBinding(DirectExchange directExchange, Queue orderQueue) {
    return BindingBuilder.bind(orderQueue).to(directExchange).with(NORMAL_BINDING_NAME);
}

@Bean   // 备用交换机
public FanoutExchange spareExchange() {
    return ExchangeBuilder.fanoutExchange(SPARE_EXCHANGE_NAME).build();
}

@Bean   // 备用队列
public Queue spareQueue() {
    return QueueBuilder.durable(SPARE_QUEUE_NAME).build();
}

@Bean   // 备用队列绑定
public Binding spareBinding(FanoutExchange spareExchange, Queue spareQueue) {
    return BindingBuilder.bind(spareQueue).to(spareExchange);
}