How to Build an Atomic‑Level RocketMQ Transactional Message Wrapper for Production

Why Transactional Messages Matter

In distributed systems the difficulty lies in guaranteeing that a local database transaction and the corresponding message delivery stay consistent. Typical failure scenarios include:

Local transaction commits but the message is not sent.

Message is sent but the local business logic fails.

Producer restarts, network jitter, or broker check‑back leaves the transaction status ambiguous.

High concurrency causing performance bottlenecks while trying to maintain consistency.

Most tutorials only cover the two‑phase commit API and ignore production concerns such as reliable state persistence, check‑back design, consumer idempotency, throttling, sharding, monitoring, reconciliation, and compensation.

Problem Scope of Distributed Transactions

In a monolith a single DB transaction can cover the whole business chain (order insert → inventory deduction → coupon update → commit). In a microservice architecture each service owns its own database, so a single local transaction cannot guarantee end‑to‑end consistency.

Typical inconsistency example: an order record is persisted but the inventory notification message fails, leaving a dirty state. The reverse (message sent, local transaction rolled back) also creates a non‑existent order visible to downstream services.

RocketMQ transactional messages aim to guarantee atomic coordination between "local transaction commit" and "message eventual delivery". The guarantee is limited to the producer side; consumer side consistency must be achieved via idempotent processing.

Core Principles of RocketMQ Transactional Messages

Two‑Phase Commit Flow (not traditional 2PC)

Send a half (pre‑store) message.

Broker stores the half message but hides it from consumers.

Producer executes the local transaction.

If the local transaction succeeds, the broker commits the message; otherwise it rolls back.

If the broker does not receive a clear result for a long time, it initiates a transaction check‑back.

The broker never locks multiple resources globally; it only links message storage, local transaction execution, and transaction result confirmation into a recoverable flow.

State Flow

HalfMessageStored – half message persisted.

LocalTxRunning – local transaction in progress.

CommitReady – local transaction succeeded, ready to commit.

RollbackReady – local transaction failed, ready to roll back.

VisibleForConsumer – message becomes visible to consumers.

Discarded – message is discarded.

Checkback – broker asks producer for the final status.

UNKNOWN – broker cannot determine status, waits for next check‑back.

Why Check‑Back Is Essential

Without check‑back the transaction could stay in an indefinite unknown state when:

Local transaction succeeds but the application crashes before reporting.

Network interruption occurs during the COMMIT command.

Broker stores the half message and the producer never returns a final result.

Therefore the producer must be able to reconstruct the transaction status from persistent storage (database), not from in‑memory caches.

When to Use Transactional Messages

Order creation → notify inventory, marketing, points, fulfillment.

Payment success → notify accounting, membership, invoicing.

Account changes → publish fund‑flow events.

Approval passes → trigger asynchronous resource provisioning.

High‑throughput pipelines that can tolerate short‑term eventual consistency.

When Not to Use

Strong‑consistency financial core accounting that requires simultaneous commit on both sides.

Scenarios demanding that all downstream actions succeed before the API returns.

Business that cannot accept asynchronous compensation.

Systems unable to implement idempotent consumption or reliable check‑back.

Comparison with Other Distributed Transaction Solutions

2PC/XA : Strong consistency, low performance, high implementation complexity, long lock time.

TCC : Business‑level strong consistency, medium performance, very high coding complexity (Try/Confirm/Cancel).

Saga : Eventual consistency, high performance, high implementation complexity (compensation design).

Outbox : Eventual consistency, high performance, medium complexity, requires an extra delivery component.

RocketMQ Transactional Message : Eventual consistency, high performance, medium complexity, requires well‑designed check‑back and idempotency.

Conclusion: for high‑concurrency reliable event publishing, RocketMQ transactional messages are usually lighter than TCC and more direct than a pure Outbox.

Production‑Grade Design Principles

Persist check‑back criteria. Redis may be used as a cache but the authoritative source must be the database (business table, transaction_message table, or both).

Define business unique keys up front and enforce idempotency. Three levels of idempotency are required: request‑level, message‑send level, and consumer‑processing level.

Consumers must naturally support duplicate deliveries. RocketMQ guarantees at‑least‑once delivery.

Transaction listener must only perform local DB work. Remote RPCs belong to the consumer side.

Treat exceptions as recoverable flows. Provide compensation, retry, dead‑letter, reconciliation, and alerting mechanisms.

Recommended Architecture: Business Table + Transaction Message Table + MQ Transaction Message

The transaction_message table records:

Business key

