ApsaraMQ for RocketMQ:LiteTopic model

What is a LiteTopic

A LiteTopic is a secondary container for message transmission and storage within a parent topic. It groups messages of different child classes -- such as individual sessions, tasks, or users -- under the same business logic.

LiteTopics serve two purposes:

• Exclusive consumption and data fencing: Isolate data from different child classes into separate LiteTopics for finer-grained storage and subscription boundaries.

• Identity and permissions: Control access at a granularity finer than the topic level by assigning identities and permissions to individual LiteTopics.

How LiteTopics fit into the domain model

A topic is the top-level container for message transmission and storage in ApsaraMQ for RocketMQ. If a topic is of the Lite type, LiteTopics can be created within it. The combination of a topic and a LiteTopic uniquely identifies a message storage container.

Each LiteTopic contains one queue by default, and messages within that queue are stored and consumed in order.

LiteTopic properties

Name

• Must be globally unique within the parent topic.

• Auto-created on first use: if setLiteTopic is called on a message and the specified LiteTopic does not exist, the system creates it automatically.

• For naming constraints, see Parameter limits.

Time-to-live (TTL)

• If a LiteTopic receives no messages for longer than its TTL, the system deletes it automatically. Deletion releases the quota count.

• Set the TTL through the expiration value when you create the Lite-type topic.

• For TTL constraints, see Parameter limits.

Asynchronous communication for Multi-Agent systems

As AI workflows grow more complex, single agents are giving way to Multi-Agent architectures. Because AI tasks are long-running, synchronous inter-agent calls block the caller's thread and create scalability bottlenecks.

In this pattern, a Supervisor Agent splits a request and assigns sub-tasks to domain-specific sub-agents. The asynchronous communication flow is as follows:

Request flow:

1. Create a request topic for each sub-agent to serve as a task buffer queue. Use a priority topic to process high-priority tasks first.

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

Response flow:

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

2. Each sub-agent processes its task and sends the result to a LiteTopic within the response topic. Name the LiteTopic after the task ID to create a dedicated channel per task.

3. The Supervisor Agent retrieves results in real time and pushes them to the web client through HTTP Server-Sent Events (SSE).

Distributed session state management for AI applications

AI applications rely on long-running, multi-turn sessions. When a persistent connection (such as SSE or WebSocket) breaks due to a gateway restart, timeout, or network instability, the session context and in-progress AI computation can be lost.

LiteTopics decouple session state from the connection. Name each LiteTopic after the session ID (for example, chatbot/{sessionID}). Session results are transmitted in order as messages within the LiteTopic.

When a persistent connection breaks and reconnects to a different node:

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 scheduler sends results to chat/SessionID2 based on the session ID in the request.

4. A network issue causes the connection to reconnect to application service node 2.

5. Application service node 1 unsubscribes from chat/SessionID2. Application service node 2 subscribes to chat/SessionID2.

6. The LiteTopic continues to push unconsumed messages to application service node 2 from the previous consumption offset, preserving session continuity.

Send messages

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

final Message message = provider.newMessageBuilder()
    .setTopic(topic)
    // Set the message key for tracing.
    .setKeys("messageKey")
    // Route to a specific LiteTopic.
    .setLiteTopic("lite-topic-1")
    // Set the 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

Use the LitePushConsumer class to consume messages from LiteTopics:

// Initialize LitePushConsumer with consumer group, topic, and communication parameters.
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 the LiteTopics relevant to your workload.
    litePushConsumer.subscribeLite("lite-topic-1");
    litePushConsumer.subscribeLite("lite-topic-2");
    litePushConsumer.subscribeLite("lite-topic-3");
} catch (LiteSubscriptionQuotaExceededException e) {
    // The 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);
}

// Unsubscribe from LiteTopics that are no longer needed to release resources.
litePushConsumer.unsubscribeLite("lite-topic-3");

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

Modify subscriptions at runtime

/**
 * subscribeLite() initiates a network request and performs quota verification,
 * so the call may fail. Always check the result to confirm the subscription
 * was added successfully.
 *
 * Possible failure scenarios:
 * 1. A network error occurs. Retry the operation.
 * 2. Quota verification fails with a LiteSubscriptionQuotaExceededException.
 *    Evaluate your quota and use unsubscribeLite() to release unused subscriptions.
 */

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

// Remove a subscription when no longer needed.
litePushConsumer.unsubscribeLite("lite-topic-1");

LiteTopic creation quota

The creation quota is the total number of active LiteTopics within a single instance. If this limit is reached and a client sends a message to a non-existent LiteTopic (which would trigger auto-creation), the system rejects the message.

LiteTopic subscription quota

The subscription quota is the sum of all active subscription relationships between all online consumers and LiteTopics within an instance. If this limit is reached, new subscription attempts fail.

Serverless instances (dedicated, reserved + elastic)

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