ApsaraMQ for RocketMQ:LiteTopic model

LiteTopic name

• Definition: The name of a LiteTopic. It identifies the LiteTopic and must be unique within its 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.

• Constraints: For more information, see Parameter limits.

Expiration time

• Definition: The expiration time of a LiteTopic. If a LiteTopic does not receive any new messages for a period that exceeds its expiration time, the LiteTopic is automatically deleted. When a LiteTopic is deleted, its resources are released, and the total count of LiteTopics is decremented by one.

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

• Constraints: 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, single agents are often limited in complex situations. They lack a specialized division of labor, struggle to integrate multiple domains, and cannot make dynamic collaborative decisions. As a result, single-agent applications and workflows are gradually being replaced by multi-agent systems. However, AI tasks are often long-running. Synchronous calls can block the caller's thread, which causes scalability issues in large-scale collaboration.

As shown in the preceding figure, the workflow of 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 process of using the ApsaraMQ for RocketMQ asynchronous communication solution is as follows:

1. In the request flow:

   1. Create a request topic for each sub-agent to serve as a buffer queue for tasks. You can use 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 LiteTopic for responses and subscribes to it.

   2. The sub-agents process the tasks and send the response for each task to a LiteTopic under the response topic. You can name the LiteTopic after the task ID to create a dedicated LiteTopic for each task.

   3. The supervisor agent subscribes to the LiteTopics 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 model of AI applications is unique. It involves long-running, multi-turn sessions that rely heavily on computationally expensive resources. When an application depends on persistent connections such as SSE, any connection interruption, such as a gateway restart, connection timeout, or network instability, not only causes the loss of the current session context but also invalidates ongoing AI tasks. This wastes valuable computing power.

As shown in the preceding figure, the response flow is the same as in Scenario 1. A LiteTopic is used to send real-time notifications of results. Here, each LiteTopic can be named using the session ID, for example, chatbot/{sessionID}. This way, session results are passed sequentially as messages in this topic. The solution to maintain session continuity after a persistent connection is re-established is as follows:

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

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

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

4. Due to a network issue or another 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 from where the previous consumption left off. This ensures session state and data continuity.

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 must use the LitePushConsumer class:

// Initialize LitePushConsumer. You must 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 of interest.
    litePushConsumer.subscribeLite("lite-topic-1");
    litePushConsumer.subscribeLite("lite-topic-2");
    litePushConsumer.subscribeLite("lite-topic-3");
} catch (LiteSubscriptionQuotaExceededException e) {
    // The LiteTopic 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 set of currently 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.
 * Make sure to check the result of this call to ensure that the subscription is successfully added.
 * 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 series instances

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