[root@mq-vm ~]# docker network create rocketmq-net
931485285fd0b7e0fa2ddb588432bc431e859d1ba6f997162f5d0fe30f9e5f78

[root@mq-vm ~]# docker run -d --name rocketmq-nameserver --network rocketmq-net -p 9876:9876 apache/rocketmq:4.9.4 sh mqnamesrv
02f73715456be7005dd3204c076d15e3502289850a28cc1ca172991bcf264154

[root@mq-vm ~]# sudo mkdir -p /opt/rocketmq/broker/conf
[root@mq-vm ~]# cat > /opt/rocketmq/broker/conf/broker.conf <<'EOF'
brokerClusterName=DefaultCluster
brokerName=broker-a
brokerId=0
brokerIP1=192.168.126.129
listenPort=10911
autoCreateTopicEnable=true
namesrvAddr=rocketmq-nameserver:9876
EOF

[root@mq-vm ~]# docker run -d --name rocketmq-broker --network rocketmq-net \
  -p 10911:10911 -p 10909:10909 \
  -v /opt/rocketmq/broker/store:/home/rocketmq/store:Z \
  -v /opt/rocketmq/broker/logs:/home/rocketmq/logs:Z \
  -v /opt/rocketmq/broker/conf/broker.conf:/home/rocketmq/rocketmq-4.9.4/conf/broker.conf:Z \
  apache/rocketmq:4.9.4 sh mqbroker -c /home/rocketmq/rocketmq-4.9.4/conf/broker.conf

[root@mq-vm ~]# docker run -d --name rocketmq-console --network rocketmq-net -p 8080:8080 \
   -e JAVA_OPTS="-Drocketmq.namesrv.addr=rocketmq-nameserver:9876" \
   styletang/rocketmq-console-ng:latest
e387591ad5fcb25e803969f642b0422736a4ab7153731e3b9875e1180ce5cc86

[root@mq-vm ~]# docker logs -f rocketmq-nameserver
OpenJDK 64-Bit Server VM warning: Using the DefNew young collector with the CMS collector is deprecated and will likely be removed in a future release
OpenJDK 64-Bit Server VM warning: UseCMSCompactAtFullCollection is deprecated and will likely be removed in a future release.
The Name Server boot success. serializeType=JSON

# docker-compose版本
version: '3.8'
services:
  namesrv:
    image: apache/rocketmq:4.9.4
    container_name: rocketmq-nameserver
    command: sh mqnamesrv
    networks:
      - rocketmq-net
    ports:
      - "9876:9876"

  broker:
    image: apache/rocketmq:4.9.4
    container_name: rocketmq-broker
    command: sh mqbroker -c /home/rocketmq/rocketmq-4.9.4/conf/broker.conf
    networks:
      - rocketmq-net
    volumes:
      - /opt/rocketmq/broker/store:/home/rocketmq/store:Z
      - /opt/rocketmq/broker/logs:/home/rocketmq/logs:Z
      - /opt/rocketmq/broker/conf/broker.conf:/home/rocketmq/rocketmq-4.9.4/conf/broker.conf:Z
    ports:
      - "10911:10911"
      - "10909:10909"
    depends_on:
      - namesrv

  console:
    image: styletang/rocketmq-console-ng:latest
    container_name: rocketmq-console
    networks:
      - rocketmq-net
    ports:
      - "8080:8080"
    environment:
      - JAVA_OPTS=-Drocketmq.namesrv.addr=rocketmq-nameserver:9876

networks:
  rocketmq-net:
    external: true
    
