ApsaraMQ for RocketMQ:Send and receive ordered messages

Garantie d'ordre

Le broker détermine l'ordre des messages en fonction de la séquence d'envoi utilisée par un producteur unique ou un thread unique. Si plusieurs producteurs ou threads envoient des messages simultanément, le broker les ordonne selon leur heure d'arrivée, ce qui peut différer de l'ordre métier souhaité.

Exemple de code

Remplacez les espaces réservés suivants par vos valeurs réelles :

Pour obtenir le code complet, consultez le référentiel de code ApsaraMQ for RocketMQ.

package com.aliyun.openservices.ons.example.order;

import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.ONSFactory;
import com.aliyun.openservices.ons.api.PropertyKeyConst;
import com.aliyun.openservices.ons.api.SendResult;
import com.aliyun.openservices.ons.api.order.OrderProducer;

import java.util.Date;
import java.util.Properties;


public class ProducerClient {

    public static void main(String[] args) {
        Properties properties = new Properties();
        // Group ID created in the ApsaraMQ for RocketMQ console.
        properties.put(PropertyKeyConst.GROUP_ID, "<your-group-id>");
        // AccessKey ID for authentication.
        properties.put(PropertyKeyConst.AccessKey, "<your-access-key>");
        // AccessKey secret for authentication.
        properties.put(PropertyKeyConst.SecretKey, "<your-secret-key>");
        // TCP endpoint from the Instance Details page in the ApsaraMQ for RocketMQ console.
        properties.put(PropertyKeyConst.NAMESRV_ADDR, "<your-tcp-endpoint>");

        OrderProducer producer = ONSFactory.createOrderProducer(properties);
        // Call start() once before sending any messages.
        producer.start();

        for (int i = 0; i < 1000; i++) {
            String orderId = "biz_" + i % 10;
            Message msg = new Message(
                    // Topic for ordered messages.
                    "Order_global_topic",
                    // Tag for filtering on the broker side.
                    "TagA",
                    // Message body in binary format.
                    // Producer and consumer must agree on serialization and deserialization.
                    "send order global msg".getBytes()
            );
            // Message key -- a business identifier, ideally globally unique.
            // Use this key to look up the message in the ApsaraMQ for RocketMQ console.
            msg.setKey(orderId);

            // Sharding key -- determines the partition for ordered delivery.
            // Messages with the same sharding key are delivered in FIFO order.
            String shardingKey = String.valueOf(orderId);
            try {
                SendResult sendResult = producer.send(msg, shardingKey);
                if (sendResult != null) {
                    System.out.println(new Date() + " Send mq message success. Topic is:" + msg.getTopic() + " msgId is: " + sendResult.getMessageId());
                }
            } catch (Exception e) {
                // Handle failure: retry or persist the message for later resend.
                System.out.println(new Date() + " Send mq message failed. Topic is:" + msg.getTopic());
                e.printStackTrace();
            }
        }
        // Shut down the producer before exiting.
        // Skip this if the producer sends messages frequently.
        producer.shutdown();
    }

}

Comportement de nouvelle tentative

Lorsqu'un message ordonné ne parvient pas à être consommé, le broker bloque tous les messages suivants dans la même partition jusqu'à ce que le message actuel soit traité avec succès ou atteigne le nombre maximal de nouvelles tentatives. Ce blocage préserve l'ordre FIFO, mais peut retarder le traitement des autres messages de la partition.

Deux paramètres contrôlent le comportement de nouvelle tentative :

Définissez ces valeurs en fonction de votre tolérance aux retards de traitement. Une valeur SuspendTimeMillis plus faible entraîne des nouvelles tentatives plus rapides, mais augmente la charge du broker. Une valeur MaxReconsumeTimes plus faible débloque la partition plus tôt, mais risque d'ignorer des messages qui auraient pu aboutir avec davantage de tentatives.

Exemple de code

L'exemple de code suivant utilise les mêmes espaces réservés que l'exemple de producteur. Remplacez-les par vos valeurs réelles avant d'exécuter le code.

package com.aliyun.openservices.ons.example.order;

import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.ONSFactory;
import com.aliyun.openservices.ons.api.PropertyKeyConst;
import com.aliyun.openservices.ons.api.order.ConsumeOrderContext;
import com.aliyun.openservices.ons.api.order.MessageOrderListener;
import com.aliyun.openservices.ons.api.order.OrderAction;
import com.aliyun.openservices.ons.api.order.OrderConsumer;

import java.util.Properties;


public class ConsumerClient {

    public static void main(String[] args) {
        Properties properties = new Properties();
        // Consumer group ID created in the ApsaraMQ for RocketMQ console.
        properties.put(PropertyKeyConst.GROUP_ID, "<your-group-id>");
        // AccessKey ID for authentication.
        properties.put(PropertyKeyConst.AccessKey, "<your-access-key>");
        // AccessKey secret for authentication.
        properties.put(PropertyKeyConst.SecretKey, "<your-secret-key>");
        // TCP endpoint from the Instance Details page in the ApsaraMQ for RocketMQ console.
        properties.put(PropertyKeyConst.NAMESRV_ADDR, "<your-tcp-endpoint>");
        // Wait time (ms) before retrying a failed ordered message. Valid values: 10 to 30,000.
        properties.put(PropertyKeyConst.SuspendTimeMillis, "100");
        // Maximum retry attempts for a failed message.
        properties.put(PropertyKeyConst.MaxReconsumeTimes, "20");

        OrderConsumer consumer = ONSFactory.createOrderedConsumer(properties);

        consumer.subscribe(
                // Topic to subscribe to.
                "Order_global_topic",
                // Tag filter expression:
                //   "*"                       -- subscribe to all tags
                //   "TagA || TagB || TagC"     -- subscribe to TagA, TagB, or TagC only
                "*",
                new MessageOrderListener() {
                    /**
                     * Return OrderAction.Success after processing the message.
                     * Return OrderAction.Suspend if processing fails or throws
                     * an exception -- the broker will retry after SuspendTimeMillis.
                     */
                    @Override
                    public OrderAction consume(Message message, ConsumeOrderContext context) {
                        System.out.println(message);
                        return OrderAction.Success;
                    }
                });

        // Call start() once to begin consuming.
        consumer.start();
    }
}