ApsaraMQ for RocketMQ:Send and receive ordered messages

Üretim sırası ve tüketim sırası

Uçtan uca FIFO teslimatı iki bağımsız faktöre bağlıdır:

Katı sıralama için her iki koşulun da sağlanması gerekir: tek bir üretici iş parçacığından gönderim yapılmalı ve consumeMessageOrderly ile tüketilmelidir.

import com.aliyun.mq.http.MQClient;
import com.aliyun.mq.http.MQProducer;
import com.aliyun.mq.http.model.TopicMessage;

import java.util.Date;

public class OrderProducer {

    public static void main(String[] args) {
        MQClient mqClient = new MQClient(
                "<your-http-endpoint>",
                // Read credentials from environment variables to avoid hardcoding secrets
                System.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"),
                System.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET")
        );

        final String topic = "<your-topic>";
        // Set to null or "" if the instance has no namespace
        final String instanceId = "<your-instance-id>";

        // Get the producer for the specified topic
        MQProducer producer;
        if (instanceId != null && instanceId != "") {
            producer = mqClient.getProducer(instanceId, topic);
        } else {
            producer = mqClient.getProducer(topic);
        }

        try {
            // Send 8 messages, alternating between 2 sharding keys (partitions)
            for (int i = 0; i < 8; i++) {
                TopicMessage pubMsg = new TopicMessage(
                        "hello mq!".getBytes(),   // Message body
                        "A"                        // Message tag for filtering
                );
                // Sharding key determines the partition. Messages with the same
                // sharding key are delivered to the same partition in FIFO order.
                pubMsg.setShardingKey(String.valueOf(i % 2));
                pubMsg.getProperties().put("a", String.valueOf(i));

                // Publish synchronously. No exception means the send succeeded.
                TopicMessage pubResultMsg = producer.publishMessage(pubMsg);

                System.out.println(new Date() + " Send mq message success. Topic is:" + topic
                        + ", msgId is: " + pubResultMsg.getMessageId()
                        + ", bodyMD5 is: " + pubResultMsg.getMessageBodyMD5());
            }
        } catch (Throwable e) {
            // Handle send failure: retry or persist the message for later delivery
            System.out.println(new Date() + " Send mq message failed. Topic is:" + topic);
            e.printStackTrace();
        }

        mqClient.close();
    }
}

import com.aliyun.mq.http.MQClient;
import com.aliyun.mq.http.MQConsumer;
import com.aliyun.mq.http.common.AckMessageException;
import com.aliyun.mq.http.model.Message;

import java.util.ArrayList;
import java.util.List;

public class OrderConsumer {

    public static void main(String[] args) {
        MQClient mqClient = new MQClient(
                "<your-http-endpoint>",
                System.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"),
                System.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET")
        );

        final String topic = "<your-topic>";
        final String groupId = "<your-group-id>";
        // Set to null or "" if the instance has no namespace
        final String instanceId = "<your-instance-id>";

        final MQConsumer consumer;
        if (instanceId != null && instanceId != "") {
            consumer = mqClient.getConsumer(instanceId, topic, groupId, null);
        } else {
            consumer = mqClient.getConsumer(topic, groupId);
        }

        // Continuously poll for messages
        do {
            List<Message> messages = null;

            try {
                // consumeMessageOrderly guarantees partition-level FIFO order.
                // The consumer must ACK all messages in a batch before the broker
                // delivers the next batch from the same partition.
                messages = consumer.consumeMessageOrderly(
                        3,  // Max messages per batch (up to 16)
                        3   // Long polling timeout in seconds (up to 30)
                );
            } catch (Throwable e) {
                e.printStackTrace();
                try {
                    Thread.sleep(2000);
                } catch (InterruptedException e1) {
                    e1.printStackTrace();
                }
            }

            if (messages == null || messages.isEmpty()) {
                System.out.println(Thread.currentThread().getName() + ": no new message, continue!");
                continue;
            }

            // Process messages
            System.out.println("Receive " + messages.size() + " messages:");
            for (Message message : messages) {
                System.out.println(message);
                System.out.println("ShardingKey: " + message.getShardingKey()
                        + ", a:" + message.getProperties().get("a"));
            }

            // ACK consumed messages. If the broker does not receive an ACK before
            // the retry timeout, it redelivers the message.
            {
                List<String> handles = new ArrayList<String>();
                for (Message message : messages) {
                    handles.add(message.getReceiptHandle());
                }

                try {
                    consumer.ackMessage(handles);
                } catch (Throwable e) {
                    if (e instanceof AckMessageException) {
                        AckMessageException errors = (AckMessageException) e;
                        System.out.println("Ack message fail, requestId is:"
                                + errors.getRequestId() + ", fail handles:");
                        if (errors.getErrorMessages() != null) {
                            for (String errorHandle : errors.getErrorMessages().keySet()) {
                                System.out.println("Handle:" + errorHandle
                                        + ", ErrorCode:" + errors.getErrorMessages().get(errorHandle).getErrorCode()
                                        + ", ErrorMsg:" + errors.getErrorMessages().get(errorHandle).getErrorMessage());
                            }
                        }
                        continue;
                    }
                    e.printStackTrace();
                }
            }
        } while (true);
    }
}

Sıralı tüketimin çalışma biçimi

consumeMessageOrderly çağrıldığında broker şu davranışları uygular:

1. Parça düzeyinde kilitleme: Tüketici birden çok parçadan mesaj çekebilir ancak her parça içinde mesajlar gönderilme sırasına göre gelir.

2. Toplu onaylama kapısı: Tüketici, aynı parçadan sonraki toplu işi almadan önce mevcut toplu işteki tüm mesajları onaylamalıdır (ACK). Broker, yeniden deneme zaman aşımından önce onay almazsa onaylanmamış mesajı yeniden iletir.

3. Uzun oylama: Kullanılabilir mesaj yoksa broker isteği belirtilen oylama süresi boyunca (en fazla 30 saniye) bekletir ve bir mesaj geldiğinde hemen yanıt verir.