<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-test</artifactId>
        <scope>test</scope>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-amqp</artifactId>
        <version>2.5.0</version>
    </dependency>
    <dependency>
        <groupId>junit</groupId>
        <artifactId>junit</artifactId>
        <version>4.12</version>
        <scope>test</scope>
    </dependency>
</dependencies>

@Component
public class RabbitMQConfig {
    public static final String EXCHANGE_NAME = "order_exchange_name";
    public static final String QUEUE_NAME = "order_queue_name";
    public static final String DELAY_EXCHANGE_NAME = "delay_exchange_name";
    public static final String DELAY_QUEUE_NAME = "delay_queue_name";
    @Bean("orderExchange")
    public Exchange testExchange(){
        return ExchangeBuilder.topicExchange(EXCHANGE_NAME).durable(true).build();
    }
    @Bean("delayExchange")
    public Exchange deadExchange(){
        return ExchangeBuilder.topicExchange(DELAY_EXCHANGE_NAME).durable(true).build();
    }
    //訂單隊(duì)列綁定延遲交換機(jī)并且?guī)下酚涉I
    @Bean("orderQueue")
    public Queue testQueue(){
        return QueueBuilder.durable(QUEUE_NAME).deadLetterExchange(DELAY_EXCHANGE_NAME)
                .deadLetterRoutingKey("order.delay.user").build();
    }
    @Bean("delayQueue")
    public Queue deadQueue(){
        return QueueBuilder.durable(DELAY_QUEUE_NAME).build();
    }
    @Bean
    public Binding link(@Qualifier("orderExchange") Exchange exchange,
                        @Qualifier("orderQueue") Queue queue){
        return BindingBuilder.bind(queue).to(exchange).with("order.#").noargs();
    }
    @Bean
    public Binding deadLink(@Qualifier("delayExchange") Exchange exchange,
                            @Qualifier("delayQueue") Queue queue){
        return BindingBuilder.bind(queue).to(exchange).with("order.delay.#").noargs();
    }
}

@SpringBootTest
@RunWith(SpringRunner.class)
class RabbitmqProducerApplicationTests {
    @Autowired
    private RabbitTemplate rabbitTemplate;
    @Test
    void testProducer() throws InterruptedException {
        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
            @Override
            public void confirm(CorrelationData correlationData, boolean b, String s) {
                if(b) System.out.println("交換機(jī)成功接受到了消息");
                else System.out.println("消息失敗原因" + s);
            }
        });
        // 設(shè)置交換機(jī)處理失敗消息的模式
        // true:消息到達(dá)不了隊(duì)列時(shí) 會(huì)將消息重新返回給生產(chǎn)者 false:消息到達(dá)不了隊(duì)列直接丟棄
        rabbitTemplate.setMandatory(true);
        rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
            @Override
            public void returnedMessage(Message message, int i, String s, String s1, String s2) {
                System.out.println("隊(duì)列接受不到交換機(jī)的消息進(jìn)行了失敗回調(diào)");
            }
        });
        // 以上代碼均是為了保證消息的可靠性傳遞
        // 對(duì)消息進(jìn)行后置處理 設(shè)置其過(guò)期時(shí)間為10s
        MessagePostProcessor messagePostProcessor = new MessagePostProcessor() {
            @Override
            public Message postProcessMessage(Message message) throws AmqpException {
                message.getMessageProperties().setExpiration("10000");
                return message;
            }
        };
        // 下單成功發(fā)送消息
        rabbitTemplate.convertAndSend(RabbitMQConfig.EXCHANGE_NAME,"order.delay.user","ikun書(shū)籍", messagePostProcessor);
    }
}

<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-test</artifactId>
        <scope>test</scope>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-amqp</artifactId>
        <version>2.5.0</version>
    </dependency>
    <dependency>
        <groupId>org.projectlombok</groupId>
        <artifactId>lombok</artifactId>
    </dependency>
    <dependency>
        <groupId>org.slf4j</groupId>
        <artifactId>slf4j-api</artifactId>
    </dependency>
</dependencies>

@Slf4j
@Component
public class OrderListener implements ChannelAwareMessageListener {
    @RabbitListener(queues = "delay_queue_name") // 監(jiān)聽(tīng)的是死信隊(duì)列
    @Override
    public void onMessage(Message message, Channel channel) throws Exception {
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        try {
            // 這里要注意在監(jiān)聽(tīng)這條消息前 肯定會(huì)有接口點(diǎn)擊支付去更改支付狀態(tài)的業(yè)務(wù)邏輯 此處我是做了10s的訂單業(yè)務(wù)而已
            log.info("您在時(shí)間為：{},時(shí)有一條訂單為：{}", LocalDateTime.now().minusSeconds(10), new String(message.getBody()));
            // 下面開(kāi)始接受訂單消息邏輯
            log.info("將訂單id傳入數(shù)據(jù)庫(kù)查詢(xún)訂單支付字段");
            log.info("字段為支付成功狀態(tài)就手動(dòng)確認(rèn)簽收");
            log.info("字段為未支付狀態(tài)就取消訂單并且回滾事務(wù)");
            channel.basicAck(deliveryTag,false);// 僅確認(rèn)本次消息
        } catch (Exception e){
            log.info("出現(xiàn)異常 拒絕簽收消息 并且不重回隊(duì)列");
            channel.basicNack(deliveryTag,false,false);
        }
    }
}