@Configuration
public class DirectRabbitConfig {
    public static final String DEAD_LETTER_EXCHANGE = "dead.latter.exchange";
    public static final String DEAD_LETTER_QUEUE = "dead.latter.queue";
    public static final String DEAD_LETTER_KEY = "dead.latter.key";
    //隊(duì)列 起名：TestDirectQueue
    @Bean
    public Queue TestDirectQueue() {
        //實(shí)例化隊(duì)列時(shí)各個(gè)參數(shù)的含義如下：
        //name:隊(duì)列名稱
        // durable:是否持久化,默認(rèn)是false,持久化隊(duì)列：會(huì)被存儲(chǔ)在磁盤(pán)上，當(dāng)消息代理重啟時(shí)仍然存在，暫存隊(duì)列：當(dāng)前連接有效
        // exclusive:默認(rèn)也是false，只能被當(dāng)前創(chuàng)建的連接使用，而且當(dāng)連接關(guān)閉后隊(duì)列即被刪除。此參考優(yōu)先級(jí)高于durable
        // autoDelete:默認(rèn)是false,是否自動(dòng)刪除，當(dāng)沒(méi)有生產(chǎn)者或者消費(fèi)者使用此隊(duì)列，該隊(duì)列會(huì)自動(dòng)刪除。
        //一般設(shè)置一下隊(duì)列的持久化就好,其余兩個(gè)就是默認(rèn)false（消息和交換機(jī)也可以持久化，但是消息持久化的前提是需要和隊(duì)列，交換機(jī)持久化一起使用）
        //為業(yè)務(wù)隊(duì)列綁定一個(gè)死信交換機(jī)，當(dāng)業(yè)務(wù)隊(duì)列里的消息過(guò)期了就被轉(zhuǎn)發(fā)到死信交換機(jī)，再由死信交換機(jī)發(fā)給死信隊(duì)列處理
//        Map<String, Object> args = new HashMap<>(2);
//       x-dead-letter-exchange    這里聲明當(dāng)前隊(duì)列綁定的死信交換機(jī)
//        args.put("x-dead-letter-exchange", DEAD_LETTER_EXCHANGE);
//       x-dead-letter-routing-key  這里聲明當(dāng)前隊(duì)列的死信路由key
//        args.put("x-dead-letter-routing-key", DEAD_LETTER_KEY);
        //5000毫秒
//        args.put("x-message-ttl", 5000);
//        return new Queue("TestDirectQueue",false,false,false,args);
        return new Queue("TestDirectQueue",false,false,false);
    }
    //Direct交換機(jī) 起名：TestDirectExchange
    @Bean
    DirectExchange TestDirectExchange() {
        return new DirectExchange("TestDirectExchange",true,false);
    }
    //綁定  將隊(duì)列和交換機(jī)綁定, 并設(shè)置用于匹配鍵：TestDirectRouting
    @Bean
    Binding bindingDirect() {
        return BindingBuilder.bind(TestDirectQueue()).to(TestDirectExchange()).with("TestDirectRouting");
    }
    /**
     * ========================死信隊(duì)列==================================
     */
    // 聲明死信Exchange
    @Bean
    public DirectExchange deadLetterExchange(){
        return new DirectExchange(DEAD_LETTER_EXCHANGE);
    }
    // 聲明死信隊(duì)列A
    @Bean
    public Queue deadLetterQueueA(){
        return new Queue(DEAD_LETTER_QUEUE);
    }
    @Bean
    public Binding deadLetterBindingA(){
        return BindingBuilder.bind(deadLetterQueueA()).to(deadLetterExchange()).with(DEAD_LETTER_KEY);
    }
}
@Component
@RabbitListener(queues = "TestDirectQueue")//監(jiān)聽(tīng)的隊(duì)列名稱 TestDirectQueue
public class DirectReceiver {
     Integer index = 0;
    @RabbitHandler
    public void process(Channel channel,String msg) throws Exception {
        ++index;
        channel.basicQos(0, 1, false);
        //設(shè)置非自動(dòng)ack
        DefaultConsumer consumer = new DefaultConsumer(channel);
        channel.basicConsume("TestDirectQueue", false,consumer);
        //假設(shè)業(yè)務(wù)處理需要3秒，那么當(dāng)消費(fèi)者接受到消息的時(shí)候，只處理一條，且要處理3秒，那么在服務(wù)器堆積的多條信息就不會(huì)瘋狂涌入
        Thread.sleep(3000);
        System.out.println(" DirectReceiver消費(fèi)者收到消息  : " + msg + "，第" +index+ "條"+"======"+ new Date().toString());
    }
}

import com.rabbitmq.client.*;
public class Consumer {
    private final static String QUEUE_NAME = "queue_name";
    public static void main(String[] argv) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        int prefetchCount = 5; // 設(shè)置預(yù)取值為 5
        channel.basicQos(prefetchCount);
        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody(), "UTF-8");
            System.out.println("Received: " + message);
            // 模擬消息處理
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false); // 手動(dòng)應(yīng)答消息
        };
        channel.basicConsume(QUEUE_NAME, false, deliverCallback, consumerTag -> { });
    }
}