ApsaraMQ for RocketMQ:Send and receive normal messages

Avant de commencer

1. Installez le SDK : configurez l'environnement de développement Node.js et installez le package @aliyunmq/mq-http-sdk. Pour plus d'informations, consultez la rubrique Préparer l'environnement.

2. Créez des ressources : créez une instance, un topic et un groupe de consommateurs dans la console ApsaraMQ for RocketMQ. Pour plus d'informations, consultez la rubrique Créer des ressources.

3. Configurez les identifiants : obtenez une paire AccessKey pour votre compte Alibaba Cloud et exportez les valeurs en tant que variables d'environnement. Pour plus d'informations, consultez la rubrique Créer une paire AccessKey.

      export ALIBABA_CLOUD_ACCESS_KEY_ID=<your-access-key-id>
      export ALIBABA_CLOUD_ACCESS_KEY_SECRET=<your-access-key-secret>

Espaces réservés

Remplacez les espaces réservés suivants dans l'exemple de code par vos valeurs réelles :

Envoyer des messages normaux

1. Créez un fichier nommé producer.js et ajoutez-y le code suivant :

      const {
        MQClient,
        MessageProperties
      } = require('@aliyunmq/mq-http-sdk');
   
      // HTTP endpoint of your ApsaraMQ for RocketMQ instance.
      const endpoint = "<http-endpoint>";
      // Read credentials from environment variables.
      const accessKeyId = process.env['ALIBABA_CLOUD_ACCESS_KEY_ID'];
      const accessKeySecret = process.env['ALIBABA_CLOUD_ACCESS_KEY_SECRET'];
   
      const client = new MQClient(endpoint, accessKeyId, accessKeySecret);
   
      // Topic to publish to. Create this topic in the console first.
      const topic = "<topic>";
      // Instance ID. Set to null or "" if the instance has no namespace.
      const instanceId = "<instance-id>";
   
      const producer = client.getProducer(instanceId, topic);
   
      (async function () {
        try {
          for (let i = 0; i < 4; i++) {
            const msgProps = new MessageProperties();
            // Custom property.
            msgProps.putProperty("a", i);
            // Message key for tracing or deduplication.
            msgProps.messageKey("MessageKey");
   
            // Publish the message with a body, tag, and properties.
            const res = await producer.publishMessage("hello mq.", "TagA", msgProps);
            console.log("Publish message: MessageID:%s, BodyMD5:%s",
              res.body.MessageId, res.body.MessageBodyMD5);
          }
        } catch (e) {
          // Handle failures: retry or persist the message for later redelivery.
          console.log(e);
        }
      })();

2. Exécutez le producteur :

      node producer.js

3. Vérifiez la sortie. Une exécution réussie affiche quatre lignes similaires à celles-ci :

      Publish message: MessageID:7F00000100246A4F0F2906B5BE4F0000, BodyMD5:4E69B245B8D0E47E5BE0B8520E8E608F
      Publish message: MessageID:7F00000100246A4F0F2906B5BE650001, BodyMD5:4E69B245B8D0E47E5BE0B8520E8E608F
      Publish message: MessageID:7F00000100246A4F0F2906B5BE780002, BodyMD5:4E69B245B8D0E47E5BE0B8520E8E608F
      Publish message: MessageID:7F00000100246A4F0F2906B5BE890003, BodyMD5:4E69B245B8D0E47E5BE0B8520E8E608F

Recevoir et acquitter des messages normaux

Le consommateur utilise le sondage long (long polling) : si aucun message n'est disponible, la requête reste bloquée sur le broker pendant un nombre de secondes configurable. Si un message arrive durant cette période, le broker répond immédiatement.

