RocketMQ 架构

/**
 * RocketMQ 架构
 */
public class RocketMQArchitecture {
    // NameServer
    /*
     * - 轻量级注册中心
     * - 负责 Broker 管理、服务发现
     * - 每个 NameServer 独立
     */
    
    // Broker
    /*
     * - 消息存储和转发
     * - 主从部署
     */
    
    // Producer
    /*
     * - 消息生产者
     */
    
    // Consumer
    /*
     * - 消息消费者
     * - 推/拉模式
     */
}

// 1. 创建事务消息生产者
TransactionMQProducer producer = new TransactionMQProducer("group");
producer.setNamesrvAddr("localhost:9876");

// 2. 设置事务监听器
producer.setTransactionListener(new TransactionListener() {
    @Override
    public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        // 执行本地事务
        try {
            // 更新数据库等操作
            boolean success = doBusiness();
            return success ? LocalTransactionState.COMMIT_MESSAGE : 
                           LocalTransactionState.ROLLBACK_MESSAGE;
        } catch (Exception e) {
            return LocalTransactionState.ROLLBACK_MESSAGE;
        }
    }
    
    @Override
    public LocalTransactionState checkLocalTransaction(MessageExt msg) {
        // 事务回查逻辑
        return checkBusinessStatus() ? LocalTransactionState.COMMIT_MESSAGE : 
                                     LocalTransactionState.ROLLBACK_MESSAGE;
    }
});

// 3. 发送事务消息
Message msg = new Message("Topic", "Tag", "Key", "Body".getBytes());
TransactionSendResult result = producer.sendMessageInTransaction(msg, null);