[root@opengauss-vm ~]# docker-compose up -d    

        <!-- RocketMQ Spring Boot starter (provides RocketMQTemplate, annotations) -->
        <dependency>
            <groupId>org.apache.rocketmq</groupId>
            <artifactId>rocketmq-spring-boot-starter</artifactId>
            <version>${rocketmq.spring.boot.starter.version}</version>
        </dependency>

        <!-- RocketMQ client library (explicit version match to broker) -->
        <dependency>
            <groupId>org.apache.rocketmq</groupId>
            <artifactId>rocketmq-client</artifactId>
            <version>${rocketmq.version}</version>
        </dependency>

        // 默认使用，不写也没关系
        consumer.setMessageModel(MessageModel.CLUSTERING);

        // 设置消费模式为广播模式
        consumer.setMessageModel(MessageModel.BROADCASTING);

    @Test
    public void testCompleteFlow() throws Exception {
        System.out.println("=== RocketMQ 完整流程测试开始 ===");

        // 1. 先启动消费者
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group1");
        consumer.setNamesrvAddr("192.168.126.129:9876");
        consumer.subscribe("BenchmarkTest", "*");
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);

        AtomicInteger consumedCount = new AtomicInteger(0);

        consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
            System.out.println("🎉 消费者收到 " + msgs.size() + " 条消息:");
            for (MessageExt msg : msgs) {
                System.out.println("----------------------------------------");
                System.out.println("消息ID: " + msg.getMsgId());
                System.out.println("主题: " + msg.getTopic());
                System.out.println("标签: " + msg.getTags());
                System.out.println("内容: " + new String(msg.getBody()));
                System.out.println("队列: " + msg.getQueueId());
                System.out.println("----------------------------------------");
                consumedCount.incrementAndGet();
            }
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });

        consumer.start();
        System.out.println("✅ 消费者启动成功");

        // 2. 等待消费者注册
        Thread.sleep(5000);

        // 3. 启动生产者发送消息
        DefaultMQProducer producer = new DefaultMQProducer("test-producer-group");
        producer.setNamesrvAddr("192.168.126.129:9876");
        producer.setVipChannelEnabled(false);
        producer.start();
        System.out.println("✅ 生产者启动成功");

        // 发送多条测试消息
        for (int i = 1; i <= 5; i++) {
            String messageContent = "测试消息 " + i + " - 时间: " + System.currentTimeMillis();
            Message message = new Message("BenchmarkTest", "TEST", messageContent.getBytes());
            SendResult result = producer.send(message);
            System.out.println("📤 发送消息 " + i + " 成功, MsgId: " + result.getMsgId());
            Thread.sleep(1000); // 间隔1秒
        }

        producer.shutdown();
        System.out.println("✅ 生产者已关闭");

        // 4. 等待消息被消费
        System.out.println("⏳ 等待消息消费...");
        Thread.sleep(10000);

        System.out.println("📊 总共消费消息数量: " + consumedCount.get());

        consumer.shutdown();
        System.out.println("✅ 消费者已关闭");
        System.out.println("=== 测试完成 ===");
    }

	// 发送的信息带 tag1 标签
	Message message = new Message("topic8", "tag1", msg.getBytes());
	// 只消费 vip 标签的消息
	consumer.subscribe("topic8", "vip");
	// 只消费 tag1 或 vip 标签的消息
	consumer.subscribe("topic8", "tag1 || vip");

	// 给消息设置属性	
	message.putUserProperty(String name, String value);

	message1.putUserProperty("age", "18");
	message1.putUserProperty("name", "zhangsan");

	message2.putUserProperty("age", "15");
	message2.putUserProperty("name", "lisi");

	// 只接收 message1
	consumer1.subscribe("topic8", MessageSelector.bySql("age > 16"));

	// 只接收 message2
	consumer2.subscribe("topic8", MessageSelector.bySql("age < 16 and name = 'lisi'"));

	enablePropertyFilter=true

spring:
  application:
    name: demo-service

  rocketmq:
    # NameServer 列表（必填），多个用 ; 或 , 分隔
    name-server: 127.0.0.1:9876

    # 如果是云上部署（例如阿里云），可设置 CLOUD；本地集群用 LOCAL
    access-channel: LOCAL

    producer:
      # 必填：生产者 group
      group: demo-producer-group

      # 发送超时（ms），默认几秒，生产环境可根据场景调优
      send-message-timeout: 3000

      # 发送失败时重试次数（同步发送），生产环境可设置 1~3
      retry-times-when-send-failed: 2

      # 最大消息体（字节）
      max-message-size: 4194304

      # 如果开启 ACL（鉴权），在这里填环境变量或占位符
      access-key: ${ROCKETMQ_ACCESS_KEY:}
      secret-key: ${ROCKETMQ_SECRET_KEY:}

      # 是否开启消息链路追踪（需要服务端/插件支持）
      enable-msg-trace: true

    consumer:
      # 必填：消费者 group
      group: demo-consumer-group

      # 消息模式：CLUSTERING（集群消费）或 BROADCASTING（广播）
      message-model: CLUSTERING

      # 消费线程数（并发消费线程）
      consume-thread-nums: 20

      # 达到该重试次数后进入死信（或业务处理方式），视 broker 配置
      max-reconsume-times: 16

      # 批量拉取大小
      pull-batch-size: 32

      # ACL（与 producer 对应）
      access-key: ${ROCKETMQ_ACCESS_KEY:}
      secret-key: ${ROCKETMQ_SECRET_KEY:}

