Spring boot Rabbitmq消息防丢失实践

/**
 * 当消息没有到达Rabbitmq的交换机时触发该方法（当然到达了也会触发，）
 */
@Component
public class ConfirmCallBack implements RabbitTemplate.ConfirmCallback {

    @Resource
    private RabbitTemplate rabbitTemplate;

    @PostConstruct
    public void init(){

        rabbitTemplate.setConfirmCallback(this);
    }

    /**
     *
     * @param correlationData 消息属性体
     * @param ack 是否成功，成功到达true，没有到达，false
     * @param cause rabbitmq自身给的信息
     */
    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {

        //第一个坑，如果发送端发送消息时没有对correlationData进行处理，conirm方法接收到的对象都会是null
        //当接收失败并且correlationData对象为null，证明目前已经无法追溯回业务，可以做业务日志处理
        if(!ack&&correlationData==null){
            System.out.println(cause);
            //日志处理。。。

            return;
        }
        //如果接收失败
        if(!ack){
            System.out.println("消息Id:"+correlationData.getId());
            Message message=correlationData.getReturnedMessage();
            System.out.println("消息体："+new String(message.getBody()));
            //这里可以持久化业务消息体到数据库，然后定时去进行补偿处理或者重试等等
            return;
        }

        //处理完成

    }
}

/**
 * 消息的推送
 * @return
 */
@PostMapping("push")
public boolean push(){

    TestMessage testMessage=new TestMessage();
    testMessage.setName("mq名称");
    testMessage.setBusinessId("业务Id");

    //定义CorrelationData对象以及消息属性。不然comfirm方法无论失败还是成功，CorrelationData参数永远是null
    CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
    //传递业务数据
    correlationData.setReturnedMessage(new Message(JSONObject.toJSON(testMessage).toString().getBytes(StandardCharsets.UTF_8),new MessageProperties()));

  //发送消息（这里发送给了message.log.test.exchange11交换机，但实际rabbitmq并不存在）template.convertAndSend("message.log.test.exchange11","message_loss_test",testMessage,correlationData);

    return true;
}

/**
 * 当消息没有到达Rabbitmq的队列时就会触发该方法
 */
@Component
public class ReturnCallBack implements RabbitTemplate.ReturnCallback {
    @Resource
    private RabbitTemplate rabbitTemplate;
    @PostConstruct
    public void init() {
        rabbitTemplate.setReturnCallback(this);
    }
    /**
     * @param message    消息体
     * @param replyCode  返回代码
     * @param replyText  返回文本
     * @param exchange   交换机
     * @param routingKey 发送方定义的路由key
     */
    @Override
    public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
        System.out.println("消息标识：" + message.getMessageProperties().getDeliveryTag());
        String messageBody = null;
        try {
            messageBody = new String(message.getBody(), "UTF-8");
        } catch (UnsupportedEncodingException e) {
            e.printStackTrace();
        }
        System.out.println("消息：" + messageBody);
        System.out.println(replyCode);
        System.out.println(replyText);
        System.out.println(exchange);
        System.out.println(routingKey);

    }
}

/**
 * 消息的推送
 * @return
 */
@PostMapping("push2")
public boolean push2(){

    TestMessage testMessage=new TestMessage();
    testMessage.setName("mq名称2");
    testMessage.setBusinessId("业务Id");

    template.convertAndSend("message.log.test2.exchange","message_loss_test",JSONObject.toJSON(testMessage).toString());

    return true;
}

rabbitmq:
 listener:
  simple:
    #开启手动确认
    acknowledge-mode: manual
    #开启失败后的重试机制
    retry:
      enabled: true
      #最多重试3次
      max-attempts: 3

@Component
public class TestConsumer {

    /**
     * 消费
     * @param testmessage 消息体
     * @param message 消息属性
     * @param channel mq通道对象
     */
    @RabbitListener(queues = {"message.loss.test.queue"})
    public void test(TestMessage testmessage, Message message, Channel channel) throws IOException {
        System.out.println("消费testmessage消息："+testmessage.getName());
//        channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);

    }
}

/**
 * 消费
 * @param testmessage 消息体
 * @param message 消息属性
 * @param channel mq通道对象
 */
@RabbitListener(queues = {"message.loss.test.queue"})
public void test(TestMessage testmessage, Message message, Channel channel) throws IOException {
    System.out.println("消费testmessage消息："+testmessage.getName());
    //故意触发异常
    if(!StringUtils.isEmpty(testmessage.getName())){

        throw new RuntimeException("11211");
    }
    channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
}