/**
 * 當(dāng)消息沒有到達Rabbitmq的交換機時觸發(fā)該方法（當(dāng)然到達了也會觸發(fā)，）
 */
@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) {

        //第一個坑，如果發(fā)送端發(fā)送消息時沒有對correlationData進行處理，conirm方法接收到的對象都會是null
        //當(dāng)接收失敗并且correlationData對象為null，證明目前已經(jīng)無法追溯回業(yè)務(wù)，可以做業(yè)務(wù)日志處理
        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()));
            //這里可以持久化業(yè)務(wù)消息體到數(shù)據(jù)庫，然后定時去進行補償處理或者重試等等
            return;
        }

        //處理完成

    }
}

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

    TestMessage testMessage=new TestMessage();
    testMessage.setName("mq名稱");
    testMessage.setBusinessId("業(yè)務(wù)Id");

    //定義CorrelationData對象以及消息屬性。不然comfirm方法無論失敗還是成功，CorrelationData參數(shù)永遠是null
    CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
    //傳遞業(yè)務(wù)數(shù)據(jù)
    correlationData.setReturnedMessage(new Message(JSONObject.toJSON(testMessage).toString().getBytes(StandardCharsets.UTF_8),new MessageProperties()));

  //發(fā)送消息（這里發(fā)送給了message.log.test.exchange11交換機，但實際rabbitmq并不存在）template.convertAndSend("message.log.test.exchange11","message_loss_test",testMessage,correlationData);

    return true;
}

/**
 * 當(dāng)消息沒有到達Rabbitmq的隊列時就會觸發(fā)該方法
 */
@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 發(fā)送方定義的路由key
     */
    @Override
    public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
        System.out.println("消息標(biāo)識：" + 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("業(yè)務(wù)Id");

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

    return true;
}

rabbitmq:
 listener:
  simple:
    #開啟手動確認(rèn)
    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());
    //故意觸發(fā)異常
    if(!StringUtils.isEmpty(testmessage.getName())){

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