import lombok.Data;
import java.io.Serializable;

@Data
@AllArgsConstructor
public class Order implements Serializable {
    private String orderId;
    private String userId;
    private String productId;
    private int quantity;
    private double price;
}

    @Autowired
    RocketMQTemplate rocketMQTemplate;
	
	public void test() {
        Order order = new Order("orderId111", "userId222", "productId333", 6, 6.6);
        rocketMQTemplate.convertAndSend("topic8", order);
    }

@Service
// 监听topic8主题，消费组group8，过滤tag1和tag2的消息，广播模式
@RocketMQMessageListener(topic = "topic8", consumerGroup = "group8",
                         selectorExpression = "tag1 || tag2",
                         messageModel = MessageModel.BROADCASTING)
public class DemoConsumer implements RocketMQListener<Order> {
    @Override
    public void onMessage(Order order) {
        System.out.println("接收到订单: " + order);
    }
}

// 监听topic8主题，消费组group8，使用SQL92表达式过滤消息
@RocketMQMessageListener(topic = "topic8", consumerGroup = "group8",
        selectorType = SelectorType.SQL92, selectorExpression = "quantity >= 2 AND price <= 100")

    public void test() {
        Order order = new Order("orderId111", "userId222", "productId333", 6, 6.6);

        // 发送同步消息
        SendResult syncSend = rocketMQTemplate.syncSend("topic8", order);
        // 延时等级3的延时消息
        SendResult delaySend = rocketMQTemplate.syncSend("topic8",
                MessageBuilder.withPayload(order).build(), 3000, 3);

        // 发送异步消息
        rocketMQTemplate.asyncSend("topic8", order, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                System.out.println("消息发送成功: " + sendResult);
            }

            @Override
            public void onException(Throwable e) {
                System.out.println("消息发送失败: " + e.getMessage());
            }
        }, 3000);   // 超时时间3秒

        // 发送单向消息
        rocketMQTemplate.sendOneWay("topic8", order);
    }

consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
    // 多线程并发执行，无法保证顺序
    msgs.forEach(msg -> process(msg));
    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
});

// 按订单 ID 哈希选择队列
producer.send(message, new MessageQueueSelector() {
    @Override
    public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
        String orderId = (String) arg; // 订单ID作为参数
        int index = Math.abs(orderId.hashCode()) % mqs.size();
        return mqs.get(index);
    }
}, orderId); // 传入订单ID

consumer.registerMessageListener((MessageListenerOrderly) (msgs, context) -> {
    // RocketMQ 保证同一队列的消息串行处理
    msgs.forEach(msg -> process(msg));
    return ConsumeOrderlyStatus.SUCCESS;
});

import org.apache.rocketmq.client.producer.*;
import org.apache.rocketmq.common.message.Message;

public class TransactionProducer {
    public static void main(String[] args) throws Exception {
        // 1. 创建事务消息生产者
        TransactionMQProducer producer = new TransactionMQProducer("transaction-producer-group");
        producer.setNamesrvAddr("192.168.126.129:9876");

        // 2. 设置事务监听器（处理本地事务 + 事务回查）
        producer. setTransactionListener(new TransactionListener() {
            
            // 执行本地事务
            @Override
            public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
                try {
                    System.out.println("执行本地事务:  " + new String(msg.getBody()));
                    
                    // 模拟本地事务（如数据库操作）
                    boolean success = doLocalTransaction(); 
                    
                    if (success) {
                        return LocalTransactionState.COMMIT_MESSAGE;  // 提交
                    } else {
                        return LocalTransactionState.ROLLBACK_MESSAGE; // 回滚
                    }
                } catch (Exception e) {
                    return LocalTransactionState.ROLLBACK_MESSAGE;
                }
            }

            // 事务回查（Broker 主动调用）
            @Override
            public LocalTransactionState checkLocalTransaction(MessageExt msg) {
                System.out.println("Broker 回查事务状态: " + msg.getTransactionId());
                
                // 查询本地事务执行结果（如查数据库）
                boolean committed = checkTransactionStatus(msg.getTransactionId());
                
                if (committed) {
                    return LocalTransactionState.COMMIT_MESSAGE;
                } else {
                    return LocalTransactionState.ROLLBACK_MESSAGE;
                }
            }
        });

