Send and receive transactional messages - ApsaraMQ for RocketMQ: Central de Documentação do Alibaba Cloud

Funcionamento das mensagens transacionais

Uma mensagem transacional passa por um processo de commit em duas fases que vincula uma transação local à entrega da mensagem. As etapas abaixo descrevem a interação completa entre o produtor, o broker e o consumidor:

O produtor envia uma half message ao broker. Essa mensagem é armazenada, mas permanece invisível para os consumidores.
O broker persiste a half message e retorna um reconhecimento (ACK) com um receipt handle.
O produtor executa a transação local (por exemplo, atualizando um banco de dados).
Com base no resultado da transação local, o produtor faz o commit ou rollback da half message usando o receipt handle.
Se o produtor confirmar a half message, o broker a marca como entregável. Os consumidores podem então recebê-la.
Caso o produtor faça o rollback da half message, o broker a descarta. Os consumidores nunca a visualizam.
Quando o broker não recebe uma decisão de commit ou rollback (devido a falha de rede ou reinício da aplicação), ele inicia uma verificação de transação após o intervalo TransCheckImmunityTime.
O produtor recebe a solicitação de verificação e reavalia o status da transação local.
O produtor envia a decisão de commit ou rollback ao broker. Se a decisão ainda for desconhecida, o broker tenta novamente a cada 10 segundos por até 24 horas.

图片1.png

Para mais informações, consulte Mensagens transacionais.

Considerações sobre o SDK HTTP

O SDK cliente HTTP usa um modelo baseado em pull para verificações de transação. Diferentemente dos SDKs nativos do RocketMQ, onde o broker envia solicitações de verificação ao produtor, o SDK HTTP exige que o produtor consulte periodicamente as half messages não confirmadas com consumeHalfMessage. Execute esse loop de consulta em um processo ou thread dedicado para garantir a resolução confiável das transações.

Pré-requisitos

Instale o SDK para Node.js. Para mais informações, consulte Preparar o ambiente.
Crie os recursos necessários no console do ApsaraMQ for RocketMQ: uma instância, um tópico e um grupo de consumidores. Para mais detalhes, veja Criar recursos.
Obtenha um par de AccessKey para sua conta Alibaba Cloud. Consulte Criar um par de AccessKey para instruções.

Parâmetros principais

Antes de executar o código de exemplo, substitua os placeholders abaixo pelos seus valores reais:

Placeholder	Descrição	Onde encontrar
`${HTTP_ENDPOINT}`	Endpoint HTTP da sua instância	Página Instance Details > seção HTTP Endpoint no console do ApsaraMQ for RocketMQ
`${TOPIC}`	Nome do tópico	Criado no console do ApsaraMQ for RocketMQ
`${GROUP_ID}`	ID do grupo de consumidores	Criado no console do ApsaraMQ for RocketMQ
`${INSTANCE_ID}`	ID da instância (defina como `null` ou `""` se a instância não tiver namespace)	Página Instance Details no console do ApsaraMQ for RocketMQ

As credenciais AccessKey são lidas das variáveis de ambiente:

ALIBABA_CLOUD_ACCESS_KEY_ID
ALIBABA_CLOUD_ACCESS_KEY_SECRET

Enviar mensagens transacionais

O código do produtor possui duas partes: envio de mensagens transacionais e consulta de half messages não confirmadas.

API do produtor transacional

Método	Finalidade
`client.getTransProducer(instanceId, topic, groupId)`	Crie um produtor transacional
`mqTransProducer.publishMessage(body, tag, msgProps)`	Enviar uma half message
`mqTransProducer.commit(receiptHandle)`	Confirmar uma half message
`mqTransProducer.rollback(receiptHandle)`	Reverter uma half message
`mqTransProducer.consumeHalfMessage(numOfMessages, waitSeconds)`	Consultar half messages não confirmadas

TransCheckImmunityTime

O parâmetro TransCheckImmunityTime controla quanto tempo o broker aguarda antes da primeira verificação de transação após o envio de uma half message.

Unidade: segundos
Valores válidos: 10 a 300
Comportamento após a primeira verificação: Se a half message permanecer sem confirmação, o broker verifica a cada 10 segundos por até 24 horas.

Código de exemplo

O exemplo a seguir envia quatro mensagens transacionais. A primeira mensagem é confirmada imediatamente após a publicação. As três restantes são resolvidas durante o loop de consulta de half messages com base na propriedade personalizada a:

a=1: confirmada na primeira verificação
a=2: confirmada após a segunda verificação (ConsumedTimes > 1)
a=3: revertida

const {
  MQClient,
  MessageProperties
} = require('@aliyunmq/mq-http-sdk');

