#RocketMQ 顺序消息实现

// 方式1：按订单ID hash到同一队列（最常用）
Order order = new Order();
order.setOrderId("order12345");

Message msg = new Message("order-topic", "order-tag",
    order.getOrderId().getBytes(), orderJson.getBytes());

// 使用MessageQueueSelector指定队列
producer.send(msg, new MessageQueueSelector() {
    @Override
    public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
        String orderId = (String) arg;
        // 按订单ID hash取模，选择队列
        int index = Math.abs(orderId.hashCode()) % mqs.size();
        return mqs.get(index);
    }
}, order.getOrderId());  // arg参数会传给select方法

// 效果：相同orderId的消息一定路由到同一Queue

// 方式2：使用内置的SelectMessageQueueByHash（推荐）
producer.send(msg, new SelectMessageQueueByHash(), orderId);

// 方式3：自定义路由（更精细的控制）
public class OrderAwareSelector implements MessageQueueSelector {

    @Override
    public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
        String orderId = (String) arg;

        // 大订单路由到高性能队列（队列0）
        // 小订单路由到普通队列（队列1~9）
        if (isLargeOrder(orderId)) {
            return mqs.get(0);
        }
        return mqs.get(Math.abs(orderId.hashCode()) % (mqs.size() - 1) + 1);
    }
}

// 设置顺序消费模式
consumer.registerMessageListener(new MessageListenerOrderly() {

    @Override
    public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs,
                                               ConsumeOrderlyContext context) {
        // ConsumeOrderlyContext 关键参数：
        // context.setAutoCommit(true);  // 自动提交消费进度
        // context.setSuspendCurrentQueueTimeMillis(1000);  // 消费失败后暂停时间

        for (MessageExt msg : msgs) {
            try {
                processOrderMessage(msg);  // 处理消息
            } catch (Exception e) {
                // 顺序消费的异常处理：
                // 返回SUSPEND_CURRENT_QUEUE_A_MOMENT，暂停当前队列消费
                // 等一会儿再重试，保证顺序不乱
                return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
            }
        }
        return ConsumeOrderlyStatus.SUCCESS;
    }
});

// 方案1：批量+异步处理
consumer.registerMessageListener(new MessageListenerOrderly() {
    @Override
    public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs,
                                               ConsumeOrderlyContext context) {
        // msgs是批量拉取的消息，按Queue分组的

        // 将消息按业务ID分组
        Map<String, List<MessageExt>> groupByOrder =
            msgs.stream().collect(Collectors.groupingBy(msg -> msg.getKeys()));

        // 同一订单的消息按seq排序后顺序处理
        for (Map.Entry<String, List<MessageExt>> entry : groupByOrder.entrySet()) {
            List<MessageExt> orderMsgs = entry.getValue();
            orderMsgs.sort(Comparator.comparingInt(msg -> msg.getSeqNum()));

            for (MessageExt msg : orderMsgs) {
                processOrderMessage(msg);
            }
        }
        return ConsumeOrderlyStatus.SUCCESS;
    }
});

// 方案2：重量级处理异步化，重量级结果同步写
// 轻量级校验在顺序消费线程中完成
// 重量级计算（外部API调用）放到异步线程池
// 异步线程池处理完成后，更新本地缓存/Db

// 方案1：关闭自动提交，手动控制提交时机
consumer.registerMessageListener(new MessageListenerOrderly() {
    @Override
    public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs,
                                               ConsumeOrderlyContext context) {
        context.setAutoCommit(false);  // 关闭自动提交

        for (MessageExt msg : msgs) {
            try {
                boolean success = processOrderMessage(msg);
                if (!success) {
                    return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
                }
            } catch (Exception e) {
                return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
            }
        }

        // 全部处理成功才提交
        context.commit();  // 手动提交消费进度
        return ConsumeOrderlyStatus.SUCCESS;
    }
});

// 方案2：业务层幂等
// 即使消息被重复消费，也不会有副作用