ApsaraMQ for RocketMQ：輕量主題模型

輕量主題名稱

• 定義：輕量主題的名稱，用於標識輕量主題，輕量主題名稱在所屬主題內全域唯一。

• 取值：當主題類型為Lite時，使用者對message進行了setLiteTopic，如對應的輕量主題不存在，系統會自動建立。

• 約束：請參見參數限制。

到期時間

• 定義：輕量主題的到期時間，當輕量主題距離最近一次訊息寫入時間超過到期時間後，該輕量主題會被自動刪除。刪除是指釋放輕量主題佔用的個數，即佔用總數-1。

• 取值：當建立主題類型為Lite時，可以設定到期時間 expiration 值。

• 約束：請參見參數限制。

應用情境 1：Multi-Agent 的非同步通訊，解決長耗時調用阻塞痛點

隨著AI需求情境變得更加複雜，大多數單Agent在複雜情境中面臨著局限性：缺乏專業化分工、難以對多領域進行整合；無法實現動態協作決策。單Agent應用和單Agent工作流程會逐步轉向 Multi-Agent 應用。但因為 AI 任務長耗時的特點，同步調用會造成調用者的線程阻塞，存在大規模協作擴充性問題。

如上圖所示 Multi-Agent的工作流程為：Supervisor Agent 負責將需求拆分給兩個子Agent；兩個子Agent負責各自領域問題解答並將結果返回給Supervisor Agent； Supervisor Agent將結果匯總返回給Web端。使用 RocketMQ 非同步通訊方案流程如下：

1. 在接收請求的流程中

   1. 為每個子Agent建立一個Topic (Request)用於請求任務的緩衝隊列，可以是優先順序Topic，能先處理高優任務。

   2. Supervisor Agent 將分割任務資訊發送到對應的請求主題中。

2. 在返迴響應結果流程中

   1. Supervisor Agent 建立 Lite 類型的Topic (Response) 並訂閱這個Topic。

   2. 子 Agent 處理將每個任務的響應結果發送到Topic (Response) 的LiteTopic中。LiteTopic可以用任務ID命名，為每個任務建立一個專屬的LiteTopic。

   3. Supervisor Agent 通過訂閱即時擷取結果，然後通過HTTP SSE協議推送給Web端。

應用情境 2：分布式工作階段狀態管理，終結 AI 應用的工作階段狀態管理難題

AI 應用的互動模式具有特殊性，即長耗時、多輪次且高度依賴高成本計算的會話。當應用依賴 SSE 等長串連時，一旦串連中斷（如網關重啟、連線逾時、網路不穩定觸發），不僅會導致當前會話內容相關的丟失，更會直接造成已投入的 AI 任務作廢，從而浪費寶貴的算力資源。

如上圖所示，和情境 1的返迴響應結果流程相同，使用Lite 類型的Topic作為即時通知響應結果。這裡每個LiteTopic可以使用 SessionID 命名（例如 chatbot/{sessionID}），這樣會話結果都作為訊息在這個主題中有序傳遞。長串連重連後繼續保持會話的連續性的解決方案如下：

1. Web端2 和應用服務節點 1 建立長連結，建立會話Session2。

2. 應用服務節點 1 監聽 LiteTopic [chat/SessionID2]

3. 大模型任務調度組件，根據請求的SessionID資訊，將返回結果發送到 LiteTopic [chat/SessionID2]

4. 由於網路等異常，WebSocket 自動重連到了應用服務節點2。

5. 應用服務節點 1 取消訂閱LiteTopic [chat/SessionID2]，應用服務節點 2 訂閱LiteTopic [chat/SessionID2]。

6. LiteTopic [chat/SessionID2] 根據之前的消費進度，將後續未消費的訊息繼續推送給應用服務節點 2。保證了工作階段狀態和會話資料的連續性。

發送訊息

Producer producer = provider.newProducerBuilder()
    .setTopics(topic)
    .setClientConfiguration(clientConfiguration)
    .build();

final Message message = provider.newMessageBuilder()
    .setTopic(topic)
    //設定訊息索引鍵，可根據關鍵字精確尋找某條訊息。
    .setKeys("messageKey")
    //設定LiteTopic
    .setLiteTopic("lite-topic-1")
    //訊息體
    .setBody("messageBody".getBytes())
    .build();

try {
    final SendReceipt sendReceipt = producer.send(message);
    log.info("Send message successfully, messageId={}", sendReceipt.getMessageId());
} catch (LiteTopicQuotaExceededException e) {
    // LiteTopic配額超限，評估並增加配額
    log.error("Lite topic quota exceeded", e);
} catch (Throwable t) {
    log.error("Failed to send message", t);
}

消費訊息

需要使用 LitePushConsumer 消費類：

//初始化LitePushConsumer，需要綁定消費者分組ConsumerGroup、目標主題Topic、通訊參數等。
LitePushConsumer litePushConsumer = provider.newLitePushConsumerBuilder()
    .setClientConfiguration(clientConfiguration)
    // 控制台建立ConsumerGroup時綁定的Topic
    .bindTopic(topicName)
    //設定消費者分組
    .setConsumerGroup(consumerGroup)
    .setMessageListener(messageView -> {
        //處理訊息並返回消費結果。
        LOGGER.info("Consume message={}", messageView);
        return ConsumeResult.SUCCESS;
    })
    .build();

try {
    // 訂閱感興趣的LiteTopic集合
    litePushConsumer.subscribeLite("lite-topic-1");
    litePushConsumer.subscribeLite("lite-topic-2");
    litePushConsumer.subscribeLite("lite-topic-3");
} catch (LiteSubscriptionQuotaExceededException e) {
    // LiteTopic配額超限，評估並增加配額
    log.error("Lite subscription quota exceeded", e);
} catch (Throwable t) {
    log.error("Failed to subscribe lite topic", t);
}

// 業務處理完畢後，及時取消不再使用的LiteTopic訂閱
litePushConsumer.unsubscribeLite("lite-topic-3");

// 擷取當前訂閱的LiteTopic集合
Set<String> liteTopicSet = litePushConsumer.getLiteTopicSet();

動態修改訂閱關係

/**
 * 動態添加訂閱關係
 * subscribeLite() 方法會發起網路請求並執行配額驗證，因此可能會調用失敗。
 * 請務必檢查此調用的結果，以確保訂閱被成功添加。
 * 可能的失敗情境包括：
 * 1. 網路請求錯誤，可以重試。
 * 2. 配額驗證失敗，拋出 LiteSubscriptionQuotaExceededException 異常。
 * 請評估配額是否滿足需求，並及時使用 unsubscribeLite() 取消訂閱不再使用的主題以釋放資源。
 */

litePushConsumer.subscribeLite("lite-topic-1");

//動態刪除訂閱關係
litePushConsumer.unsubscribeLite("lite-topic-1");

Serverless系列執行個體

非Serverless系列（訂用帳戶、隨用隨付）執行個體