1. Créez un fichier nommé consumer.js et ajoutez-y le code suivant :

      const {
        MQClient
      } = require('@aliyunmq/mq-http-sdk');
   
      // HTTP endpoint of your ApsaraMQ for RocketMQ instance.
      const endpoint = "<http-endpoint>";
      // Read credentials from environment variables.
      const accessKeyId = process.env['ALIBABA_CLOUD_ACCESS_KEY_ID'];
      const accessKeySecret = process.env['ALIBABA_CLOUD_ACCESS_KEY_SECRET'];
   
      const client = new MQClient(endpoint, accessKeyId, accessKeySecret);
   
      // Topic to consume from.
      const topic = "<topic>";
      // Consumer group ID. Create this in the console first.
      const groupId = "<group-id>";
      // Instance ID. Set to null or "" if the instance has no namespace.
      const instanceId = "<instance-id>";
   
      const consumer = client.getConsumer(instanceId, topic, groupId);
   
      (async function () {
        while (true) {
          try {
            // consumeMessage(batchSize, pollingSeconds)
            //   batchSize:      max messages per request (1-16)
            //   pollingSeconds: long-polling timeout in seconds (max 30)
            const res = await consumer.consumeMessage(3, 3);
   
            if (res.code === 200) {
              console.log("Consume messages, requestId: %s", res.requestId);
   
              const handles = res.body.map((message) => {
                console.log(
                  "\tMessageId:%s, Tag:%s, PublishTime:%d, NextConsumeTime:%d, " +
                  "FirstConsumeTime:%d, ConsumedTimes:%d, Body:%s, Props:%j, MessageKey:%s, Prop-A:%s",
                  message.MessageId,
                  message.MessageTag,
                  message.PublishTime,
                  message.NextConsumeTime,
                  message.FirstConsumeTime,
                  message.ConsumedTimes,
                  message.MessageBody,
                  message.Properties,
                  message.MessageKey,
                  message.Properties.a
                );
                return message.ReceiptHandle;
              });
   
              // Acknowledge consumed messages. If the broker does not receive
              // an ACK before NextConsumeTime, the message is delivered again.
              const ackRes = await consumer.ackMessage(handles);
              if (ackRes.code !== 204) {
                // If the handle of the message times out, the broker fails to receive an ACK.
                console.log("Ack failed:");
                const failHandles = ackRes.body.map((error) => {
                  console.log(
                    "\tErrorHandle:%s, Code:%s, Reason:%s",
                    error.ReceiptHandle, error.ErrorCode, error.ErrorMessage
                  );
                  return error.ReceiptHandle;
                });
                handles.forEach((handle) => {
                  if (failHandles.indexOf(handle) < 0) {
                    console.log("\tSucHandle:%s", handle);
                  }
                });
              } else {
                console.log("Ack succeeded, requestId: %s\n\t", ackRes.requestId, handles.join(','));
              }
            }
          } catch (e) {
            if (e.Code && e.Code.indexOf("MessageNotExist") > -1) {
              // No messages available -- long polling continues on next iteration.
              console.log("No new messages. requestId: %s, Code: %s", e.RequestId, e.Code);
            } else {
              console.log(e);
            }
          }
        }
      })();

2. Exécutez le consommateur :

      node consumer.js

3. Vérifiez la sortie. Lorsque des messages sont disponibles, le consommateur affiche :

      Consume messages, requestId: B1A341B6F2A6XXXX
      	MessageId:7F00000100246A4F0F2906B5BE4F0000, Tag:TagA, PublishTime:1620000000000, NextConsumeTime:1620000030000, FirstConsumeTime:1620000000000, ConsumedTimes:1, Body:hello mq., Props:{"a":"0"}, MessageKey:MessageKey
      Ack succeeded, requestId: B1A341B6F2A6XXXX

   Lorsqu'aucun message n'est disponible, il affiche :

      No new messages. requestId: B1A341B6F2A6XXXX, Code: MessageNotExist

Méthodes API clés

Acquittement des messages

Après avoir traité un message, acquittez-le en appelant ackMessage avec le handle de réception du message. Si le broker ne reçoit pas d'acquittement avant l'échéance NextConsumeTime, le message est redistribué.

À chaque consommation d'un message, le broker attribue un nouveau handle de réception comportant un horodatage unique. Utilisez toujours le dernier handle lors de l'acquittement.

Échecs d'acquittement courants :