// HTTP endpoint from the Instance Details page in the ApsaraMQ for RocketMQ console.
const endpoint = "${HTTP_ENDPOINT}";
// AccessKey credentials from environment variables.
const accessKeyId = process.env['ALIBABA_CLOUD_ACCESS_KEY_ID'];
const accessKeySecret = process.env['ALIBABA_CLOUD_ACCESS_KEY_SECRET'];

var client = new MQClient(endpoint, accessKeyId, accessKeySecret);

// Topic created in the ApsaraMQ for RocketMQ console.
const topic = "${TOPIC}";
// Consumer group ID created in the ApsaraMQ for RocketMQ console.
const groupId = "${GROUP_ID}";
// Instance ID. Set to null or "" if the instance has no namespace.
const instanceId = "${INSTANCE_ID}";

const mqTransProducer = client.getTransProducer(instanceId, topic, groupId);

async function processTransResult(res, msgId) {
    if (!res) {
        return;
    }
    if (res.code != 204) {
      // Commit or rollback failed. The receipt handle may have expired
      // if TransCheckImmunityTime or the consumeHalfMessage timeout elapsed.
      console.log("Commit/Rollback Message Fail:");
      const failHandles = res.body.map((error) => {
        console.log("\tErrorHandle:%s, Code:%s, Reason:%s\n", error.ReceiptHandle, error.ErrorCode, error.ErrorMessage);
        return error.ReceiptHandle;
      });
    } else {
      console.log("Commit/Rollback Message suc!!! %s", msgId);
    }
}

var halfMessageCount = 0;
var halfMessageConsumeCount = 0;

(async function(){
  try {
    // Send four transactional messages.
    for(var i = 0; i < 4; i++) {
      let res;
      msgProps = new MessageProperties();
      // Custom property used by the half-message polling loop
      // to decide whether to commit or roll back.
      msgProps.putProperty("a", i);
      msgProps.messageKey("MessageKey");
      // First transaction check delay: 10 seconds. Valid values: 10-300.
      // After the first check, the broker rechecks every 10 seconds for up to 24 hours.
      msgProps.transCheckImmunityTime(10);
      res = await mqTransProducer.publishMessage("hello mq.", "tagA", msgProps);
      console.log("Publish message: MessageID:%s,BodyMD5:%s,Handle:%s", res.body.MessageId, res.body.MessageBodyMD5, res.body.ReceiptHandle);
      if (res && i == 0) {
          // Commit the first message immediately using its receipt handle.
          const msgId = res.body.MessageId;
          res = await mqTransProducer.commit(res.body.ReceiptHandle);
          console.log("Commit msg when publish, %s", msgId);
        processTransResult(res, msgId);
      }
    }
  } catch(e) {
    // Add retry or persistence logic for failed messages.
    console.log(e)
  }
})();

// Run a separate loop to poll and resolve uncommitted half messages.
(async function() {
  // Poll for half messages, similar to consuming normal messages.
  while(halfMessageCount < 3 && halfMessageConsumeCount < 15) {
    try {
        halfMessageConsumeCount++;
      res = await mqTransProducer.consumeHalfMessage(3, 3);
      if (res.code == 200) {
        console.log("Consume Messages, requestId:%s", res.requestId);
        res.body.forEach(async (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);

          var propA = message.Properties && message.Properties.a ? parseInt(message.Properties.a) : 0;
                  var opResp;
                  if (propA == 1 || (propA == 2 && message.ConsumedTimes > 1)) {
                      opResp = await mqTransProducer.commit(message.ReceiptHandle);
                      console.log("Commit msg when check half, %s", message.MessageId);
                      halfMessageCount++;
                  } else if (propA == 3) {
                      opResp = await mqTransProducer.rollback(message.ReceiptHandle);
                      console.log("Rollback msg when check half, %s", message.MessageId);
                      halfMessageCount++;
                  }
                  processTransResult(opResp, message.MessageId);
        });
      }
    } catch(e) {
      if (e.Code && e.Code.indexOf("MessageNotExist") > -1) {
        // No half messages available. Long polling continues.
        console.log("Consume Transaction Half msg: no new message, RequestId:%s, Code:%s", e.RequestId, e.Code);
      } else {
        console.log(e);
      }
    }
  }
})();

Como funciona o código de exemplo

O código executa duas funções assíncronas simultâneas:

Loop de envio: Publica quatro half messages com a tag tagA e uma propriedade personalizada a definida com o índice do loop (0-3). A primeira mensagem (a=0) é confirmada imediatamente após a publicação.

Loop de consulta de half messages: Consulta half messages não confirmadas e as resolve com base na propriedade a: O loop termina após resolver 3 half messages ou após 15 tentativas de consulta.

