ApsaraMQ for RocketMQ:Consumer types

Nasıl çalışır?

SDK dahili olarak bir Reactor iş parçacığı modeli kullanır:

1. Yerleşik bir uzun oylama iş parçacığı sunucudan mesajları zaman uyumsuz olarak çeker.

2. Mesajlar dahili bir önbellek kuyruğuna yerleştirilir.

3. SDK mesajları tüketici iş parçacıklarına dağıtır ve bu iş parçacıkları dinleyicinizi çağırır.

Dinleyici sonuçları

Mesaj dinleyicisi aşağıdaki sonuçlardan birini döndürmelidir:

Güvenilirlik kısıtlamaları

Push consumer başarı veya başarısızlığı kesinlikle dinleyici dönüş değerine göre belirler. Bu garantinin korunması için:

1. Zaman uyumlu işleyin. Sonucu döndürmeden önce tüm işlemleri tamamlayın.

2. Mesajları yeniden dağıtmayın. Mesajları diğer iş parçacıklarına devretmeyin ve bu iş parçacıkları bitmeden sonucu döndürmeyin.

Dinleyici işleme tamamlanmadan önce başarılı döner ve işleme daha sonra başarısız olursa sunucu mesajın tüketildiğini varsayar ve yeniden denemez.

Sıralı mesaj teslimi

Bir tüketici grubu sıralı tüketim modunu kullandığında Push consumer dinleyiciyi ek yapılandırma olmadan sıkı mesaj sırasına göre çağırır. Daha fazla bilgi için bkz. Sıralı mesajlar.

Push tüketicilerini ne zaman kullanmalısınız?

Push tüketicileri şu durumlarda en iyi performansı gösterir:

• İşleme süresi öngörülebilir. Öngörülemeyen süreler sık zaman aşımlarına neden olur; bu da yeniden denemeler yoluyla yinelenen mesajlara yol açar.

• Standart tüketim yeterlidir. SDK iş parçacığı modelini kontrol eder ve mesajları maksimum aktarım hızında iletir. Bu geliştirmeyi basitleştirir ancak zaman uyumsuz işlemeyi veya özel hız denetimini desteklemez.

SDK sınıfları

ApsaraMQ for RocketMQ Push tüketicileri için iki SDK sınıfı sağlar:

• **PushConsumer** -- Standart (Lite olmayan) konulardan mesajları tüketir.

• **LitePushConsumer** -- Lite tipi konulardan mesajları tüketir; tüketim denetimi Lite konu granülerliğinde yapılır.

// Consume normal messages with PushConsumer
ClientServiceProvider provider = ClientServiceProvider.loadService();
String topic = "<your-topic>";
FilterExpression filterExpression = new FilterExpression("<your-filter-tag>", FilterExpressionType.TAG);

PushConsumer pushConsumer = provider.newPushConsumerBuilder()
        // Set the consumer group
        .setConsumerGroup("<your-consumer-group>")
        // Set the access endpoint
        .setClientConfiguration(ClientConfiguration.newBuilder().setEndpoints("<your-endpoint>").build())
        // Bind the subscription
        .setSubscriptionExpressions(Collections.singletonMap(topic, filterExpression))
        // Register the message listener
        .setMessageListener(new MessageListener() {
            @Override
            public ConsumeResult consume(MessageView messageView) {
                // Process the message and return the result
                return ConsumeResult.SUCCESS;
            }
        })
        .build();

// Consume normal messages with LitePushConsumer
ClientServiceProvider provider = ClientServiceProvider.loadService();

LitePushConsumer litePushConsumer = provider.newLitePushConsumerBuilder()
        // Set the access endpoint
        .setClientConfiguration(ClientConfiguration.newBuilder().setEndpoints("<your-endpoint>").build())
        // Set the topic
        .bindTopic("<your-topic>")
        // Set the consumer group
        .setConsumerGroup("<your-consumer-group>")
        // Register the message listener
        .setMessageListener(messageView -> {
            // Process the message and return the result
            return ConsumeResult.SUCCESS;
        })
        .build();

// Subscribe to Lite topics
litePushConsumer.subscribeLite("<your-lite-topic-1>");
litePushConsumer.subscribeLite("<your-lite-topic-2>");

Nasıl çalışır?

1. Sunucudan bir toplu mesaj çekmek için ReceiveMessage işlemini çağırın.

2. Mesajları işlemek için iş parçacıklarınıza dağıtın.

