RocketMQ进阶03：顺序消息原理 - 局部有序与全局有序的设计权衡

public class OrderedProducer {

    public static void main(String[] args) throws Exception {
        DefaultMQProducer producer = new DefaultMQProducer("ordered_producer_group");
        producer.setNamesrvAddr("localhost:9876");
        producer.start();

        // 模拟订单消息
        List<OrderStep> orderSteps = buildOrders();

        for (OrderStep step : orderSteps) {
            Message msg = new Message(
                "order_topic",
                step.getTags(),
                step.toString().getBytes()
            );

            // 顺序发送：相同orderId的消息发送到同一个Queue
            SendResult sendResult = producer.send(
                msg,
                new MessageQueueSelector() {
                    @Override
                    public MessageQueue select(
                        List<MessageQueue> mqs,
                        Message msg,
                        Object arg
                    ) {
                        Long orderId = (Long) arg;
                        // Hash取模，确保相同orderId路由到同一Queue
                        int index = (int) (orderId % mqs.size());
                        return mqs.get(index);
                    }
                },
                step.getOrderId()  // orderKey
            );

            System.out.printf("发送：%s, Queue=%d%n",
                step, sendResult.getMessageQueue().getQueueId());
        }

        producer.shutdown();
    }

    // 构建订单步骤
    private static List<OrderStep> buildOrders() {
        List<OrderStep> steps = new ArrayList<>();

        // 订单1001
        steps.add(new OrderStep(1001L, "创建"));
        steps.add(new OrderStep(1001L, "支付"));
        steps.add(new OrderStep(1001L, "发货"));

        // 订单1002
        steps.add(new OrderStep(1002L, "创建"));
        steps.add(new OrderStep(1002L, "支付"));

        // 订单1003
        steps.add(new OrderStep(1003L, "创建"));

        return steps;
    }
}

class OrderStep {
    private Long orderId;
    private String status;

    public OrderStep(Long orderId, String status) {
        this.orderId = orderId;
        this.status = status;
    }

    public Long getOrderId() {
        return orderId;
    }

    public String getTags() {
        return status;
    }

    @Override
    public String toString() {
        return "Order=" + orderId + ", Status=" + status;
    }
}

public class OrderedConsumer {

    public static void main(String[] args) throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("ordered_consumer_group");
        consumer.setNamesrvAddr("localhost:9876");
        consumer.subscribe("order_topic", "*");

        // 注册顺序消息监听器
        consumer.registerMessageListener(new MessageListenerOrderly() {

            @Override
            public ConsumeOrderlyStatus consumeMessage(
                List<MessageExt> msgs,
                ConsumeOrderlyContext context
            ) {
                // 单线程顺序消费
                for (MessageExt msg : msgs) {
                    System.out.printf(
                        "消费：Thread=%s, Queue=%d, Msg=%s%n",
                        Thread.currentThread().getName(),
                        msg.getQueueId(),
                        new String(msg.getBody())
                    );
                }

                return ConsumeOrderlyStatus.SUCCESS;
            }
        });

        consumer.start();
        System.out.println("顺序消费启动成功");
    }
}

public class MessageListenerOrderly {

    // 消费时对Queue加锁
    public ConsumeOrderlyStatus consumeMessage(
        List<MessageExt> msgs,
        ConsumeOrderlyContext context
    ) {
        // 1. 获取Queue锁（本地锁 + Broker锁）
        MessageQueue mq = context.getMessageQueue();
        Object lock = this.processQueueTable.get(mq);

        synchronized (lock) {  // Queue级别锁
            // 2. 顺序消费消息
            for (MessageExt msg : msgs) {
                // 处理消息
            }

            // 3. 提交消费进度
            this.updateOffset(mq, msgs);
        }

        return ConsumeOrderlyStatus.SUCCESS;
    }
}

// Consumer启动时向Broker申请Queue锁
public class RebalanceImpl {

    public void lock(MessageQueue mq) {
        // 向Broker发送锁定请求
        Set<MessageQueue> lockedMqs =
            this.mQClientFactory.getMQClientAPIImpl().lockBatchMQ(
                brokerAddr,
                requestBody,
                1000
            );

        // 锁定成功，开始消费
        if (lockedMqs.contains(mq)) {
            processQueue.setLocked(true);
        }
    }
}