        producer.start();

        // 3. 发送事务消息
        Message message = new Message("TransactionTopic", "TagA", "订单创建成功". getBytes());
        TransactionSendResult result = producer. sendMessageInTransaction(message, null);
        System.out.println("发送结果: " + result. getSendStatus());

        // 不关闭，等待回查
        // producer.shutdown();
    }

    // 模拟本地事务
    private static boolean doLocalTransaction() {
        // 实际业务：操作数据库、调用其他服务等
        return true; 
    }

    // 模拟回查本地事务状态
    private static boolean checkTransactionStatus(String transactionId) {
        // 实际业务：查数据库判断事务是否已提交
        return true;
    }
}

 ┌──────────────────────────────────────────────────────────────┐
 │                         NameServer                           │
 │  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐        │
 │  │ NameServer 1 │  │ NameServer 2 │  │ NameServer 3 │        │
 │  └──────────────┘  └──────────────┘  └──────────────┘        │
 │         ▲                 ▲                 ▲                │
 └─────────┼─────────────────┼─────────────────┼────────────────┘
          │ 心跳注册        │                 │
          │                 │                 │
 ┌─────────┴─────────────────┴─────────────────┴────────────────┐
 │                           Broker                             │
 │  ┌─────────────────────┐      ┌─────────────────────┐        │
 │  │   Broker Master A   │      │   Broker Master B   │        │
 │  │   (可读可写)         │      │   (可读可写)         │        │
 │  └──────────┬──────────┘      └──────────┬──────────┘        │
 │             │ 同步/异步复制                │                   │
 │  ┌──────────┴──────────┐      ┌──────────┴──────────┐        │
 │  │   Broker Slave A    │      │   Broker Slave B    │        │
 │  │   (只读)             │      │   (只读)            │        │
 │  └─────────────────────┘      └─────────────────────┘        │
 └──────────────────────────────────────────────────────────────┘
           ▲                                      ▲
           │                                      │
   ┌───────┴────────┐                    ┌───────┴────────┐
   │    Producer    │                    │    Consumer    │
   │  ┌──────────┐  │                    │  ┌──────────┐  │
   │  │Producer 1│  │                    │  │Consumer 1│  │
   │  ├──────────┤  │                    │  ├──────────┤  │
   │  │Producer 2│  │                    │  │Consumer 2│  │
   │  └──────────┘  │                    │  └──────────┘  │
   └────────────────┘                    └────────────────┘

 ┌─────────────┐
 │   Master    │
 │  (单节点)    │
 └─────────────┘

 ┌──────────┐    ┌──────────┐    ┌──────────┐
 │ Master A │    │ Master B │    │ Master C │
 └──────────┘    └──────────┘    └──────────┘

 ┌──────────┐         ┌──────────┐
 │ Master A │───────▶│ Slave A  │
 └──────────┘  异步   └──────────┘
                     
 ┌──────────┐         ┌──────────┐
 │ Master B │───────▶│ Slave B  │
 └──────────┘  异步   └──────────┘

# broker-a.properties
brokerClusterName=DefaultCluster
brokerName=broker-a
brokerId=0                      # 0 表示 Master
brokerRole=ASYNC_MASTER          # 异步主节点
flushDiskType=ASYNC_FLUSH        # 异步刷盘

# broker-a-slave.properties
brokerClusterName=DefaultCluster
brokerName=broker-a              # 与 Master 同名
brokerId=1                       # 非 0 表示 Slave
brokerRole=SLAVE                 # 从节点

 ┌──────────┐  同步   ┌──────────┐
 │ Master A │───────▶│ Slave A  │
 └──────────┘  等待   └──────────┘
                     
 ┌──────────┐  同步   ┌──────────┐
 │ Master B │───────▶│ Slave B  │
 └──────────┘  等待   └──────────┘

brokerRole=SYNC_MASTER           # 同步主节点
flushDiskType=SYNC_FLUSH         # 同步刷盘（可选，进一步提高可靠性）

CommitLog 文件：
[消息1][消息2][消息3][消息4]...   ← 顺序追加，磁头不回退

# broker.conf
flushDiskType=ASYNC_FLUSH  # 异步刷盘（推荐）
# flushDiskType=SYNC_FLUSH   # 同步刷盘（金融场景）

# broker.conf
deleteWhen=04                # 删除时间（凌晨 4 点）
fileReservedTime=48          # 文件保留时间（小时）