3. Başarıyla işlenen her mesaj için AckMessage işlemini çağırın.

İşleme başarısız olursa onay göndermeyin. Mesaj görünmezlik süresi dolduğunda tekrar kullanılabilir hale gelir ve yeniden deneme tetiklenir. Daha fazla bilgi için bkz. SimpleConsumer yeniden deneme ilkesi.

API işlemleri

Hata işleme

Aşağıdaki tabloda farklı hata senaryolarının mesaj teslimini nasıl etkilediği açıklanmaktadır:

Sıralı mesaj teslimi

Bir Simple consumer sıralı mesajları depolama sırasına göre işler. Sıralı kalması gereken bir mesaj grubu için önceki mesaj işlenmeden sonraki mesaj alınamaz.

Simple tüketicilerini ne zaman kullanmalısınız?

Simple tüketicileri şu durumlarda en iyi performansı gösterir:

• İşleme süresi öngörülemez. ReceiveMessage çağrılırken başlangıç mesaj görünmezlik süresini belirtin; gerekirse ChangeInvisibleDuration ile bu süreyi uzatın.

• Özel iş akışları gereklidir. SDK herhangi bir iş parçacığı modeli empoze etmez; zaman uyumsuz dağıtım, toplu iş tüketimi veya herhangi bir özel düzeni uygulayın.

• Hız denetimi önemlidir. Kodunuz ReceiveMessage işlemini ne zaman ve ne sıklıkla çağıracağınıza karar verir; bu da aktarım hızı üzerinde doğrudan kontrol sağlar.

SDK sınıfı

ApsaraMQ for RocketMQ Simple tüketicileri için tek bir SDK sınıfı sağlar: SimpleConsumer. Bu sınıf Lite konularından mesajları tüketemez.

// Consume normal messages with SimpleConsumer
ClientServiceProvider provider = ClientServiceProvider.loadService();
String topic = "<your-topic>";
FilterExpression filterExpression = new FilterExpression("<your-filter-tag>", FilterExpressionType.TAG);

SimpleConsumer simpleConsumer = provider.newSimpleConsumerBuilder()
        // Set the consumer group
        .setConsumerGroup("<your-consumer-group>")
        // Set the access endpoint
        .setClientConfiguration(ClientConfiguration.newBuilder().setEndpoints("<your-endpoint>").build())
        // Bind the subscription
        .setSubscriptionExpressions(Collections.singletonMap(topic, filterExpression))
        .build();

try {
    // Pull up to 10 messages, wait up to 30 seconds
    List<MessageView> messageViewList = simpleConsumer.receive(10, Duration.ofSeconds(30));
    messageViewList.forEach(messageView -> {
        System.out.println(messageView);
        // Acknowledge each message after successful processing
        try {
            simpleConsumer.ack(messageView);
        } catch (ClientException e) {
            e.printStackTrace();
        }
    });
} catch (ClientException e) {
    // Handle failures such as throttling, then retry the receive call
    e.printStackTrace();
}

Push tüketicileri için işleme süresini kontrol edin

Mesaj işlemeyi zaman aşımı eşiği içinde tutun. Sık zaman aşımları gereksiz yeniden denemelere ve yinelenen mesajlara neden olur. Uygulamanız düzenli olarak uzun süren görevleri işliyorsa bir Simple consumer'a geçin ve uygun bir mesaj görünmezlik süresi ayarlayın.

LitePullConsumer'dan SimpleConsumer'a geçiş yapın

LitePullConsumer Apache RocketMQ 4.x Remoting SDK'da bir tüketici türüydü. 5.x gRPC SDK'da kaldırılmıştır ve Go, Java ve Python 5.x SDK'ları bu türü sağlamaz.

Daha önce çekme tabanlı tüketim için LitePullConsumer kullandıysanız 5.x SDK'ya yükselttikten sonra **SimpleConsumer** türüne geçiş yapın. SimpleConsumer aşağıdaki işlemleri sağlar:

• ReceiveMessage: Talep üzerine sunucudan mesajları çeker.

• AckMessage: Başarılı tüketimi onaylar.

• ChangeInvisibleDuration: Yeniden deneme aralığını kontrol etmek için mesaj görünmezlik süresini değiştirir.

Go SDK'sında karşılık gelen yöntemler Receive ve Ack iken Java SDK'sında karşılık gelen yöntemler receive ve ack şeklindedir.