§1 RabbitMQ延遲佇列

RabbitMQ延遲佇列，主要是藉助訊息的TTL（Time to Live）和死信exchange（Dead Letter Exchanges）來實現。

涉及到2個佇列，一個用於傳送訊息，一個用於訊息過期後的轉發目標佇列。

本例中， 定義2組exchange和queue。

agentpayquery1exchange		agentpayquery1queue（routingkey為delay）
agentpayquery2exchange		agentpayquery2queue（routingkey為delay）

agentpayquery1queue是緩衝佇列，訊息過期路由到agentpayquery2queue

§2 生產者

生產者配置：

<!-- 連線服務配置 -->
<rabbit:connection-factory
        id="connectionFactoryProducer"
        addresses="${mq.ip}"    //192.168.40.40:5672
        username="${username}"
        password="${password}"
        channel-cache-size="${cache.size}"
        publisher-confirms
 
="${publisher.confirms}"
        publisher-returns="${publisher.returns}"
        virtual-host="/"
/>

<!--========================出款查詢 延遲佇列配置 begin =========================-->
<rabbit:queue id="agentpayquery2queue" durable="true" auto-delete="false" exclusive="false" name="agentpayquery2queue"
 
/>
<rabbit:direct-exchange name="agentpayquery2exchange" durable="true" auto-delete="false" id="agentpayquery2exchange">
    <rabbit:bindings>
        <rabbit:binding queue="agentpayquery2queue" key="delay" />
    </rabbit:bindings>
</rabbit:direct-exchange>


<rabbit:queue id="agentpayquery1queue" durable="true" auto-delete="false" exclusive="false" name="agentpayquery1queue" >
    <rabbit:queue-arguments>
        <entry key="x-dead-letter-exchange" value="agentpayquery2exchange"/>
    </rabbit:queue-arguments>
</rabbit:queue>
<rabbit:direct-exchange name="agentpayquery1exchange" durable="true" auto-delete="false" id="agentpayquery1exchange">
    <rabbit:bindings>
        <rabbit:binding queue="agentpayquery1queue" key="delay" />
    </rabbit:bindings>
</rabbit:direct-exchange>

<!--定義RabbitTemplate例項-->
<rabbit:template id="agentpayQueryMsgTemplate"
                 exchange="agentpayquery1exchange"  routing-key="delay"
                 queue="agentpayquery1queue"
                 connection-factory="connectionFactoryProducer" message-converter="mqMessageConverter"
                 mandatory="true"
                 confirm-callback="publisherConfirmsReturns" return-callback="publisherConfirmsReturns"/>
<!--========================出款查詢 延遲佇列配置 end =========================-->

生產者訊息入隊：

import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageDeliveryMode;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

@Service
public class AgentpayQueryProducer {

    private static final Logger log = LogManager.getLogger(AgentpayQueryProducer.class.getSimpleName());

    @Autowired
    private RabbitTemplate agentpayQueryMsgTemplate;

    public void sendDelay(String message, int delaySeconds) {
        String expiration = String.valueOf(delaySeconds * 1000);
        agentpayQueryMsgTemplate.convertAndSend((Object) message, new MessagePostProcessor() {
            @Override
            public Message postProcessMessage(Message message)
                    throws AmqpException {
                message.getMessageProperties().setExpiration(expiration);
                message.getMessageProperties().setDeliveryMode(MessageDeliveryMode.PERSISTENT);
                log.info("出款查詢資料入隊:{}", new String(message.getBody()));
                return message;
            }
        });
    }
}

§3消費者

消費端的配置無他：

<!-- 連線服務配置  channel-cache-size="25" -->
<rabbit:connection-factory id="connectionFactory"
                           addresses="${mq.ip}"
                           username="${username}"
                           password="${password}" />

<bean id="agentpayQueryConsumer" class="com.emaxcard.rpc.payment.service.impl.batchpay.AgentpayQueryConsumer" />

<!-- queue litener  觀察 監聽模式 當有訊息到達時會通知監聽在對應的佇列上的監聽物件-->
<rabbit:queue id="agentpayquery2queue" durable="true" auto-delete="false" exclusive="false" name="agentpayquery2queue" />

<rabbit:listener-container connection-factory="connectionFactory" acknowledge="auto"

                           max-concurrency="20"
                           concurrency="10"
                           prefetch="10">
    <rabbit:listener ref="agentpayQueryConsumer" queues="agentpayquery2queue" />
</rabbit:listener-container>

訊息消費：

import com.alibaba.fastjson.JSON;
import com.emaxcard.enums.BatchPayStatus;
import com.emaxcard.exceptions.ResponseException;
import com.emaxcard.payment.vo.PaymentRecord;
import com.emaxcard.rpc.payment.model.PaymentRecordModel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageListener;
import org.springframework.beans.factory.annotation.Autowired;

public class AgentpayQueryConsumer implements MessageListener {

    private static final Logger log = LogManager.getLogger();

    @Autowired
    QueryGatewayService queryGatewayService;
    @Autowired
    AgentpayQueryProducer agentpayQueryProducer;

    @Override
    public void onMessage(Message message) {
        String mqMsg = new String(message.getBody());
        log.info("出款查詢資料出隊：{}", mqMsg);
        PaymentRecord paymentRecordModel;
        try {
            paymentRecordModel = JSON.parseObject(mqMsg, PaymentRecord.class);
        } catch (Exception ex) {
            log.info("訊息格式不是PaymentRecordModel，結束。");
            return;
        }

        try {
            BatchPayStatus payStatus = queryGatewayService.queryGateway(paymentRecordModel);

            // 非終態，繼續放入延遲佇列
            if (BatchPayStatus.SUCCESS != payStatus && BatchPayStatus.FAILED != payStatus) {
                if (BatchPayStatus.NOTEXIST == payStatus) {
                    log.info("查詢結果是{}，不再處理", payStatus);
                } else {
                    agentpayQueryProducer.sendDelay(mqMsg, 10);
                }
            }
        } catch (Exception ex) {
            if (ex instanceof ResponseException) {
                log.info("轉賬查詢{},paymentId{},處理錯誤：{}",
                        paymentRecordModel.getTransNo(), paymentRecordModel.getPaymentId(), ex.getMessage());
            } else {
                log.error("處理訊息異常：", ex);
            }
        }

    }
}

§4 使用延遲佇列要注意

1. 因為是佇列，所以即使一個訊息比在同一佇列中的其他訊息提前過期，提前過期的也不會優先進入死信佇列，它們還是按照入庫的順序讓消費者消費。如果第一進去的訊息過期時間是1小時，那麼死信佇列的消費者也許等1小時才能收到第一個訊息。

2. 當緩衝佇列裡一旦出現未設定過期時間的訊息，那麼就會造成整個佇列堵塞。消費端也無法消費到訊息。通過日誌可以看到，打印出來的都是 BlockingQueueConsumer。

Get messages Ack Mode選擇“Ack message requeue false”，可以將訊息消費掉