/**
 * 當(dāng)消息沒有到達(dá)Rabbitmq的交換機(jī)時(shí)觸發(fā)該方法（當(dāng)然到達(dá)了也會(huì)觸發(fā)，）
 */
@Component
public class ConfirmCallBack implements RabbitTemplate.ConfirmCallback {

    @Resource
    private RabbitTemplate rabbitTemplate;

    @PostConstruct
    public void init(){

        rabbitTemplate.setConfirmCallback(this);
    }

    /**
     *
     * @param correlationData 消息屬性體
     * @param ack 是否成功，成功到達(dá)true，沒有到達(dá)，false
     * @param cause rabbitmq自身給的信息
     */
    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {

        //第一個(gè)坑，如果發(fā)送端發(fā)送消息時(shí)沒有對(duì)correlationData進(jìn)行處理，conirm方法接收到的對(duì)象都會(huì)是null
        //當(dāng)接收失敗并且correlationData對(duì)象為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ù)庫，然后定時(shí)去進(jìn)行補(bǔ)償處理或者重試等等
            return;
        }

        //處理完成

    }
}

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

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

    //定義CorrelationData對(duì)象以及消息屬性。不然comfirm方法無論失敗還是成功，CorrelationData參數(shù)永遠(yuǎn)是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交換機(jī)，但實(shí)際rabbitmq并不存在）template.convertAndSend("message.log.test.exchange11","message_loss_test",testMessage,correlationData);

    return true;
}

/**
 * 當(dāng)消息沒有到達(dá)Rabbitmq的隊(duì)列時(shí)就會(huì)觸發(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   交換機(jī)
     * @param routingKey 發(fā)送方定義的路由key
     */
    @Override
    public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
        System.out.println("消息標(biāo)識(shí)：" + 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:
    #開啟手動(dòng)確認(rèn)
    acknowledge-mode: manual
    #開啟失敗后的重試機(jī)制
    retry:
      enabled: true
      #最多重試3次
      max-attempts: 3

@Component
public class TestConsumer {

    /**
     * 消費(fèi)
     * @param testmessage 消息體
     * @param message 消息屬性
     * @param channel mq通道對(duì)象
     */
    @RabbitListener(queues = {"message.loss.test.queue"})
    public void test(TestMessage testmessage, Message message, Channel channel) throws IOException {
        System.out.println("消費(fèi)testmessage消息："+testmessage.getName());
//        channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);

    }
}

/**
 * 消費(fèi)
 * @param testmessage 消息體
 * @param message 消息屬性
 * @param channel mq通道對(duì)象
 */
@RabbitListener(queues = {"message.loss.test.queue"})
public void test(TestMessage testmessage, Message message, Channel channel) throws IOException {
    System.out.println("消費(fèi)testmessage消息："+testmessage.getName());
    //故意觸發(fā)異常
    if(!StringUtils.isEmpty(testmessage.getName())){

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