Topic, tag, message key

Transaction ID

Payload summary (JSON)

Status (PREPARED, COMMIT, ROLLBACK, UNKNOWN)

Check status (INIT, CHECKING, RESOLVED)

Retry count and last error

It serves two purposes: a reliable source for broker check‑back and an operational handle for reconciliation and compensation.

SQL Definitions

CREATE TABLE `t_order` (
  `id` BIGINT NOT NULL AUTO_INCREMENT,
  `order_no` VARCHAR(64) NOT NULL,
  `user_id` BIGINT NOT NULL,
  `product_id` BIGINT NOT NULL,
  `amount` DECIMAL(18,2) NOT NULL,
  `status` VARCHAR(32) NOT NULL,
  `request_id` VARCHAR(64) NOT NULL,
  `created_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
  `updated_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  PRIMARY KEY (`id`),
  UNIQUE KEY `uk_order_no` (`order_no`),
  UNIQUE KEY `uk_request_id` (`request_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

CREATE TABLE `t_transaction_message` (
  `id` BIGINT NOT NULL AUTO_INCREMENT,
  `transaction_id` VARCHAR(64) NOT NULL,
  `business_key` VARCHAR(64) NOT NULL,
  `topic` VARCHAR(128) NOT NULL,
  `tag` VARCHAR(64) DEFAULT NULL,
  `message_key` VARCHAR(128) NOT NULL,
  `payload` JSON NOT NULL,
  `status` VARCHAR(32) NOT NULL,
  `check_status` VARCHAR(32) NOT NULL,
  `retry_count` INT NOT NULL DEFAULT 0,
  `last_error` VARCHAR(512) DEFAULT NULL,
  `created_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
  `updated_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  PRIMARY KEY (`id`),
  UNIQUE KEY `uk_transaction_id` (`transaction_id`),
  UNIQUE KEY `uk_message_key` (`message_key`),
  KEY `idx_business_key` (`business_key`),
  KEY `idx_status_created_at` (`status`, `created_at`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

Status Machine

status = PREPARED

– local transaction preparing. status = COMMIT – local transaction succeeded, MQ should commit. status = ROLLBACK – local transaction failed, MQ should roll back. status = UNKNOWN – broker cannot determine, waits for next check‑back. check_status = INIT – check‑back not started. check_status = CHECKING – broker is checking. check_status = RESOLVED – check‑back result determined.

End‑to‑End Flow for Order Creation (High Concurrency)

User submits an order request with a unique requestId.

Application layer performs idempotent request check.

Producer sends a half message to RocketMQ.

Transaction listener executes the local transaction.

Within a single DB transaction:

Insert order record.

Insert transaction_message record.

Update business status.

If the local transaction succeeds, the listener returns COMMIT; otherwise ROLLBACK.

Broker makes the message visible to consumers.

Consumers process the message idempotently (using messageKey).

Downstream services (inventory, points, notification) each complete their own local transactions.

Monitoring collects success rate, check‑back rate, duplicate‑consume rate, etc.

High‑Concurrency Design Highlights

Request Idempotency: Use requestId, orderNo, and clientToken to filter duplicate submissions early.

Message Key Sharding: Use the business primary key as messageKey (e.g., ORDER_CREATED:202604120001) to aid log retrieval, deduplication, and troubleshooting.

Hotspot Isolation: During flash sales, apply gateway rate limiting, bucketed routing in the order service, per‑product lock shards in inventory, and horizontally scale consumer groups.

Ordering vs Parallelism: Keep ordering local to the same order or account while allowing different orders to be processed in parallel.

Production‑Grade Code Packaging (Java / Spring Boot)

Maven Dependencies

<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
    <dependency>
        <groupId>org.apache.rocketmq</groupId>
        <artifactId>rocketmq-spring-boot-starter</artifactId>
        <version>2.3.2</version>
    </dependency>
    <dependency>
        <groupId>com.baomidou</groupId>
        <artifactId>mybatis-plus-boot-starter</artifactId>
        <version>3.5.5</version>
    </dependency>
    <dependency>
        <groupId>com.mysql</groupId>
        <artifactId>mysql-connector-j</artifactId>
        <scope>runtime</scope>
    </dependency>
    <dependency>
        <groupId>com.fasterxml.jackson.core</groupId>
        <artifactId>jackson-databind</artifactId>
    </dependency>
</dependencies>

Configuration Sample (application.yml style)

spring:
  application:
    name: order-service
  datasource:
    url: jdbc:mysql://127.0.0.1:3306/order_db?useUnicode=true&characterEncoding=utf8&serverTimezone=Asia/Shanghai
    username: root
    password: 123456
  redis:
    host: 127.0.0.1
    port: 6379
rocketmq:
  name-server: 127.0.0.1:9876
  producer:
    group: order-tx-producer-group
    send-message-timeout: 3000
    retry-times-when-send-failed: 2
    retry-times-when-send-async-failed: 2
order:
  topic: order-event-topic
  tag: order-created

Domain Objects

public record CreateOrderCommand(
    String requestId,
    String orderNo,
    Long userId,
    Long productId,
    Integer quantity,
    BigDecimal amount) {}

public enum TransactionMessageStatus { PREPARED, COMMIT, ROLLBACK, UNKNOWN }

public enum OrderStatus { INIT, CREATED, CANCELLED }

public class TransactionMessageEntity {
    private Long id;
    private String transactionId;
    private String businessKey;
    private String topic;
    private String tag;
    private String messageKey;
    private String payload;
    private String status;
    private String checkStatus;
    private Integer retryCount;
    private String lastError;
    private LocalDateTime createdAt;
    private LocalDateTime updatedAt;
}

Order Application Service (sends half message)

@Service
@RequiredArgsConstructor
public class OrderApplicationService {
    private final RocketMQTemplate rocketMQTemplate;
    private final ObjectMapper objectMapper;
    private final OrderProperties orderProperties;

    public SendResult createOrder(CreateOrderCommand command) {
        String transactionId = UUID.randomUUID().toString();
        String messageKey = "ORDER_CREATED:" + command.orderNo();
        OrderCreatedEvent event = new OrderCreatedEvent(
                transactionId,
                messageKey,
                command.requestId(),
                command.orderNo(),
                command.userId(),
                command.productId(),
                command.quantity(),
                command.amount());
        Message<String> message = MessageBuilder.withPayload(writeValue(event))
                .setHeader(RocketMQHeaders.KEYS, messageKey)
                .setHeader(RocketMQHeaders.TRANSACTION_ID, transactionId)
                .setHeader(RocketMQHeaders.TAGS, orderProperties.getTag())
                .build();
        TransactionSendResult result = rocketMQTemplate.sendMessageInTransaction(
                orderProperties.getTopic(), message, command);
        if (result == null || result.getSendStatus() != SendStatus.SEND_OK) {
            throw new IllegalStateException("failed to send transaction message");
        }
        return result;
    }

    private String writeValue(Object value) {
        try {
            return objectMapper.writeValueAsString(value);
        } catch (JsonProcessingException e) {
            throw new IllegalArgumentException("serialize event failed", e);
        }
    }
}

Local Transaction Service (writes order and transaction_message atomically)

@Service
@RequiredArgsConstructor
public class OrderLocalTransactionService {
    private final OrderMapper orderMapper;
    private final TransactionMessageMapper transactionMessageMapper;
    private final ObjectMapper objectMapper;

    @Transactional(rollbackFor = Exception.class)
    public void createOrderAndRecordMessage(String transactionId, CreateOrderCommand command) {
        if (orderMapper.selectByRequestId(command.requestId()) != null) {
            return; // idempotent
        }
        OrderEntity order = new OrderEntity();
        order.setOrderNo(command.orderNo());
        order.setRequestId(command.requestId());
        order.setUserId(command.userId());
        order.setProductId(command.productId());
        order.setAmount(command.amount());
        order.setStatus(OrderStatus.CREATED.name());
        orderMapper.insert(order);

        TransactionMessageEntity txMessage = new TransactionMessageEntity();
        txMessage.setTransactionId(transactionId);
        txMessage.setBusinessKey(command.orderNo());
        txMessage.setTopic("order-event-topic");
        txMessage.setTag("order-created");
        txMessage.setMessageKey("ORDER_CREATED:" + command.orderNo());
        txMessage.setPayload(toJson(command));
        txMessage.setStatus(TransactionMessageStatus.COMMIT.name());
        txMessage.setCheckStatus("INIT");
        txMessage.setRetryCount(0);
        transactionMessageMapper.insert(txMessage);
    }

    @Transactional(rollbackFor = Exception.class)
    public void markRollback(String transactionId, CreateOrderCommand command, Exception ex) {
        TransactionMessageEntity txMessage = transactionMessageMapper.selectByTransactionId(transactionId);
        if (txMessage == null) {
            txMessage = new TransactionMessageEntity();
            txMessage.setTransactionId(transactionId);
            txMessage.setBusinessKey(command.orderNo());
            txMessage.setTopic("order-event-topic");
            txMessage.setTag("order-created");
            txMessage.setMessageKey("ORDER_CREATED:" + command.orderNo());
            txMessage.setPayload(toJson(command));
            txMessage.setCheckStatus("RESOLVED");
            txMessage.setRetryCount(0);
            txMessage.setLastError(ex.getMessage());
            txMessage.setStatus(TransactionMessageStatus.ROLLBACK.name());
            transactionMessageMapper.insert(txMessage);
            return;
        }
        txMessage.setStatus(TransactionMessageStatus.ROLLBACK.name());
        txMessage.setCheckStatus("RESOLVED");
        txMessage.setLastError(ex.getMessage());
        transactionMessageMapper.updateById(txMessage);
    }

    public TransactionMessageStatus queryTransactionStatus(String transactionId) {
        TransactionMessageEntity txMessage = transactionMessageMapper.selectByTransactionId(transactionId);
        if (txMessage == null) {
            return TransactionMessageStatus.UNKNOWN;
        }
        return TransactionMessageStatus.valueOf(txMessage.getStatus());
    }

    private String toJson(Object value) {
        try {
            return objectMapper.writeValueAsString(value);
        } catch (JsonProcessingException e) {
            throw new IllegalArgumentException("serialize payload failed", e);
        }
    }
}

Transaction Listener (only DB work, provides check‑back)

@Slf4j
@Component
@RequiredArgsConstructor
@RocketMQTransactionListener(txProducerGroup = "order-tx-producer-group")
public class OrderTransactionListener implements RocketMQLocalTransactionListener {
    private final OrderLocalTransactionService localTransactionService;

    @Override
    public RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        CreateOrderCommand command = (CreateOrderCommand) arg;
        String transactionId = (String) msg.getHeaders().get(RocketMQHeaders.TRANSACTION_ID);
        try {
            localTransactionService.createOrderAndRecordMessage(transactionId, command);
            return RocketMQLocalTransactionState.COMMIT;
        } catch (DuplicateKeyException e) {
            log.warn("duplicate request, transactionId={}", transactionId, e);
            return RocketMQLocalTransactionState.COMMIT;
        } catch (Exception ex) {
            log.error("local transaction failed, transactionId={}", transactionId, ex);
            try {
                localTransactionService.markRollback(transactionId, command, ex);
            } catch (Exception recordEx) {
                log.error("record rollback state failed, transactionId={}", transactionId, recordEx);
            }
            return RocketMQLocalTransactionState.ROLLBACK;
        }
    }

    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(MessageExt msg) {
        String transactionId = msg.getUserProperty(RocketMQHeaders.TRANSACTION_ID);
        TransactionMessageStatus status = localTransactionService.queryTransactionStatus(transactionId);
        return switch (status) {
            case COMMIT -> RocketMQLocalTransactionState.COMMIT;
            case ROLLBACK -> RocketMQLocalTransactionState.ROLLBACK;
            default -> RocketMQLocalTransactionState.UNKNOWN;
        };
    }
}

Consumer Idempotent Design (Inventory Service Example)

CREATE TABLE `t_consume_dedup` (
  `id` BIGINT NOT NULL AUTO_INCREMENT,
  `consumer_group` VARCHAR(128) NOT NULL,
  `message_key` VARCHAR(128) NOT NULL,
  `biz_key` VARCHAR(64) NOT NULL,
  `status` VARCHAR(32) NOT NULL,
  `created_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
  `updated_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  PRIMARY KEY (`id`),
  UNIQUE KEY `uk_consumer_message` (`consumer_group`, `message_key`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

@Slf4j
@Component
@RequiredArgsConstructor
@RocketMQMessageListener(
        topic = "order-event-topic",
        selectorExpression = "order-created",
        consumerGroup = "inventory-consumer-group",
        consumeMode = ConsumeMode.CONCURRENTLY,
        maxReconsumeTimes = 16)
public class InventoryOrderCreatedConsumer implements RocketMQListener<String> {
    private final ObjectMapper objectMapper;
    private final ConsumeDedupService consumeDedupService;
    private final InventoryService inventoryService;

    @Override
    public void onMessage(String message) {
        OrderCreatedEvent event = readValue(message);
        String consumerGroup = "inventory-consumer-group";
        boolean firstConsume = consumeDedupService.tryMarkProcessing(
                consumerGroup, event.messageKey(), event.orderNo());
        if (!firstConsume) {
            log.info("duplicate message ignored, messageKey={}", event.messageKey());
            return;
        }
        try {
            inventoryService.reserve(event.productId(), event.quantity(), event.orderNo());
            consumeDedupService.markSuccess(consumerGroup, event.messageKey());
        } catch (Exception ex) {
            consumeDedupService.markFail(consumerGroup, event.messageKey(), ex.getMessage());
            throw ex;
        }
    }

    private OrderCreatedEvent readValue(String value) {
        try {
            return objectMapper.readValue(value, OrderCreatedEvent.class);
        } catch (JsonProcessingException e) {
            throw new IllegalArgumentException("parse event failed", e);
        }
    }
}

Inventory Concurrency (Atomic Update)

@Transactional(rollbackFor = Exception.class)
public void reserve(Long productId, Integer quantity, String orderNo) {
    int updatedRows = inventoryMapper.reserve(productId, quantity);
    if (updatedRows == 0) {
        throw new BizException("stock not enough");
    }
    inventoryReservationMapper.insert(new InventoryReservation(orderNo, productId, quantity, "RESERVED"));
}

UPDATE t_inventory
SET available_stock = available_stock - #{quantity},
    reserved_stock = reserved_stock + #{quantity}
WHERE product_id = #{productId}
  AND available_stock >= #{quantity};

Check‑Back Decision Logic

Broker should query in the following order:

Transaction_message table.

If inconclusive, query the business main table.

Combine exception logs if needed.

Return UNKNOWN only when truly uncertain.

Return COMMIT when:

transaction_message.status = COMMIT.

Order exists with status CREATED.

Duplicate request already succeeded.

Return ROLLBACK when:

Local transaction clearly failed.

transaction_message.status = ROLLBACK.

Business record missing and failure logged.

Return UNKNOWN only for half‑finished states, temporary DB invisibility, or storage outage.

Preventing Check‑Back Storms

Keep transaction listener short; avoid remote RPCs.

Index transaction_message table properly.

Scale producer instances horizontally.

Tune RocketMQ check‑back intervals.

Split large transactions into smaller local units.

High‑Concurrency Engineering Upgrades

Entry‑Layer Rate Limiting: per‑user, per‑product, burst limits via gateway.

Thread‑Pool Isolation: separate pools for web requests, message sending, consumer business logic, and compensation tasks.

Database Optimizations: sharding order tables, dedicated transaction_message table, covering indexes, short transactions, periodic archiving.

Message Slimming: keep only essential fields in payload; large fields fetched from object storage; include explicit version numbers.

Consumer Horizontal Scaling: ensure deduplication table supports unique constraints; keep consumer logic short; offload slow work asynchronously; handle dead‑letter messages.

Hotspot Partitioning: for flash‑sale products, partition by productId, use separate topics/tags, allocate dedicated consumer groups.

Monitoring & Observability

Key Metrics

Producer side: transaction send success rate, half‑message backlog, local transaction failure rate, check‑back count & latency, UNKNOWN proportion.

Consumer side: consume TPS, failure rate, duplicate consume rate, retry distribution, dead‑letter count, business processing latency percentiles.

Database side: transaction_message growth speed, status distribution, reconciliation scan latency, slow‑SQL count.

Trace Propagation Fields

traceId

transactionId

messageKey

businessKey

requestId

Alert Rules (examples)

Check‑back rate spikes within 5 minutes.

Dead‑letter queue size exceeds threshold.

UNKNOWN status count continuously grows.

Consumer group failure rate sudden increase.

Repeated consumption of the same business key.

Deployment & Operations Recommendations

RocketMQ cluster with multiple NameServers and broker HA (master‑slave or DLedger), SSD storage, and separated write‑flush resources.

Deploy producer instances on multiple nodes to avoid a single‑point check‑back.

Horizontally scale consumers according to consumption capacity.

Compensation jobs should use distributed locks to prevent duplication.

In Kubernetes, run brokers as StatefulSets with dedicated PVCs, set resource requests/limits, and isolate broker I/O from business pods.

Transaction_message table retention: hot data ≤ 7 days online, archive 7‑30 days, purge > 30 days according to audit requirements.

Final Takeaways

RocketMQ transactional messages provide a practical balance: they do not aim for expensive global strong consistency, yet they guarantee producer‑side atomicity and consumer‑side eventual consistency through half‑messages, check‑backs, and idempotent processing. A production‑ready solution must include persistent transaction state, robust check‑back logic, comprehensive idempotency, monitoring, reconciliation, and scaling strategies—not just a demo that sends a message and prints a log.