ApsaraMQ for RocketMQ:Send and receive normal messages

Başlamadan önce

1. SDK'yı yükleyin -- Node.js geliştirme ortamını kurun ve @aliyunmq/mq-http-sdk paketini yükleyin. Daha fazla bilgi için bkz. Ortamı hazırlama.

2. Kaynakları oluşturun -- ApsaraMQ for RocketMQ konsolunda bir örnek, bir konu ve bir tüketici grubu oluşturun. Daha fazla bilgi için bkz. Kaynakları oluşturma.

3. Kimlik bilgilerini yapılandırın -- Alibaba Cloud hesabınız için bir AccessKey çifti edinin ve değerleri ortam değişkenleri olarak dışa aktarın. Daha fazla bilgi için bkz. AccessKey çifti oluşturma.

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

Yer tutucular

Örnek kodda bulunan aşağıdaki yer tutucuları kendi gerçek değerlerinizle değiştirin:

Normal mesaj gönderme

1. producer.js adlı bir dosya oluşturun ve aşağıdaki kodu ekleyin:

      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. Üreticiyi çalıştırın:

      node producer.js

3. Çıktıyı doğrulayın. Başarılı bir çalıştırma sonucunda aşağıdakine benzer dört satır görüntülenir:

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

Normal mesajları alma ve onaylama

Tüketici uzun oylama (long polling) kullanır: Mesaj mevcut değilse istek, aracı sunucuda yapılandırılabilir saniye sayısı kadar beklemeye geçer. Bu süre zarfında bir mesaj gelirse aracı sunucu hemen yanıt verir.

1. consumer.js adlı bir dosya oluşturun ve aşağıdaki kodu ekleyin:

      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. Tüketiciyi çalıştırın:

      node consumer.js

3. Çıktıyı doğrulayın. Mesajlar mevcut olduğunda tüketici şunları yazdırır:

      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

   Mesaj mevcut olmadığında ise şunları yazdırır:

      No new messages. requestId: B1A341B6F2A6XXXX, Code: MessageNotExist

Temel API yöntemleri

Mesaj onaylama

Bir mesajı işledikten sonra mesajın alındı tanıtıcısını kullanarak ackMessage çağrısı yaparak onaylayın. Aracı sunucu NextConsumeTime son tarihinden önce bir onay almazsa mesaj yeniden teslim edilir.

Her mesaj tüketildiğinde aracı sunucu benzersiz bir zaman damgasına sahip yeni bir alındı tanıtıcısı atar. Onaylarken her zaman en güncel tanıtıcıyı kullanın.

Yaygın ACK hataları: