ApsaraMQ for RocketMQ:Enviar e receber mensagens ordenadas

Como funciona

As mensagens ordenadas se dividem em duas categorias:

• Mensagens com ordenação global: todas as mensagens de um tópico seguem uma única sequência FIFO. Essa abordagem oferece a garantia de ordenação mais forte, porém limita o throughput a uma única partição.

• Mensagens com ordenação por partição: as mensagens são distribuídas entre partições por meio de uma chave de particionamento (sharding key). Dentro de cada partição, a ordem FIFO é mantida. Partições diferentes são consumidas de forma independente, o que possibilita maior throughput e preserva a ordem onde ela é necessária.

A sharding key identifica a qual partição uma mensagem pertence. Mensagens com a mesma sharding key são sempre entregues na mesma partição e consumidas em ordem. Uma sharding key é diferente de uma chave de mensagem (message key).

Pré-requisitos

Antes de começar, certifique-se de que as seguintes etapas foram concluídas:

• Instale o SDK para Java. Para mais informações, consulte Defina seu ambiente Java.

• Crie os recursos que deseja especifique no código por meio do console do ApsaraMQ for RocketMQ. Os recursos incluem instâncias, tópicos e grupos de consumidores. Para mais informações, consulte Crie recursos.

• Obtenha o par de AccessKey da sua conta Alibaba Cloud. Para mais informações, consulte Crie um AccessKey.

Enviar mensagens ordenadas

O exemplo a seguir envia oito mensagens ordenadas em duas partições usando sharding keys.

Substitua os espaços reservados pelos seus valores reais:

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

Receber mensagens ordenadas

O exemplo a seguir consome mensagens ordenadas com consumeMessageOrderly, que emprega long polling e garante a ordem FIFO no nível da partição.

Substitua os espaços reservados pelos seus valores reais:

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

Observações de uso

• Tópicos de tipo único: cada tópico suporta apenas um tipo de mensagem. Não misture mensagens ordenadas com mensagens normais no mesmo tópico.

• Produtor único para ordenação rigorosa: envie todas as mensagens que exigem ordenação relativa a partir de um único produtor em um único thread. O uso de múltiplos produtores ou threads concorrentes invalida a garantia de ordenação.

• Design da sharding key: escolha uma sharding key que agrupe mensagens relacionadas sem criar partições sobrecarregadas. Por exemplo, use um id de pedido para manter todos os eventos de um mesmo pedido em sequência, em vez de um id de cliente que pode concentrar mensagens demais em uma única partição.

• Confirmação imediata: a confirmação tardia bloqueia a entrega de mensagens subsequentes para toda a partição. Processe e confirme (ACK) cada lote o mais rápido possível.

• Expiração do receipt handle: cada receipt handle de mensagem possui um timestamp exclusivo. Se o handle expirar antes do ACK, a confirmação falha e o broker reenvia a mensagem.

Consulte também

• Ordered messages: garantias de ordenação, tipos de ordenação e conceitos detalhados

• Crie recursos: configure instâncias, tópicos e grupos de consumidores no console