ApsaraMQ for RocketMQ:Lite topic model

LiteTopic name

• Definition: The name of the LiteTopic. The name must be globally unique within the parent topic.

• Value: If the topic type is Lite and you call setLiteTopic for a message, the system automatically creates the corresponding LiteTopic if it does not exist.

• Constraint: For more information, see Parameter limits.

Time-to-live (TTL)

• Definition: The time-to-live (TTL) of a LiteTopic. If a LiteTopic does not receive any messages for a period longer than its TTL, it is automatically deleted. Deleting a LiteTopic releases its quota count.

• Value: When you create a Lite-type topic, you can set the expiration value.

• Constraint: For more information, see Parameter limits.

Scenario 1: Asynchronous communication for Multi-Agent systems to resolve blocking from long-running calls

As AI scenarios become more complex, most single agents face limitations in complex situations. They lack specialized division of labor, struggle to integrate multiple domains, and cannot make dynamic collaborative decisions. Single-agent applications and workflows are gradually shifting toward Multi-Agent applications. However, because AI tasks are long-running, synchronous calls can block the caller's thread, causing scalability issues in large-scale collaboration.

As shown in the preceding figure, the workflow for a Multi-Agent system is as follows: A Supervisor Agent splits a request and assigns it to two sub-agents. The two sub-agents handle their respective domain-specific problems and return the results to the Supervisor Agent. The Supervisor Agent aggregates the results and returns them to the web client. The ApsaraMQ for RocketMQ asynchronous communication solution works as follows:

1. In the request receiving flow:

   1. You can create a request topic for each sub-agent to serve as a buffer queue for tasks. This can be a priority topic to process high-priority tasks first.

   2. The Supervisor Agent sends the split task information to the corresponding request topics.

2. In the response flow:

   1. The Supervisor Agent creates a Lite-type response topic and subscribes to it.

   2. The sub-agent processes the task and sends the response result to a LiteTopic within the response topic. The LiteTopic can be named after the task ID to create a dedicated LiteTopic for each task.

   3. The Supervisor Agent subscribes to retrieve the results in real-time and then pushes them to the web client using the HTTP Server-Sent Events (SSE) protocol.

Scenario 2: Distributed session state management to solve session management challenges in AI applications

The interaction pattern of AI applications is unique. It involves long-running, multi-turn sessions that rely heavily on high-cost computing. When an application depends on a persistent connection, such as SSE, any interruption (such as a gateway restart, connection timeout, or network instability) can cause the loss of the current session context. This also invalidates the ongoing AI task, wasting valuable computing power.

As shown in the preceding figure, the response result flow is the same as in Scenario 1. A Lite-type topic is used for real-time notification of response results. Here, each LiteTopic can be named using the SessionID (for example, chatbot/{sessionID}). This way, session results are transmitted in order as messages in this topic. The following solution describes how to maintain session continuity after a persistent connection is re-established:

1. Web client 2 establishes a persistent connection with application service node 1 and creates Session2.

2. Application service node 1 subscribes to the LiteTopic [chat/SessionID2].

3. The large model task scheduling component sends the result to the LiteTopic [chat/SessionID2] based on the SessionID information in the request.

4. Due to a network issue or other exception, the WebSocket connection automatically reconnects to application service node 2.

5. Application service node 1 unsubscribes from the LiteTopic [chat/SessionID2], and application service node 2 subscribes to the LiteTopic [chat/SessionID2].

6. The LiteTopic [chat/SessionID2] continues to push subsequent unconsumed messages to application service node 2 based on the previous consumption progress. This ensures the continuity of the session state and data.

Send messages

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

final Message message = provider.newMessageBuilder()
    .setTopic(topic)
    // Set the message key. You can use the key to accurately find a specific message.
    .setKeys("messageKey")
    // Set the LiteTopic.
    .setLiteTopic("lite-topic-1")
    // Message body.
    .setBody("messageBody".getBytes())
    .build();

try {
    final SendReceipt sendReceipt = producer.send(message);
    log.info("Send message successfully, messageId={}", sendReceipt.getMessageId());
} catch (LiteTopicQuotaExceededException e) {
    // The LiteTopic quota is exceeded. Evaluate and increase the quota.
    log.error("Lite topic quota exceeded", e);
} catch (Throwable t) {
    log.error("Failed to send message", t);
}

Consume messages

You need to use the LitePushConsumer class:

// Initialize the LitePushConsumer. You need to bind the consumer group, target topic, communication parameters, and more.
LitePushConsumer litePushConsumer = provider.newLitePushConsumerBuilder()
    .setClientConfiguration(clientConfiguration)
    // The topic bound when the consumer group was created in the console.
    .bindTopic(topicName)
    // Set the consumer group.
    .setConsumerGroup(consumerGroup)
    .setMessageListener(messageView -> {
        // Process the message and return the consumption result.
        LOGGER.info("Consume message={}", messageView);
        return ConsumeResult.SUCCESS;
    })
    .build();

try {
    // Subscribe to a collection of LiteTopics that you are interested in.
    litePushConsumer.subscribeLite("lite-topic-1");
    litePushConsumer.subscribeLite("lite-topic-2");
    litePushConsumer.subscribeLite("lite-topic-3");
} catch (LiteSubscriptionQuotaExceededException e) {
    // The LiteTopic subscription quota is exceeded. Evaluate and increase the quota.
    log.error("Lite subscription quota exceeded", e);
} catch (Throwable t) {
    log.error("Failed to subscribe lite topic", t);
}

// After business processing is complete, promptly unsubscribe from LiteTopics that are no longer in use.
litePushConsumer.unsubscribeLite("lite-topic-3");

// Get the current set of subscribed LiteTopics.
Set<String> liteTopicSet = litePushConsumer.getLiteTopicSet();

Dynamically modify subscription relationships

/**
 * Dynamically add a subscription relationship.
 * The subscribeLite() method initiates a network request and performs quota verification, so the call may fail.
 * Always check the result of this call to ensure the subscription was added successfully.
 * Possible failure scenarios include the following:
 * 1. A network request error occurs. You can retry the operation.
 * 2. Quota verification fails, and a LiteSubscriptionQuotaExceededException is thrown.
 * Evaluate whether the quota meets your needs and promptly use unsubscribeLite() to unsubscribe from topics that are no longer in use to release resources.
 */

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

// Dynamically delete a subscription relationship.
litePushConsumer.unsubscribeLite("lite-topic-1");

Serverless instances

Non-Serverless (subscription and pay-as-you-go) instances