Implementing Transactional Messages with RocketMQ in Microservices

RocketMQ provides a transaction message type that ensures eventual consistency between message production and local transactions in distributed environments.

The basic concept of a transaction message involves a producer and a consumer, where the producer sends a half‑message that is stored on the broker until a second‑phase commit or rollback is confirmed.

Half messages are marked as “temporarily undeliverable”. If the broker detects a half‑message lingering, it initiates a back‑check to the producer to determine the final transaction state.

The overall architecture is illustrated in Figure 1, and the execution flow (seven steps) is shown in Figure 2, involving Service A (producer) and Service B (consumer).

Implementation steps include creating a transaction record table, implementing the TransactionListener interface, and using RocketMQTemplate.sendMessageInTransaction to send messages.

CREATE TABLE `tx_record` (
    `tx_no` varchar(64) NOT NULL COMMENT '事务Id',
    `create_time` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
    PRIMARY KEY (`tx_no`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8 COMMENT='事务记录表'

The TransactionListener defines two methods: executeLocalTransaction for executing the local business logic and checkLocalTransaction for transaction status back‑check.

public interface TransactionListener {
    LocalTransactionState executeLocalTransaction(final Message msg, final Object arg);
    LocalTransactionState checkLocalTransaction(final MessageExt msg);
}

Sample Spring service code shows how to send a transactional message and handle idempotency using the transaction record table.

@Service
public class CustomerTicketServiceImpl implements ICustomerTicketService {
    @Autowired TxRecordMapper txRecordMapper;
    @Autowired RocketMQTemplate rocketMQTemplate;

    @Override
    public void generateTicket(AddCustomerTicketReqVO req) {
        // build event, convert to JSON, create Message
        Message
message = MessageBuilder.withPayload(jsonString).build();
        rocketMQTemplate.sendMessageInTransaction("producer_group_ticket","topic_ticket",message,null);
    }

    @Transactional
    public void doGenerateTicket(TicketGeneratedEvent event) {
        if (Objects.nonNull(txRecordMapper.findTxRecordByTxNo(event.getTxNo()))) return;
        // insert ticket and transaction record
    }
}

The listener implementation commits or rolls back based on the success of the local transaction and provides a back‑check method that queries the transaction record table.

@Component
@RocketMQTransactionListener(txProducerGroup = "producer_group_ticket")
public class ProducerListener implements RocketMQLocalTransactionListener {
    @Autowired ICustomerTicketService customerTicketService;
    @Autowired TxRecordMapper txRecordMapper;

    @Override
    @Transactional
    public RocketMQLocalTransactionState executeLocalTransaction(Message message, Object o) {
        // parse event, execute local transaction, return COMMIT or ROLLBACK
    }

    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message message) {
        // query tx_record, return COMMIT or UNKNOWN
    }
}

Consumer side only needs to implement RocketMQListener and process the message, also using the transaction record table for idempotent handling.

@Component
@RocketMQMessageListener(consumerGroup = "consumer_group_ticket",topic = "topic_ticket")
public class Consumer implements RocketMQListener
{
    @Autowired IChatRecordService chatRecordService;

    @Override
    public void onMessage(String message) {
        // parse and handle message
    }
}

Overall, the article demonstrates how to integrate RocketMQ transactional messages into a microservice architecture to achieve reliable event processing and distributed transaction consistency.