<rabbit:connection-factory id="connectionFactory" 
                           host="${rabbitmq.host}"
                           port="${rabbitmq.port}"
                           username="${rabbitmq.username}"
                           password="${rabbitmq.password}"
                           virtual-host="${rabbitmq.virtual-host}"
                           publisher-confirms="true"
/>

@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations = "classpath:spring-rabbitmq-producer.xml")
public class ProducerTest {
    /**
     * 注入RabbitTemplate
     */
    @Autowired
    private RabbitTemplate rabbitTemplate;
    /**
     * 測試默認的隊列發(fā)送消息
     */
    @Test
    public void testConfirmCallback() throws InterruptedException {
        // 設(shè)置回調(diào)
        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
            /**
             * 回調(diào)方法
             * @param correlationData 回調(diào)的相關(guān)數(shù)據(jù)。
             * @param ack true 表示發(fā)送成功, false 發(fā)送失敗
             * @param cause 失敗原因，ack==true->null
             */
            @Override
            public void confirm(CorrelationData correlationData, boolean ack, String cause) {
                if (ack) {
                    System.out.println("發(fā)送成功");
                } else {
                    System.out.println("發(fā)送失敗，原因：" + cause);
                    // 失敗后處理流程
                }
            }
        });
        rabbitTemplate.convertAndSend("spring_queue", "hello world");
        // 防止發(fā)送完成后，未完成回調(diào)關(guān)閉通道
        Thread.sleep(5000);
    }
}

<rabbit:connection-factory id="connectionFactory" host="${rabbitmq.host}"
                           port="${rabbitmq.port}"
                           username="${rabbitmq.username}"
                           password="${rabbitmq.password}"
                           virtual-host="${rabbitmq.virtual-host}"
                           publisher-returns="true"
/>

@Test
    public void testReturnCallback() throws InterruptedException {
        // 設(shè)置交換機處理失敗消息的模式
        rabbitTemplate.setMandatory(true);
        // 設(shè)置回調(diào)
        rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
            /**
             * 返回消息
             * @param message 消息對象
             * @param replyCode 錯誤碼
             * @param replyText 交換信息
             * @param exchange 交換機
             * @param routingKey 路由鍵
             */
            @Override
            public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
                System.out.println("消息對象：" + new String(message.getBody()));
                System.out.println("錯誤碼：" + replyCode);
                System.out.println("交換信息：" + replyText);
                System.out.println("交換機：" + exchange);
                System.out.println("路由鍵：" + routingKey);
            }
        });
        rabbitTemplate.convertAndSend("spring_direct_exchange", "direct_key_3",
                "spring_direct_exchange_direct_key_1");
        // 防止發(fā)送完成后，未完成回調(diào)關(guān)閉通道
        Thread.sleep(5000);
    }

/**
 * @author zhong
 * <p>
 * Consumer Ack機制
 * 1.設(shè)置手動簽收，acknowledge="manual"
 * 2.讓監(jiān)聽器實現(xiàn)ChannelAwareMessageListener接口
 * 3.如果消息成功處理，則調(diào)用channel.basicAck()
 * 4.如果消息處理失敗，則調(diào)用 channel.basicNack(),broker重新發(fā)送consumer
 */
@Component
public class AckSpringQueueListener implements ChannelAwareMessageListener {
    @Override
    public void onMessage(Message message, Channel channel) throws Exception {
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        // 接收消息
        System.out.println("Message:" + new String(message.getBody()));
        // 手動簽收
        /**
         * deliveryTag: 標(biāo)識id
         * multiple: 確認所有消息
         */
        channel.basicAck(deliveryTag, true);
        // 手動拒絕
        /**
         * requeue:如果被拒絕的消息應(yīng)該被重新排隊而不是被丟棄/死信
         */
        //channel.basicNack(deliveryTag, true, true);
    }
}

<context:component-scan base-package="org.example"/>
<rabbit:listener-container connection-factory="connectionFactory" acknowledge="manual" auto-declare="true">
    <rabbit:listener ref="ackSpringQueueListener" queue-names="spring_queue"/>
</rabbit:listener-container>

<context:component-scan base-package="org.example"/>
<rabbit:listener-container connection-factory="connectionFactory" acknowledge="manual" prefetch="1"
                           auto-declare="true">
    <rabbit:listener ref="qosListener" queue-names="spring_queue"/>
</rabbit:listener-container>

/**
 * Consumer 限流機制
 * 1.確保ack機制為手動確認
 * 2.listener-container 配置屬性
 * perfetch = 1 表示消費端每次從mq拉取一條消息來消費，直到手動確認消費完畢后，才會繼續(xù)拉去下一條消息。
 */
@Component
public class QosListener implements ChannelAwareMessageListener {
    @Override
    public void onMessage(Message message, Channel channel) throws Exception {
        System.out.println("QosListener:" + new String(message.getBody()));
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        // 簽收消息
        Thread.sleep(1000);
        channel.basicAck(deliveryTag, true);
    }
}

@Test
public void testTTL() {
    // 消息后處理隊列，設(shè)置一下消息參數(shù)信息
    MessagePostProcessor messagePostProcessor = message -> {
        // 1.設(shè)置message的消息
        message.getMessageProperties().setExpiration("50000");// 設(shè)置過期時間，字符串，毫秒
        // 2.返回消息
        return message;
    };
    // 傳入
    rabbitTemplate.convertAndSend("spring_fanout_exchange", "key", "RabbitMQ", messagePostProcessor);
}