ApsaraMQ for RocketMQ:Modèle de sujet Lite

Nom du sujet Lite

• Définition : Nom identifiant un sujet Lite. Les noms de sujets Lite sont globalement uniques au sein de leur sujet parent.

• Valeur : Lorsque le type de sujet est Lite et que vous appelez setLiteTopic sur un message, le système crée automatiquement le sujet Lite s'il n'existe pas déjà.

• Contrainte : Consultez les Limites des paramètres.

Durée de vie (TTL)

• Définition : Durée d'expiration d'un sujet Lite. Si aucun nouveau message n'est écrit dans le sujet Lite pendant une période supérieure à sa TTL, le système le supprime automatiquement. La suppression libère le quota alloué au sujet Lite (le compteur total est décrémenté de un).

• Valeur : Lors de la création d'un sujet de type Lite, vous pouvez définir le paramètre d'expiration.

• Contrainte : Consultez les Limites des paramètres.

Cas d'utilisation 1 : Communication asynchrone pour les systèmes Multi-Agent afin de résoudre le blocage des appels de longue durée

À mesure que les scénarios d'IA se complexifient, les systèmes mono-agent rencontrent des limites : manque de spécialisation, difficulté à intégrer plusieurs domaines et incapacité à permettre une prise de décision collaborative dynamique. Les applications et workflows mono-agent évoluent vers des architectures Multi-Agent. Toutefois, comme les tâches d'IA prennent souvent beaucoup de temps, les appels synchrones bloquent le thread de l'appelant, ce qui limite l'évolutivité pour une collaboration à grande échelle.

Comme illustré ci-dessus, le workflow Multi-Agent fonctionne de la manière suivante : l'agent Superviseur divise une requête en deux sous-tâches confiées à deux agents enfants. Chaque agent enfant résout sa partie et renvoie les résultats à l'agent Superviseur, qui les agrège et envoie la réponse finale au client web. L'utilisation de RocketMQ pour la communication asynchrone permet :

1. Flux de gestion des requêtes :

   1. Créez un sujet (Request) pour chaque agent enfant servant de file d'attente tampon pour les tâches. Utilisez un sujet prioritaire pour traiter d'abord les tâches à haute priorité.

   2. L'agent Superviseur envoie les détails des tâches divisées vers le sujet de requête correspondant.

2. Flux de gestion des réponses :

   1. L'agent Superviseur crée un sujet de type Lite (Response) et s'y abonne.

   2. Chaque agent enfant envoie le résultat de sa tâche vers un LiteTopic sous le sujet Response. Nommez chaque LiteTopic à l'aide de l'ID de tâche afin d'attribuer à chaque tâche son propre LiteTopic dédié.

   3. L'agent Superviseur reçoit les résultats en temps réel via l'abonnement et les transmet au client web en utilisant HTTP SSE.

Cas d'utilisation 2 : Gestion distribuée de l'état des sessions pour résoudre les problèmes de continuité de session dans les applications d'IA

Les interactions des applications d'IA présentent des caractéristiques uniques : elles sont de longue durée, comportent plusieurs tours et dépendent fortement de ressources de calcul coûteuses par session. Lorsque les applications s'appuient sur des connexions persistantes telles que SSE, toute déconnexion (due à des redémarrages de passerelle, des délais d'expiration ou une instabilité du réseau) entraîne la perte du contexte de session actuel et gaspille les ressources de calcul d'IA déjà investies.

Comme dans le flux de réponse du cas d'utilisation 1, utilisez un sujet de type Lite pour les notifications de résultats en temps réel. Nommez chaque LiteTopic à l'aide du SessionID (par exemple, chatbot/{sessionID}). Tous les résultats de session sont livrés sous forme de messages ordonnés dans ce sujet. Pour maintenir la continuité de la session après une reconnexion :

1. Le client web établit une connexion persistante avec le nœud 1 du serveur d'application et démarre la session Session2.

2. Le nœud 1 du serveur d'application s'abonne au LiteTopic [chat/SessionID2].

3. Le planificateur de tâches du grand modèle de langage (LLM) envoie les résultats vers le LiteTopic [chat/SessionID2] en fonction du SessionID présent dans la requête.

4. En raison d'un problème réseau, le WebSocket se reconnecte au nœud 2 du serveur d'application.

5. Le nœud 1 du serveur d'application se désabonne du LiteTopic [chat/SessionID2]. Le nœud 2 du serveur d'application s'y abonne.

6. Le LiteTopic [chat/SessionID2] reprend la livraison à partir du dernier offset consommé, garantissant ainsi la continuité de l'état et des données de la session.

Envoyer des messages

Producer producer = provider.newProducerBuilder()
    .setTopics(topic)
    .setClientConfiguration(clientConfiguration)
    .build();
final Message message = provider.newMessageBuilder()
    .setTopic(topic)
    // Set a message key for precise lookup by keyword.
    .setKeys("messageKey")
    // Set 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) {
    // LiteTopic quota exceeded. Evaluate and increase quota.
    log.error("Lite topic quota exceeded", e);
} catch (Throwable t) {
    log.error("Failed to send message", t);
}

Consommer des messages

Utilisez la classe LitePushConsumer :

// Initialize LitePushConsumer with consumer group, target topic, and communication parameters.
LitePushConsumer litePushConsumer = provider.newLitePushConsumerBuilder()
    .setClientConfiguration(clientConfiguration)
    // Topic bound to the ConsumerGroup when created in the console
    .bindTopic(topicName)
    // Set consumer group
    .setConsumerGroup(consumerGroup)
    .setMessageListener(messageView -> {
        // Process message and return consumption result.
        LOGGER.info("Consume message={}", messageView);
        return ConsumeResult.SUCCESS;
    })
    .build();
try {
    // Subscribe to desired LiteTopics
    litePushConsumer.subscribeLite("lite-topic-1");
    litePushConsumer.subscribeLite("lite-topic-2");
    litePushConsumer.subscribeLite("lite-topic-3");
} catch (LiteSubscriptionQuotaExceededException e) {
    // LiteTopic subscription quota exceeded. Evaluate and increase quota.
    log.error("Lite subscription quota exceeded", e);
} catch (Throwable t) {
    log.error("Failed to subscribe lite topic", t);
}
// After business processing, unsubscribe from unused LiteTopics promptly
litePushConsumer.unsubscribeLite("lite-topic-3");
// Get current set of subscribed LiteTopics
Set<String> liteTopicSet = litePushConsumer.getLiteTopicSet();

Mettre à jour dynamiquement les abonnements

/**
 * Dynamically add a subscription.
 * The subscribeLite() method makes a network call and validates quotas,
 * so it might fail.
 * Always check the result to confirm successful subscription.
 * Possible failure scenarios:
 * 1. Network error – retry the call.
 * 2. Quota validation fails – throws LiteSubscriptionQuotaExceededException.
 * Evaluate whether your quota meets requirements, and promptly call
 * unsubscribeLite() to release resources for unused topics.
 */
litePushConsumer.subscribeLite("lite-topic-1");
// Dynamically remove a subscription
litePushConsumer.unsubscribeLite("lite-topic-1");

Instances Serverless

Instances non Serverless (abonnement et paiement à l'utilisation)