Valor da propriedade `a`	Ação	Condição
0	Já confirmada durante o envio	N/A
1	Confirmar	Primeira verificação
2	Confirmar	Após nova tentativa (`ConsumedTimes > 1`)
3	Reverter	Primeira verificação

Códigos de resposta

Código	Significado
`200`	Mensagens recuperadas
`204`	Commit, rollback ou ACK bem-sucedido
Diferente de `204`	Falha na operação (o receipt handle pode ter expirado)

Receber mensagens transacionais

Os consumidores recebem mensagens transacionais confirmadas da mesma forma que mensagens normais. O long polling evita a espera ocupada suspendendo a solicitação até que uma mensagem chegue ou o tempo limite seja atingido.

API do consumidor

Método	Finalidade
`client.getConsumer(instanceId, topic, groupId)`	Crie um consumidor
`consumer.consumeMessage(numOfMessages, waitSeconds)`	Puxar mensagens usando long polling
`consumer.ackMessage(handles)`	Reconhecer mensagens consumidas

Parâmetros

Parâmetro	Descrição	Valor máximo
`numOfMessages`	Máximo de mensagens a recuperar por solicitação	16
`waitSeconds`	Duração do long polling em segundos	30

Código de exemplo

const {
  MQClient
} = require('@aliyunmq/mq-http-sdk');

// HTTP endpoint from the Instance Details page in the ApsaraMQ for RocketMQ console.
const endpoint = "${HTTP_ENDPOINT}";
// AccessKey credentials from environment variables.
const accessKeyId = process.env['ALIBABA_CLOUD_ACCESS_KEY_ID'];
const accessKeySecret = process.env['ALIBABA_CLOUD_ACCESS_KEY_SECRET'];

var client = new MQClient(endpoint, accessKeyId, accessKeySecret);

// Topic created in the ApsaraMQ for RocketMQ console.
const topic = "${TOPIC}";
// Consumer group ID created in the ApsaraMQ for RocketMQ console.
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(){
  // Consume messages in a loop.
  while(true) {
    try {
      // Long polling: if no message is available, the request suspends on the broker
      // until a message arrives or the timeout elapses.
      res = await consumer.consumeMessage(
          3, // Max messages per request (up to 16)
          3  // Long polling timeout in seconds (up to 30)
          );

      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;
        });

        // ACK all processed messages. If the broker does not receive an ACK
        // before NextConsumeTime, it redelivers the message with a new receipt handle.
        res = await consumer.ackMessage(handles);
        if (res.code != 204) {
          // ACK failed. The receipt handle may have expired.
          console.log("Ack Message Fail:");
          const failHandles = res.body.map((error)=>{
            console.log("\tErrorHandle:%s, Code:%s, Reason:%s\n", error.ReceiptHandle, error.ErrorCode, error.ErrorMessage);
            return error.ReceiptHandle;
          });
          handles.forEach((handle)=>{
            if (failHandles.indexOf(handle) < 0) {
              console.log("\tSucHandle:%s\n", handle);
            }
          });
        } else {
          console.log("Ack Message suc, RequestId:%s\n\t", res.requestId, handles.join(','));
        }
      }
    } catch(e) {
      if (e.Code.indexOf("MessageNotExist") > -1) {
        // No messages available. Long polling continues.
        console.log("Consume Message: no new message, RequestId:%s, Code:%s", e.RequestId, e.Code);
      } else {
        console.log(e);
      }
    }
  }
})();

Funcionamento do consumidor

O consumidor executa um loop infinito que:

Recupera até 3 mensagens por solicitação com tempo limite de long polling de 3 segundos.
Processa cada mensagem e coleta os receipt handles.
Envia um ACK para todas as mensagens processadas.

Se o consumidor não enviar o ACK de uma mensagem antes do NextConsumeTime, o broker a reentrega. Cada reentrega gera um novo receipt handle.

Quando não há mensagens disponíveis, o código de erro MessageNotExist é retornado. Isso é normal no modo de long polling — o loop continua aguardando.

Observações de uso

Faça commit ou rollback prontamente. Half messages não resolvidas consomem armazenamento do broker. Evite retornar um status desconhecido da verificação de transação local, a menos que a transação esteja realmente ainda em andamento.
Lide com a expiração do receipt handle. Os receipt handles têm um período de validade limitado. Se um commit ou rollback falhar com uma resposta diferente de 204, o handle pode ter expirado. O broker aciona automaticamente uma nova verificação de transação.
Execute o loop de consulta de half messages de forma confiável. Em produção, execute o loop consumeHalfMessage como um processo dedicado de longa duração para garantir que as half messages não resolvidas sejam tratadas em tempo hábil.

Próximos passos

Saiba como funcionam as mensagens transacionais: Mensagens transacionais
Configure seu ambiente de desenvolvimento: Preparar o ambiente
Crie recursos de mensageria: Criar recursos