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

Antes de empezar

1. Instale el SDK: configure el entorno de desarrollo de Node.js e instale el paquete @aliyunmq/mq-http-sdk. Para obtener más información, consulte Preparar el entorno.

2. Cree los recursos: cree una instancia, un tema y un grupo de consumidores en la consola de ApsaraMQ for RocketMQ. Para obtener más información, consulte Crear recursos.

3. Configure las credenciales: obtenga un par de AccessKey para su cuenta de Alibaba Cloud y exporte los valores como variables de entorno. Para obtener más información, consulte Crear un par de AccessKey.

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

Marcadores de posición

Reemplace estos marcadores de posición en el código de ejemplo con sus valores reales:

Envío de mensajes normales

1. Cree un archivo llamado producer.js y agregue a continuación el código:

      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. Ejecute el productor:

      node producer.js

3. Compruebe la salida. Una ejecución exitosa muestra cuatro líneas similares a a continuación:

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

Recepción y confirmación de mensajes normales

El consumidor utiliza sondeo largo: si no hay mensajes disponibles, la solicitud se bloquea en el broker durante un número configurable de segundos. Si llega un mensaje dentro de esa ventana, el broker responde inmediatamente.

1. Cree un archivo llamado consumer.js y agregue a continuación el código:

      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. Ejecute el consumidor:

      node consumer.js

3. Compruebe la salida. Cuando hay mensajes disponibles, el consumidor muestra a continuación:

      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

   Cuando no hay mensajes disponibles, se muestra a continuación:

      No new messages. requestId: B1A341B6F2A6XXXX, Code: MessageNotExist

Métodos clave de la API

Confirmación de mensajes

Después de procesar un mensaje, confírmelo mediante una llamada a ackMessage con el identificador de recibo del mensaje. Si el broker no recibe una confirmación antes de la fecha límite de NextConsumeTime, el mensaje se vuelve a entregar.

Cada vez que se consume un mensaje, el broker asigna un nuevo identificador de recibo con una marca de tiempo única. Utilice siempre el identificador más reciente al confirmar.

Errores comunes de confirmación: