ApsaraMQ for RocketMQ:Send and receive normal messages

Antes de começar

1. Instale o SDK -- Configure o ambiente de desenvolvimento Node.js e instale o pacote @aliyunmq/mq-http-sdk. Para mais informações, consulte Preparar o ambiente.

2. Crie recursos -- Crie uma instância, um tópico e um grupo de consumidores no console do ApsaraMQ for RocketMQ. Para mais detalhes, consulte Criar recursos.

3. Configure as credenciais -- Obtenha um par de AccessKey da sua conta Alibaba Cloud e exporte os valores como variáveis de ambiente. Para mais informações, consulte Criar um par de AccessKey.

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

Espaços reservados

Substitua estes espaços reservados no código de exemplo pelos seus valores reais:

Enviar mensagens normais

1. Crie um arquivo chamado producer.js e adicione o seguinte 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. Execute o produtor:

      node producer.js

3. Verifique a saída. Uma execução bem-sucedida exibe quatro linhas semelhantes a:

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

Receber e confirmar mensagens normais

O consumidor usa long polling: se nenhuma mensagem estiver disponível, a solicitação fica bloqueada no broker por um número configurável de segundos. Caso uma mensagem chegue nesse intervalo, o broker responde imediatamente.

1. Crie um arquivo chamado consumer.js e adicione o seguinte 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. Execute o consumidor:

      node consumer.js

3. Verifique a saída. Quando há mensagens disponíveis, o consumidor exibe:

      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

   Quando não há mensagens disponíveis, a saída é:

      No new messages. requestId: B1A341B6F2A6XXXX, Code: MessageNotExist

Principais métodos da API

Confirmação de mensagens

Após processar uma mensagem, confirme-a chamando ackMessage com o recibo (receipt handle) da mensagem. Se o broker não receber a confirmação antes do prazo NextConsumeTime, a mensagem será entregue novamente.

Sempre que uma mensagem é consumida, o broker atribui um novo recibo com um timestamp único. Use sempre o recibo mais recente ao confirmar.

Falhas comuns de ACK: