RabbitMQ之消息可靠性投递解读

  @Component
    public class MessageConfirmCallBack implements RabbitTemplate.ConfirmCallback {
        /**
         * 交换机收到消息后，会回调该方法
         *
         * @param correlationData  相关联的数据
         * @param ack  有两个取值，true和false，true表示成功：消息正确地到达交换机，反之false就是消息没有正确地到达交换机
         * @param cause 消息没有正确地到达交换机的原因是什么
         */
        @Override
        public void confirm(CorrelationData correlationData, boolean ack, String cause) {
            System.out.println("correlationData = " + correlationData);
            System.out.println("ack = " + ack);
            System.out.println("cause = " + cause);

            if (ack) {
                //正常
            } else {
                //不正常的，可能需要记日志或重新发送
            }
        }
    }

@Service
public class MessageService {

    @Resource
    private RabbitTemplate rabbitTemplate;

    @Resource
    private MessageConfirmCallBack messageConfirmCallBack;

    @PostConstruct //bean在初始化的时候，会调用一次该方法，只调用一次，起到初始化的作用
    public void init() {
        rabbitTemplate.setConfirmCallback(messageConfirmCallBack);
    }

    /**
     * 发送消息
     */
    public void sendMessage() {
        //关联数据对象
        CorrelationData correlationData = new CorrelationData();
        correlationData.setId("O159899323"); //比如设置一个订单ID，到时候在confirm回调里面，你就可以知道是哪个订单没有发送到交换机上去
        rabbitTemplate.convertAndSend(RabbitConfig.EXCHANGE + 123, "info", "hello", correlationData);
        System.out.println("消息发送完毕......");
    }
}

channel.txSelect(); // 将当前channel设置成事务模式

/**
  * ConfirmConfig.exchangeName(交换机名称)
  * ConfirmConfig.routingKey(路由键)
  * message （消息内容）
  */
channel.basicPublish(ConfirmConfig.exchangeName, ConfirmConfig.routingKey, MessageProperties.PERSISTENT_TEXT_PLAIN, message));
channel.txCommit(); // 提交事务
channel.txRollback(); // 回滚事务

@Component
public class MessageReturnCallBack implements RabbitTemplate.ReturnsCallback {

    /**
     * 当消息从交换机 没有正确地 到达队列，则会触发该方法
     * 如果消息从交换机 正确地 到达队列了，那么就不会触发该方法
     *
     * @param returned
     */
    @Override
    public void returnedMessage(ReturnedMessage returned) {
        System.out.println("消息return模式：" + returned);
    }
}

@Service
public class MessageService {

    @Resource
    private RabbitTemplate rabbitTemplate;

    @Resource
    private MessageReturnCallBack messageReturnCallBack;

    @PostConstruct //bean在初始化的时候，会调用一次该方法，只调用一次，起到初始化的作用
    public void init() {
        rabbitTemplate.setReturnsCallback(messageReturnCallBack);
    }

    /**
     * 发送消息
     */
    public void sendMessage() {
        rabbitTemplate.convertAndSend(RabbitConfig.EXCHANGE, "info123", "hello");
        System.out.println("消息发送完毕......");
    }
}

Map<String, Object> arguments = new HashMap<>();
//指定当前正常的交换机的备用交换机是谁
arguments.put("alternate-exchange", EXCHANGE_ALTERNATE); 
//DirectExchange(String name, boolean durable, boolean autoDelete, Map<String, Object> arguments)
return new DirectExchange(EXCHANGE, true, false, arguments);
//return ExchangeBuilder.directExchange(EXCHANGE).withArguments(args).build();

QueueBuilder.durable(QUEUE).build();

ExchangeBuilder.directExchange(EXCHANGE).durable(true).build();

 MessageProperties messageProperties = new MessageProperties();
//设置消息持久化，当然它默认就是持久化，所以可以不用设置，可以查看源码
 messageProperties.setDeliveryMode(MessageDeliveryMode.PERSISTENT); 

#开启手动ack消息消费确认

spring.rabbitmq.listener.simple.acknowledge-mode=manual