RabbitMQ訊息佇列實現延遲任務範例

2022-04-02 19:00:26

一、序言

延遲任務應用廣泛，延遲任務典型應用場景有訂單超時自動取消；支付回撥重試。其中訂單超時取消具有冪等性屬性，無需考慮重複消費問題；支付回撥重試需要考慮重複消費問題。

延遲任務具有如下特點：在未來的某個時間點執行；一般僅執行一次。

1、實現原理

生產者將帶有延遲資訊的訊息傳送到RabbitMQ交換機中，等待延遲時間結束方將訊息轉發到繫結的佇列中，消費者通過監聽佇列消費訊息。延遲任務的關鍵在訊息在交換機中停留。

顯而易見，基於RabbitMQ實現延遲任務對伺服器的可靠性要求極高，交換機內部訊息無持久化機制，比如單機模式服務重啟，未開始的延遲任務均丟失。

2、元件選型

二、方案設計

（一）伺服器

RabbitMQ服務需要安裝x-delayed-message外掛以處理延遲訊息。

（二）生產者

延遲任務的實現對生產者的要求是將訊息可靠的投遞到交換機，因此使用confirm確認機制即可。

訂單生成之後，先入庫，然後以訂單ID為key將訂單詳情存入Redis中（持久化），向RabbitMQ傳送非同步confirm確定請求。如果收到正常投遞返回，則刪除Redis中訂單ID為key的資料，回收記憶體，否則以訂單ID為key，從Redis中查詢出訂單資料，重新傳送。

（三）消費者

延遲任務的實現對消費者的要求是以資訊不丟失的方式消費訊息，具體表現在：手動確認訊息的消費，防止訊息丟失；消費端持續穩定，防止訊息堆積；訊息消費失敗有重試機制。

考慮到訂單延遲取消屬於冪等性操作，因此無需考慮訊息的重複消費問題。

三、SpringBoot實現

實現部分僅貼一部分核心原始碼，完整專案請存取GitHub。

（一）生產者

考慮到下單是極為重要的操作，因此首先將訂單落庫、存檔，然後進行後續操作。

for (long i = 1; i <= 10; i++) {
    /* 1.模擬生成訂單 */
    BuOrder order = createOrder(i);
    /* 2.訂單入庫 */
    orderService.removeById(order);
    orderService.saveOrUpdate(order);
    /* 3.將訂單存入資訊Redis */
    RedisUtils.setObject(RabbitTemplateConfig.ORDER_PREFIX + i, order);
    /* 4.向RabbitMQ非同步投遞訊息 */
    rabbitTemplate.convertAndSend(RabbitmqConfig.DELAY_EXCHANGE_NAME, RabbitmqConfig.DELAY_KEY, order, RabbitUtils.setDelay(30000), RabbitUtils.correlationData(order.getOrderId()));
}

生產者可靠投遞訊息

public void confirm(CorrelationData correlationData, boolean ack, String cause) {
    if (correlationData == null) {
        return;
    }
    String key = ORDER_PREFIX + correlationData.getId();
    if (ack) {
        /* 如果訊息投遞成功，則刪除Redis中訂單資料，回收記憶體 */
        RedisUtils.deleteObject(key);
    } else {
        /* 從Redis中讀取訂單資料，重新投遞 */
        BuOrder order = RedisUtils.getObject(key, BuOrder.class);
        /* 重新投遞訊息 */
        rabbitTemplate.convertAndSend(RabbitmqConfig.DELAY_EXCHANGE_NAME, RabbitmqConfig.DELAY_KEY, order, RabbitUtils.setDelay(30000), RabbitUtils.correlationData(order.getOrderId()));
    }
}

（二）消費者

消費者端手動確認，避免訊息丟失；失敗自動重試。

@RabbitListener(queues = RabbitmqConfig.DELAY_QUEUE_NAME)
public void consumeNode01(Channel channel, Message message, BuOrder order) throws IOException {
    if (Objects.equals(0, order.getOrderStatus())) {
        /* 修改訂單狀態，設定為關閉狀態 */
        orderService.updateById(new BuOrder(order.getOrderId(), -1));
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        log.info(String.format("消費者節點01消費編號為【%s】的訊息", order.getOrderId()));
    }
}

消費者可靠消費應至少開啟兩個及以上應用，確保訊息佇列中不積壓訊息。

（三）通用工具包

上述程式碼涉及一個工具類RabbitUtils，存在於如下依賴中，主要封裝RabbitMQ極常用的工具方法。

<dependency>
  <groupId>xin.altitude.cms</groupId>
  <artifactId>ucode-cms-common</artifactId>
  <version>1.4.3.1</version>
</dependency>

以上就是RabbitMQ訊息佇列實現延遲任務範例的詳細內容，更多關於RabbitMQ訊息佇列延遲任務的資料請關注it145.com其它相關文章！