ApsaraMQ for RocketMQ:Modelo de lite topic

Nome do lite topic

• Definição: Nome que identifica um lite topic. Os nomes dos lite topics são globalmente exclusivos dentro de seu tópico pai.

• Valor: Quando o tipo do tópico é Lite e você chama setLiteTopic em uma mensagem, o sistema cria automaticamente o lite topic caso ele ainda não exista.

• Restrição: Consulte Limites de parâmetros.

Tempo de vida (TTL)

• Definição: Tempo de expiração de um lite topic. Se nenhuma nova mensagem for gravada no lite topic por um período superior ao seu TTL, o sistema o exclui automaticamente. A exclusão libera a contagem alocada ao lite topic (contagem total menos um).

• Valor: Ao criar um tópico do tipo Lite, defina o parâmetro de expiração.

• Restrição: Consulte Limites de parâmetros.

Caso de uso 1: Comunicação assíncrona para sistemas Multi-Agent visando resolver bloqueios em chamadas de longa duração

À medida que os cenários de IA se tornam mais complexos, os sistemas de agente único enfrentam limitações: falta de especialização, dificuldade de integração de múltiplos domínios e incapacidade de permitir tomada de decisão colaborativa dinâmica. Aplicações e fluxos de trabalho de agente único estão migrando para arquiteturas Multi-Agent. No entanto, como as tarefas de IA geralmente levam muito tempo, chamadas síncronas bloqueiam a thread do chamador e limitam a escalabilidade para colaboração em larga escala.

Conforme ilustrado acima, o fluxo de trabalho Multi-Agent funciona da seguinte maneira: O Supervisor Agent divide uma solicitação em duas subtarefas para dois agentes filhos. Cada agente filho resolve sua parte e retorna os resultados ao Supervisor Agent, que os agrega e envia a resposta final ao cliente web. Utilize o RocketMQ para comunicação assíncrona:

1. Fluxo de tratamento de solicitações:

   1. Crie um tópico (Request) para cada agente filho como uma fila de buffer de tarefas. Use um tópico prioritário para processar primeiro as tarefas de alta prioridade.

   2. O Supervisor Agent envia os detalhes das tarefas divididas para o tópico de solicitação correspondente.

2. Fluxo de tratamento de respostas:

   1. O Supervisor Agent cria um tópico do tipo Lite (Response) e o assina.

   2. Cada agente filho envia o resultado de sua tarefa para um LiteTopic sob o tópico Response. Nomeie cada LiteTopic usando o ID da tarefa para fornecer a cada tarefa seu próprio LiteTopic dedicado.

   3. O Supervisor Agent recebe os resultados em tempo real por meio da assinatura e os envia ao cliente web usando HTTP SSE.

Caso de uso 2: Gerenciamento distribuído de estado de sessão para resolver problemas de continuidade de sessão em aplicações de IA

As interações em aplicações de IA são únicas: de longa duração, com múltiplas turnos e altamente dependentes de recursos computacionais caros por sessão. Quando as aplicações dependem de conexões persistentes, como SSE, qualquer desconexão (devido a reinicializações de gateway, tempos limite ou instabilidade de rede) causa perda do contexto da sessão atual e desperdiça recursos computacionais de IA já investidos.

Assim como no fluxo de resposta do caso de uso 1, utilize um tópico do tipo Lite para notificações de resultados em tempo real. Nomeie cada LiteTopic usando o SessionID (por exemplo, chatbot/{sessionID}). Todos os resultados da sessão são entregues como mensagens ordenadas neste tópico. Para manter a continuidade da sessão após a reconexão:

1. O cliente web estabelece uma conexão persistente com o nó 1 do servidor de aplicações e inicia a sessão Session2.

2. O nó 1 do servidor de aplicações assina o LiteTopic [chat/SessionID2].

3. O agendador de tarefas do Large Language Model (LLM) envia os resultados para o LiteTopic [chat/SessionID2] com base no SessionID da solicitação.

4. Devido a um problema de rede, o WebSocket se reconecta ao nó 2 do servidor de aplicações.

5. O nó 1 do servidor de aplicações cancela a assinatura do LiteTopic [chat/SessionID2]. O nó 2 do servidor de aplicações passa a assiná-lo.

6. O LiteTopic [chat/SessionID2] retoma a entrega a partir do último offset consumido, garantindo a continuidade do estado e dos dados da sessão.

Enviar mensagens

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);
}

Consumir mensagens

Utilize a 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();

Atualizar assinaturas dinamicamente

/**
 * 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");

Instâncias Serverless

Instâncias não Serverless (assinatura e pagamento conforme o uso)