O ApsaraMQ for RocketMQ utiliza mensagens transacionais para garantir a consistência das transações entre serviços distribuídos. Uma mensagem transacional vincula a entrega da mensagem ao resultado de uma operação de negócio local: os consumidores recebem a mensagem somente após o commit da transação local.
O código de exemplo em Python a seguir usa o SDK cliente HTTP para demonstrar três operações: envio de mensagens transacionais, tratamento de verificações de status de transação e consumo de mensagens com commit confirmado.
Como funciona
As mensagens transacionais seguem um protocolo de commit em duas fases semelhante ao X/Open XA.
O produtor, o broker e o consumidor interagem da seguinte forma:
O produtor envia uma half message ao broker. Essa mensagem é persistida no broker, mas marcada como temporariamente não entregável aos consumidores.
O broker confirma o recebimento da half message e retorna um receipt handle.
O produtor executa a transação local (por exemplo, uma gravação no banco de dados).
Com base no resultado da transação local, o produtor envia uma confirmação de segunda fase para fazer commit (entregar aos consumidores) ou rollback (descartar a mensagem).
Caso o broker não receba a confirmação de segunda fase — devido a uma falha de rede ou reinicialização do produtor —, ele inicia uma verificação de status da transação consultando o produtor.
O produtor reavalia o estado da transação local e responde com commit, rollback ou desconhecido.
O broker repete a verificação a cada 10 segundos por até 24 horas.
Para mais informações, consulte Mensagens transacionais.
Pré-requisitos
Antes de começar, certifique-se de ter:
O SDK cliente HTTP para Python instalado — consulte Preparar o ambiente
Uma instância do ApsaraMQ for RocketMQ, um tópico e um grupo de consumidores — consulte Criar recursos
Um par de AccessKey para sua conta Alibaba Cloud — consulte Criar um par de AccessKey
Conceitos principais
|
Termo |
Descrição |
|
Half message |
Mensagem enviada ao broker, mas marcada como temporariamente não entregável. Ela torna-se visível aos consumidores apenas após o produtor confirmar o commit da transação. |
|
Verificação de status da transação |
Quando o broker não recebe uma confirmação de commit ou rollback, ele consulta proativamente o produtor sobre o resultado da transação. |
|
|
O atraso (em segundos) antes que o broker inicie a primeira verificação de status após o envio de uma half message. Valores válidos: 10 a 300. |
|
Receipt handle |
Token exclusivo atribuído sempre que uma mensagem é consumida. Utilizado para confirmar commit, fazer rollback ou reconhecer mensagens. |
Enviar mensagens transacionais
O envio de uma mensagem transacional envolve três partes:
Inicializar o cliente e publicar half messages — O produtor envia mensagens que o broker retém como half messages.
Tratar verificações de status da transação — Uma thread em segundo plano verifica periodicamente as half messages pendentes e confirma o commit ou faz rollback de cada uma com base no resultado da transação local.
Confirmar commit imediatamente após a publicação — Opcionalmente, confirme o commit de uma half message logo após publicá-la, usando o receipt handle obtido na resposta da publicação.
O código a seguir demonstra todas as três partes. Substitua os placeholders pelos seus valores reais:
|
Placeholder |
Descrição |
Exemplo |
|
|
Endpoint HTTP obtido na página Instance Details no console do ApsaraMQ for RocketMQ |
|
|
|
Tópico criado no console do ApsaraMQ for RocketMQ |
|
|
|
ID do grupo de consumidores criado no console do ApsaraMQ for RocketMQ |
|
|
|
ID da instância. Se a instância possuir um namespace, especifique o ID. Caso contrário, use uma string vazia |
|
#!/usr/bin/env python
# coding=utf8
import sys
import os
import time
import threading
from mq_http_sdk.mq_exception import MQExceptionBase
from mq_http_sdk.mq_producer import *
from mq_http_sdk.mq_client import *
# ---------------------------------------------------------------------------
# 1. Initialize the client
# ---------------------------------------------------------------------------
# Get the HTTP endpoint from the Instance Details page in the
# ApsaraMQ for RocketMQ console.
mq_client = MQClient(
"<your-http-endpoint>",
# Retrieve credentials from environment variables to avoid
# hardcoding sensitive information.
os.environ['ALIBABA_CLOUD_ACCESS_KEY_ID'],
os.environ['ALIBABA_CLOUD_ACCESS_KEY_SECRET']
)
topic_name = "<your-topic>"
group_id = "<your-group-id>"
instance_id = "<your-instance-id>"
# Number of transactional messages to send in this example.
msg_count = 4
# ---------------------------------------------------------------------------
# 2. Define error handling for commit/rollback failures
# ---------------------------------------------------------------------------
def process_trans_error(exp):
"""Handle commit or rollback failures.
A failure occurs when the operation takes longer than the
TransCheckImmunityTime value or the consume_half_message timeout
(10 seconds in this example).
"""
print("\nCommit/Roll Transaction Message Fail! Exception:%s" % exp)
if exp.sub_errors:
for sub_error in exp.sub_errors:
print("\tErrorHandle:%s,ErrorCode:%s,ErrorMsg:%s" %
(sub_error["ReceiptHandle"], sub_error["ErrorCode"],
sub_error["ErrorMessage"]))
# ---------------------------------------------------------------------------
# 3. Set up the transaction status checker (background thread)
# ---------------------------------------------------------------------------
# The broker queries this checker when it has not received a commit or
# rollback for a half message. Run it in a separate thread so it can
# process checks while the producer continues sending messages.
class TransactionCheckThread(threading.Thread):
def __init__(self):
threading.Thread.__init__(self)
self.count = 0
# Create a separate client for the checker thread.
self.mq_client = MQClient(
"<your-http-endpoint>",
os.environ['ALIBABA_CLOUD_ACCESS_KEY_ID'],
os.environ['ALIBABA_CLOUD_ACCESS_KEY_SECRET']
)
self.trans_producer = self.mq_client.get_trans_producer(
instance_id, topic_name, group_id
)
def run(self):
while True:
if self.count == 3:
break
try:
# Poll for half messages awaiting a transaction decision.
# Arguments: batch size = 1, wait timeout = 3 seconds.
half_msgs = self.trans_producer.consume_half_message(1, 3)
for half_msg in half_msgs:
print("Receive Half Message, MessageId: %s\n"
"MessageBodyMD5: %s\n"
"MessageTag: %s\n"
"ConsumedTimes: %s\n"
"PublishTime: %s\n"
"Body: %s\n"
"NextConsumeTime: %s\n"
"ReceiptHandle: %s\n"
"Properties: %s" %
(half_msg.message_id,
half_msg.message_body_md5,
half_msg.message_tag,
half_msg.consumed_times,
half_msg.publish_time,
half_msg.message_body,
half_msg.next_consume_time,
half_msg.receipt_handle,
half_msg.properties))
# Decide whether to commit or roll back based on local
# transaction results. In this example, the decision
# depends on the custom property "a":
# a == 1 -> commit immediately
# a == 2 -> commit on the second check (simulates
# delayed local transaction completion)
# a == 3 -> rollback (local transaction failed)
# other -> unknown (check again later)
a = int(half_msg.get_property("a"))
try:
if a == 1:
self.trans_producer.commit(half_msg.receipt_handle)
self.count += 1
print("------>commit")
elif a == 2 and half_msg.consumed_times > 1:
self.trans_producer.commit(half_msg.receipt_handle)
self.count += 1
print("------>commit")
elif a == 3:
self.trans_producer.rollback(half_msg.receipt_handle)
self.count += 1
print("------>rollback")
else:
print("------>unknown")
except MQExceptionBase as rec_commit_roll_e:
process_trans_error(rec_commit_roll_e)
except MQExceptionBase as half_e:
if half_e.type == "MessageNotExist":
print("No half message! RequestId: %s" % half_e.req_id)
continue
print("Consume half message Fail! Exception:%s\n" % half_e)
break
# ---------------------------------------------------------------------------
# 4. Start the checker thread and send transactional messages
# ---------------------------------------------------------------------------
check_thread = TransactionCheckThread()
check_thread.setDaemon(True)
check_thread.start()
try:
trans_producer = mq_client.get_trans_producer(
instance_id, topic_name, group_id
)
for i in range(msg_count):
msg = TopicMessage(
"I am test message %s." % i, # Message body
"tagA" # Message tag
)
# Set a custom property used by the checker to simulate
# different transaction outcomes.
msg.put_property("a", i)
msg.set_message_key("MessageKey")
# TransCheckImmunityTime: delay (in seconds) before the first
# transaction status check. Valid values: 10 to 300.
# After the first check, the broker retries every 10 seconds
# for up to 24 hours.
msg.set_trans_check_immunity_time(10)
re_msg = trans_producer.publish_message(msg)
print("Publish Transaction Message Succeed. MessageID:%s, "
"BodyMD5:%s, Handle:%s" %
(re_msg.message_id, re_msg.message_body_md5,
re_msg.receipt_handle))
time.sleep(1)
# For the first message (i == 0), commit immediately using
# the receipt handle returned by publish_message.
if i == 0:
try:
trans_producer.commit(re_msg.receipt_handle)
except MQExceptionBase as pub_commit_roll_e:
process_trans_error(pub_commit_roll_e)
except MQExceptionBase as pub_e:
if pub_e.type == "TopicNotExist":
print("Topic not exist, please create it.")
sys.exit(1)
print("Publish Message Fail. Exception:%s" % pub_e)
# Wait for the checker thread to finish processing.
while True:
if not check_thread.is_alive():
break
time.sleep(1)O que o exemplo faz
O produtor envia quatro mensagens transacionais com a propriedade personalizada a definida como 0, 1, 2 e 3. Cada valor aciona um resultado diferente para a transação:
|
**Propriedade |
Comportamento do verificador |
Resultado |
|
0 |
O produtor confirma o commit imediatamente após a publicação (ignora o verificador). |
Mensagem entregue aos consumidores. |
|
1 |
O verificador confirma o commit na primeira verificação de status. |
Mensagem entregue aos consumidores. |
|
2 |
O verificador retorna desconhecido na primeira verificação e confirma o commit na segunda. |
Mensagem entregue após um atraso. |
|
3 |
O verificador faz rollback. |
Mensagem descartada, nunca entregue. |
Inscrever-se em mensagens transacionais
Mensagens transacionais com commit confirmado são consumidas da mesma forma que mensagens normais. O consumidor recebe apenas mensagens cujas transações foram confirmadas.
O código a seguir usa long polling para buscar mensagens em lotes e reconhecê-las.
import os
import time
from mq_http_sdk.mq_exception import MQExceptionBase
from mq_http_sdk.mq_consumer import *
from mq_http_sdk.mq_client import *
# Initialize the consumer client.
mq_client = MQClient(
"<your-http-endpoint>",
os.environ['ALIBABA_CLOUD_ACCESS_KEY_ID'],
os.environ['ALIBABA_CLOUD_ACCESS_KEY_SECRET']
)
topic_name = "<your-topic>"
group_id = "<your-group-id>"
instance_id = "<your-instance-id>"
consumer = mq_client.get_consumer(instance_id, topic_name, group_id)
# Long polling timeout in seconds. Maximum: 30.
wait_seconds = 3
# Maximum messages per batch. Maximum: 16.
batch = 3
print("Consume Message From Topic: %s, Consumer: %s, WaitSeconds: %s\n" %
(topic_name, group_id, wait_seconds))
while True:
try:
# Long polling: if no message is available, the request is held
# on the broker for up to wait_seconds before returning.
recv_msgs = consumer.consume_message(batch, wait_seconds)
for msg in recv_msgs:
print("Receive, MessageId: %s\n"
"MessageBodyMD5: %s\n"
"MessageTag: %s\n"
"ConsumedTimes: %s\n"
"PublishTime: %s\n"
"Body: %s\n"
"NextConsumeTime: %s\n"
"ReceiptHandle: %s\n"
"Properties: %s\n" %
(msg.message_id, msg.message_body_md5,
msg.message_tag, msg.consumed_times,
msg.publish_time, msg.message_body,
msg.next_consume_time, msg.receipt_handle,
msg.properties))
except MQExceptionBase as e:
if e.type == "MessageNotExist":
print("No new message! RequestId: %s" % e.req_id)
continue
print("Consume Message Fail! Exception:%s\n" % e)
time.sleep(2)
continue
# Acknowledge consumed messages. If the broker does not receive an
# ACK before msg.next_consume_time, it redelivers the message.
# Each delivery assigns a new receipt handle.
try:
receipt_handle_list = [msg.receipt_handle for msg in recv_msgs]
consumer.ack_message(receipt_handle_list)
print("Ack %s Message Succeed.\n\n" % len(receipt_handle_list))
except MQExceptionBase as e:
print("\nAck Message Fail! Exception:%s" % e)
if e.sub_errors:
for sub_error in e.sub_errors:
print("\tErrorHandle:%s,ErrorCode:%s,ErrorMsg:%s" %
(sub_error["ReceiptHandle"],
sub_error["ErrorCode"],
sub_error["ErrorMessage"]))Notas de uso
**Intervalo de
TransCheckImmunityTime**: Os valores válidos variam de 10 a 300 segundos.Tentativas de verificação de status: Após a primeira verificação, o broker consulta o produtor a cada 10 segundos por até 24 horas.
Cliente separado para a thread de verificação: O verificador de status da transação deve usar sua própria instância de
MQClient. Compartilhar um cliente entre a thread do produtor e a thread de verificação causa problemas de concorrência.Timeout de ACK: Se o consumidor não reconhecer uma mensagem antes do prazo limite de
next_consume_time, o broker a reentrega.
Próximos passos
Mensagens transacionais — Entenda o modelo e o ciclo de vida das mensagens transacionais.
Preparar o ambiente — Instale e configure o SDK cliente HTTP para Python.
Criar recursos — Configure instâncias, tópicos e grupos de consumidores.