ApsaraMQ for RocketMQ:Lite konu modeli

Lite konu adı

• Tanım: Bir lite konuyu tanımlayan addır. Lite konu adları, üst konuları içinde küresel olarak benzersizdir.

• Değer: Konu türü Lite olduğunda ve bir mesajda setLiteTopic çağrıldığında, sistem lite konu henüz mevcut değilse otomatik olarak oluşturur.

• Kısıtlama: Bkz. Parametre sınırları.

Yaşam süresi (TTL)

• Tanım: Bir lite konu için son kullanma süresidir. Lite konuya TTL süresinden daha uzun süre yeni mesaj yazılmazsa, sistem bunu otomatik olarak siler. Silme işlemi, lite konuya ayrılan kotanın düşülmesi anlamına gelir (toplam kota eksi bir).

• Değer: Lite türünde bir konu oluştururken son kullanma parametresini ayarlayabilirsiniz.

• Kısıtlama: Bkz. Parametre sınırları.

Kullanım durumu 1: Uzun süren çağrı engellemelerini çözmek için Çoklu Ajan sistemlerinde asenkron iletişim

Yapay zeka senaryoları karmaşıklaştıkça tek ajanlı sistemler yetersiz kalmaktadır: uzmanlık eksikliği, birden çok alanı entegre etmede zorluk ve dinamik iş birliğine dayalı karar verme yeteneğinin olmaması. Tek ajanlı uygulamalar ve iş akışları Çoklu Ajan mimarilerine doğru kaymaktadır. Ancak, yapay zeka görevleri genellikle uzun sürdüğü için senkron çağrılar arayanın iş parçacığını engeller ve bu durum büyük ölçekli iş birliği için ölçeklenebilirliği sınırlar.

Yukarıda gösterildiği gibi, Çoklu Ajan iş akışı şu şekilde çalışır: Supervisor Agent bir isteği iki alt göreve ayırarak iki alt ajana iletir. Her alt ajan kendi kısmını çözer ve sonuçları Supervisor Agent'a döndürür. Supervisor Agent bu sonuçları birleştirerek web istemcisine nihai yanıtı gönderir. Asenkron iletişim için RocketMQ kullanımı:

1. İstek işleme akışı:

   1. Her alt ajan için bir görev arabellek kuyruğu olarak bir konu (Request) oluşturun. Yüksek öncelikli görevleri önce işlemek için bir öncelik konusu kullanın.

   2. Supervisor Agent, bölünmüş görev ayrıntılarını ilgili istek konusuna gönderir.

2. Yanıt işleme akışı:

   1. Supervisor Agent, Lite türünde bir konu (Response) oluşturur ve buna abone olur.

   2. Her alt ajan, görev sonucunu Response konusu altındaki bir LiteTopic'e gönderir. Her göreve kendine ait özel bir LiteTopic vermek için her LiteTopic'i görev ID'sini kullanarak adlandırın.

   3. Supervisor Agent, abonelik yoluyla sonuçları gerçek zamanlı olarak alır ve bunları HTTP SSE kullanarak web istemcisine iletir.

Kullanım durumu 2: Yapay zeka uygulamalarında oturum sürekliliği sorunlarını çözmek için dağıtılmış oturum durumu yönetimi

Yapay zeka uygulama etkileşimleri benzersizdir: uzun süren, çok turlu ve oturum başına pahalı hesaplama kaynaklarına yoğun şekilde bağımlıdır. Uygulamalar SSE gibi kalıcı bağlantılara güvendiğinde, herhangi bir bağlantı kesilmesi (ağ geçidi yeniden başlatmaları, zaman aşımları veya ağ kararsızlığı nedeniyle) mevcut oturum bağlamının kaybına ve zaten harcanmış yapay zeka hesaplama kaynaklarının boşa gitmesine neden olur.

1. kullanım durumundaki yanıt akışında olduğu gibi, gerçek zamanlı sonuç bildirimleri için Lite türünde bir konu kullanın. Her LiteTopic'i SessionID kullanarak adlandırın (örneğin, chatbot/{sessionID}). Tüm oturum sonuçları bu konuda sıralı mesajlar olarak teslim edilir. Yeniden bağlanma sonrası oturum sürekliliğini korumak için:

1. Web istemcisi, uygulama sunucusu düğümü 1 ile kalıcı bir bağlantı kurar ve Session2 oturumunu başlatır.

2. Uygulama sunucusu düğümü 1, LiteTopic [chat/SessionID2]'ye abone olur.

3. Büyük Dil Modeli (LLM) görev zamanlayıcısı, istekteki SessionID'ye göre sonuçları LiteTopic [chat/SessionID2]'ye gönderir.

4. Bir ağ sorunu nedeniyle WebSocket, uygulama sunucusu düğümü 2'ye yeniden bağlanır.

5. Uygulama sunucusu düğümü 1, LiteTopic [chat/SessionID2] aboneliğini iptal eder. Uygulama sunucusu düğümü 2 buna abone olur.

6. LiteTopic [chat/SessionID2], teslimatı son tüketilen ofsetten devam ettirerek sürekli oturum durumu ve verilerini sağlar.

Mesaj gönderme

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

Mesaj tüketme

LitePushConsumer sınıfını kullanın:

// 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();

Abonelikleri dinamik olarak güncelleme

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

Serverless örnekler

Serverless olmayan örnekler (abonelik ve kullandıkça öde)