package cn.linkpower.config;

import org.springframework.amqp.core.*;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
/**
 * 死信隊(duì)列配置
 */
@Configuration
public class DeadMsgMqConfig {
    // 定義正常交換機(jī)和正常隊(duì)列信息(交換機(jī)名、隊(duì)列名、路由key)
    public static final String queue_name = "xj_natural_queue";
    public static final String exchange_name = "xj_natural_exchange";
    public static final String routing_key = "xj_natural_routingKey";
    // 定義死信交換機(jī)名、死信隊(duì)列名、路由key
    public static final String queue_name_dead = "xj_dead_queue";
    public static final String exchange_name_dead = "xj_dead_exchange";
    public static final String routing_key_dead = "xj_dead_routingKey";
    /**
     * 設(shè)置正常的消息隊(duì)列；
     * 正常的消息隊(duì)列具備以下幾種功能：
     * 1、消息正常消費(fèi)，需要綁定對(duì)應(yīng)的消費(fèi)者(這里為了測(cè)試死信，不創(chuàng)建消費(fèi)者)
     * 2、當(dāng)消息失效后，需要將指定的消息發(fā)送至 死信交換機(jī) 中
     * @return
     */
    @Bean(value = "getNaturalQueue")
    public Queue getNaturalQueue(){
        return QueueBuilder.durable(queue_name)
                // 正常的隊(duì)列，在消息失效后，需要將消息丟入 死信 交換機(jī)中
                // 這里只需要針對(duì)名稱進(jìn)行綁定
                .withArgument("x-dead-letter-exchange",exchange_name_dead)
                // 丟入 死信交換機(jī)，需要設(shè)定指定的 routingkey
                .withArgument("x-dead-letter-routing-key",routing_key_dead)
                // 設(shè)置正常隊(duì)列中消息的存活時(shí)間為 10s，當(dāng)然也可以針對(duì)單個(gè)消息進(jìn)行設(shè)定不同的過(guò)期時(shí)間
                .withArgument("x-message-ttl",10000)
                // 設(shè)定當(dāng)前隊(duì)列中，允許存放的最大消息數(shù)目
                .withArgument("x-max-length",10)
                .build();
    }
     * 設(shè)定正常的消息交換機(jī)
    @Bean(value = "getNaturalExchange")
    public Exchange getNaturalExchange(){
        // 這里為了測(cè)試，采取 direct exchange
        return ExchangeBuilder.directExchange(exchange_name)
                .durable(true) // 設(shè)定持久化
     * 將正常的消息交換機(jī)和正常的消息隊(duì)列進(jìn)行綁定
     * @param queue
     * @param directExchange
    @Bean
    public Binding bindNaturalExchangeAndQueue(
            @Qualifier(value = "getNaturalQueue") Queue queue,
            @Qualifier(value = "getNaturalExchange") Exchange directExchange
    ){
        return BindingBuilder
                // 綁定消息隊(duì)列
                .bind(queue)
                // 至指定的消息交換機(jī)
                .to(directExchange)
                // 匹配 routingkey
                .with(routing_key)
                // 無(wú)參數(shù)，不加會(huì)報(bào)錯(cuò)提示
                .noargs();
     * 定義死信隊(duì)列
    @Bean(value = "getDealQueue")
    public Queue getDealQueue(){
        return QueueBuilder.durable(queue_name_dead).build();
     * 定義死信交換機(jī)
    @Bean(value = "getDeadExchange")
    public Exchange getDeadExchange(){
        return ExchangeBuilder.directExchange(exchange_name_dead).durable(true).build();
     * 將死信交換機(jī)和死信隊(duì)列進(jìn)行綁定
     * @param deadQueue
     * @param directDeadExchange
    public Binding bindDeadExchangeAndQueue(
            @Qualifier(value = "getDealQueue") Queue deadQueue,
            @Qualifier(value = "getDeadExchange") Exchange directDeadExchange
        return BindingBuilder.bind(deadQueue).to(directDeadExchange).with(routing_key_dead).noargs();
}

package cn.linkpower.service;

import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
@Slf4j
@Component
public class RabbitmqService implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnCallback {
    @Autowired
    private RabbitTemplate rabbitTemplate;
    /**
     * 直接發(fā)送消息
     * @param exchange
     * @param routingKey
     * @param msg
     */
    public void sendMessage(String exchange,String routingKey,Object msg) {
        // 設(shè)置交換機(jī)處理失敗消息的模式     true 表示消息由交換機(jī) 到達(dá)不了隊(duì)列時(shí)，會(huì)將消息重新返回給生產(chǎn)者
        // 如果不設(shè)置這個(gè)指令，則交換機(jī)向隊(duì)列推送消息失敗后，不會(huì)觸發(fā) setReturnCallback
        rabbitTemplate.setMandatory(true);
        //消息消費(fèi)者確認(rèn)收到消息后，手動(dòng)ack回執(zhí)
        rabbitTemplate.setConfirmCallback(this);
        // return 配置
        rabbitTemplate.setReturnCallback(this);
        //發(fā)送消息
        rabbitTemplate.convertAndSend(exchange,routingKey,msg);
    }
     * 交換機(jī)并未將數(shù)據(jù)丟入指定的隊(duì)列中時(shí)，觸發(fā)
     *  channel.basicPublish(exchange_name,next.getKey(), true, properties,next.getValue().getBytes());
     *  參數(shù)三：true  表示如果消息無(wú)法正常投遞，則return給生產(chǎn)者 ；false 表示直接丟棄
     * @param message   消息對(duì)象
     * @param replyCode 錯(cuò)誤碼
     * @param replyText 錯(cuò)誤信息
     * @param exchange 交換機(jī)
     * @param routingKey 路由鍵
    @Override
    public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
        log.info("---- returnedMessage ----replyCode="+replyCode+" replyText="+replyText+" ");
     * 消息生產(chǎn)者發(fā)送消息至交換機(jī)時(shí)觸發(fā)，用于判斷交換機(jī)是否成功收到消息
     * @param correlationData  相關(guān)配置信息
     * @param ack exchange 交換機(jī)，判斷交換機(jī)是否成功收到消息    true 表示交換機(jī)收到
     * @param cause  失敗原因
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        log.info("---- confirm ----ack="+ack+"  cause="+String.valueOf(cause));
        log.info("correlationData -->"+correlationData.toString());
        if(ack){
            // 交換機(jī)接收到
            log.info("---- confirm ----ack==true  cause="+cause);
        }else{
            // 沒(méi)有接收到
            log.info("---- confirm ----ack==false  cause="+cause);
        }
}

package cn.linkpower.controller;

import cn.linkpower.config.DeadMsgMqConfig;
import cn.linkpower.service.RabbitmqService;
import lombok.extern.slf4j.Slf4j;
import org.apache.tomcat.jni.Time;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import java.util.concurrent.TimeUnit;
@Slf4j
@RestController
public class DeadMsgController {
    @Autowired
    private RabbitmqService rabbitmqService;
    @RequestMapping("/deadMsgTest")
    public String deadMsgTest() throws InterruptedException {
        // 向正常的消息隊(duì)列中丟數(shù)據(jù)，測(cè)試限定時(shí)間未消費(fèi)后，死信隊(duì)列的情況
        // 配置文件中，針對(duì)于正常隊(duì)列而言，設(shè)置有10條上限大小
        for (int i = 0; i < 20; i++) {
            String msg = "dead msg test "+i;
            log.info("發(fā)送消息，消息信息為：{}",msg);
            // 向正常的消息交換機(jī)中傳遞數(shù)據(jù)
            rabbitmqService.sendMessage(DeadMsgMqConfig.exchange_name,DeadMsgMqConfig.routing_key,msg);
            TimeUnit.SECONDS.sleep(2);
        }
        return "ok";
    }
}

channel.basicNack(message.getMessageProperties().getDeliveryTag(),false,false);