ApsaraMQ for RocketMQ:Envío y recepción de mensajes ordenados

Claves de fragmentación

Una clave de fragmentación identifica a qué partición pertenece un mensaje. Los mensajes que comparten la misma clave de fragmentación siempre se entregan en orden. Elija claves de fragmentación que coincidan con la agrupación de su negocio. Por ejemplo, use identificadores de pedido para que todas las actualizaciones de estado de un mismo pedido se procesen de forma secuencial, mientras que las actualizaciones de distintos pedidos se procesan en paralelo.

Evite usar una sola clave de fragmentación para todos los mensajes. Esta práctica dirige todo el tráfico a una única partición y elimina el paralelismo.

Para obtener más información sobre el modelo de mensajes ordenados, consulte Mensajes ordenados.

import java.util.List;
import org.apache.rocketmq.acl.common.AclClientRPCHook;
import org.apache.rocketmq.acl.common.SessionCredentials;
import org.apache.rocketmq.client.AccessChannel;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.RPCHook;
import org.apache.rocketmq.remoting.common.RemotingHelper;

public class RocketMQOrderProducer {
    private static RPCHook getAclRPCHook() {
        // Retrieve AccessKey ID and AccessKey secret from environment variables.
        // Set ALIBABA_CLOUD_ACCESS_KEY_ID and ALIBABA_CLOUD_ACCESS_KEY_SECRET
        // before running this example.
        return new AclClientRPCHook(new SessionCredentials(
            System.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"),
            System.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET")));
    }

    public static void main(String[] args) throws MQClientException {
        // Create a producer with message trace enabled.
        // To disable message trace, use:
        // DefaultMQProducer producer = new DefaultMQProducer("<YOUR_GROUP_ID>", getAclRPCHook());
        DefaultMQProducer producer = new DefaultMQProducer("<YOUR_GROUP_ID>", getAclRPCHook(), true, null);

        // Required for message trace on Alibaba Cloud.
        producer.setAccessChannel(AccessChannel.CLOUD);

        // Set the endpoint obtained from the ApsaraMQ for RocketMQ console.
        producer.setNamesrvAddr("<YOUR_ENDPOINT>");
        producer.start();

        for (int i = 0; i < 128; i++) {
            try {
                int orderId = i % 10;
                Message msg = new Message("<YOUR_TOPIC>",
                    "<YOUR_TAG>",
                    "OrderID188",
                    "Hello world".getBytes(RemotingHelper.DEFAULT_CHARSET));

                // Set the sharding key. Messages with the same sharding key
                // are delivered to the same partition in FIFO order.
                // For SDK 5.x and later, use:
                // msg.putUserProperty(MessageConst.PROPERTY_SHARDING_KEY, orderId + "");
                msg.putUserProperty("__SHARDINGKEY", orderId + "");

                SendResult sendResult = producer.send(msg, new MessageQueueSelector() {
                    @Override
                    public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                        // Route messages with the same orderId to the same queue.
                        Integer id = (Integer) arg;
                        int index = id % mqs.size();
                        return mqs.get(index);
                    }
                }, orderId);

                System.out.printf("%s%n", sendResult);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
        producer.shutdown();
    }
}

import java.util.List;
import org.apache.rocketmq.acl.common.AclClientRPCHook;
import org.apache.rocketmq.acl.common.SessionCredentials;
import org.apache.rocketmq.client.AccessChannel;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
import org.apache.rocketmq.client.consumer.rebalance.AllocateMessageQueueAveragely;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.RPCHook;

public class RocketMQOrderConsumer {
    private static RPCHook getAclRPCHook() {
        // Retrieve AccessKey ID and AccessKey secret from environment variables.
        // Set ALIBABA_CLOUD_ACCESS_KEY_ID and ALIBABA_CLOUD_ACCESS_KEY_SECRET
        // before running this example.
        return new AclClientRPCHook(new SessionCredentials(
            System.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"),
            System.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET")));
    }

    public static void main(String[] args) throws MQClientException {
        // Create a push consumer with message trace enabled.
        // To disable message trace, use:
        // DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("<YOUR_GROUP_ID>", getAclRPCHook(), null);
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("<YOUR_GROUP_ID>",
            getAclRPCHook(), new AllocateMessageQueueAveragely(), true, null);

        // Required for message trace on Alibaba Cloud.
        consumer.setAccessChannel(AccessChannel.CLOUD);

        // Set the endpoint obtained from the ApsaraMQ for RocketMQ console.
        consumer.setNamesrvAddr("<YOUR_ENDPOINT>");

        // Subscribe to ordered messages. Use "*" to receive all tags,
        // or specify a tag expression to filter messages.
        consumer.subscribe("<YOUR_TOPIC>", "*");

        // Start consuming from the earliest available offset.
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);

        // Register an orderly message listener to preserve FIFO order.
        consumer.registerMessageListener(new MessageListenerOrderly() {
            @Override
            public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
                context.setAutoCommit(true);
                System.out.printf("%s Receive New Messages: %s %n",
                    Thread.currentThread().getName(), msgs);

                // Return SUCCESS after processing.
                // To suspend and retry on failure, return:
                // ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT
                return ConsumeOrderlyStatus.SUCCESS;
            }
        });

        consumer.start();
        System.out.printf("Consumer Started.%n");
    }
}

Envío

• Utilice un solo hilo de producción para garantizar un orden estricto. Si varios productores o hilos envían mensajes de forma concurrente, el broker no puede asegurar el orden previsto por el negocio. Cuando el orden estricto es obligatorio, envíe todos los mensajes relacionados desde un único hilo.

• Elija claves de fragmentación de grano fino. Use identificadores de negocio, como identificadores de pedidos o de usuarios, para que distintos grupos de mensajes se procesen en paralelo. Evite emplear una sola clave de fragmentación para todos los mensajes, ya que esto dirige todo el tráfico a una única partición y elimina el paralelismo.

Consumo

• Gestione correctamente los errores de consumo. Si ocurre un error, devuelva ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT en lugar de SUCCESS. Esto suspende el proceso y reintenta la operación, lo que preserva el orden FIFO dentro de la partición.