ApsaraMQ for RocketMQ:Consumption idempotence

Üretici yeniden denemesi

Bir üretici mesaj gönderir ve ApsaraMQ for RocketMQ aracı sunucusu (broker) bu mesajı kalıcı olarak depolar. Ancak geçici bir ağ sorunu veya üreticinin çökmesi nedeniyle aracı sunucu üreticiye onay (acknowledgment) gönderemeyebilir. Üretici bu durumu başarısız gönderim olarak algılar ve işlemi yeniden dener. Tüketici bu durumda içeriği aynı ancak mesaj kimlikleri farklı olan iki mesaj alır.

Aracı sunucunun yeniden teslim etmesi

Bir tüketici mesajı alır ve işler ancak geçici bir ağ sorunu nedeniyle aracı sunucuya gönderilen onay başarısız olur. Aracı sunucu mesajın tüketilip tüketilmediğini teyit edemediğinden en az bir kez teslim garantisi kapsamında ağ düzeldiğinde mesajı yeniden teslim eder. Tüketici bu durumda içeriği ve mesaj kimliği aynı olan iki mesaj alır.

Yük dengeleme

Ağ dalgalanmaları, aracı sunucunun yeniden başlatılması veya tüketici uygulamasının yeniden başlatılması gibi olaylar yük dengelemeyi tetikler. Yeniden dengeleme sırasında tüketici daha önce teslim edilmiş mesajları tekrar alabilir.

Adım 1: Üreticide mesaj anahtarını ayarlayın

Mesaj gönderirken benzersiz bir iş tanımlayıcısını mesaj anahtarı olarak ekleyin.

Message message = new Message();
message.setKey("ORDERID_100");
SendResult sendResult = producer.send(message);

ORDERID_100 ifadesini sipariş kimliği veya işlem kimliği gibi gerçek benzersiz iş tanımlayıcısıyla değiştirin.

Adım 2: Tüketicide mesaj anahtarını alın

Tüketici geri arama (callback) işlevinde mesaj anahtarını alın ve yinelenmez işlemeyi zorlamak için kullanın.

consumer.subscribe("ons_test", "*", new MessageListener() {
    public Action consume(Message message, ConsumeContext context) {
        String key = message.getKey()
        // Perform idempotent processing based on the message key that uniquely identifies your business.
    }
});

Adım 3: Yinelenen öğe kaldırma deposu ile yinelenmezliği zorlayın

Sadece mesaj anahtarı çift işlenmeyi engellemez; uygulamanızın anahtarın daha önce işlenip işlenmediğini kontrol etmesi gerekir. Yaygın bir desen benzersiz kısıtlamaya sahip ilişkisel bir veritabanı kullanmaktır:

1. İşlemeden önce mesaj anahtarını anahtar sütununda benzersiz kısıtlama olan bir yinelenen öğe kaldırma tablosuna ekleyin.

2. Ekleme başarılı olursa mesajı işleyin.

3. Ekleme birincil anahtar veya benzersiz kısıtlama ihlali nedeniyle başarısız olursa mesaj zaten işlenmiştir. Bu mesajı atlayın.

Yinelenen öğe kaldırma tablosu örneği (MySQL):

CREATE TABLE message_dedup (
    message_key VARCHAR(255) NOT NULL,
    created_at  TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    PRIMARY KEY (message_key)
);

Yinelenmez tüketici mantığı örneği:

consumer.subscribe("ons_test", "*", new MessageListener() {
    public Action consume(Message message, ConsumeContext context) {
        String key = message.getKey();

        try {
            // Attempt to insert the message key. Fails if already processed.
            insertMessageKey(key);
        } catch (DuplicateKeyException e) {
            // Already processed. Skip.
            return Action.CommitMessage;
        }

        // Process the business logic.
        processOrder(key);

        return Action.CommitMessage;
    }
});

insertMessageKey ve processOrder ifadelerini kendi veritabanı ve iş mantığı yöntemlerinizle değiştirin.

İlişkisel veritabanının darboğaz oluşturabileceği yüksek aktarım hızı gerektiren senaryolar için bunun yerine SETNX (SET if Not eXists) kullanan Redis tabanlı bir yaklaşım benimseyin.