158｜消息顺序性：全局顺序、分区顺序、路由键和并发消费取舍

创建订单 → 支付 → 出库 → 发货 → 签收

生产者 → [单一分区/队列] → 单一消费者

// Kafka 中创建全局有序 Topic 的唯一方式
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");

// 使用 AdminClient 创建只有一个分区的 Topic
AdminClient admin = AdminClient.create(props);
NewTopic topic = new NewTopic("order-global", 1, (short) 3);
admin.createTopics(Collections.singleton(topic));

// 生产者：所有消息发送到 Partition 0
ProducerRecord<String, String> record = new ProducerRecord<>("order-global", 0, key, value);

// RocketMQ 全局顺序消息：所有消息发送到同一个 MessageQueue
MessageQueueSelector selector = (mqs, msg, arg) -> {
    // 始终选择第一个队列
    return mqs.get(0);
};

// 生产者发送顺序消息
producer.send(msg, selector, null);

// 消费者：使用 MessageListenerOrderly 保证顺序消费
consumer.registerMessageListener((MessageListenerOrderly) (msgs, context) -> {
    for (MessageExt msg : msgs) {
        // 顺序处理每条消息
        processOrderly(msg);
    }
    return ConsumeOrderlyStatus.SUCCESS;
});

flowchart LR
    subgraph Producers["生产者"]
        P1["订单生产者"]
    end
    
    subgraph Topic["Topic: order-events (3 Partitions)"]
        P0["Partition 0<br/>orderId % 3 == 0"]
        P1pt["Partition 1<br/>orderId % 3 == 1"]
        P2["Partition 2<br/>orderId % 3 == 2"]
    end
    
    subgraph Consumers["消费者组"]
        C0["Consumer 0"]
        C1["Consumer 1"]
        C2["Consumer 2"]
    end
    
    P1 -->|"orderId=100 → P0"| P0
    P1 -->|"orderId=101 → P1"| P1pt
    P1 -->|"orderId=102 → P2"| P2
    
    P0 --> C0
    P1pt --> C1
    P2 --> C2

// Kafka 默认分区器：key.hashCode() % partitionCount
// 相同 key 的消息一定去往同一个分区
public class DefaultPartitioner implements Partitioner {
    @Override

    public int partition(String topic, Object key, byte[ ] keyBytes,


                         Object value, byte[ ] valueBytes, Cluster cluster) {

        List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
        int numPartitions = partitions.size();
        if (keyBytes == null) {
            // 无 key 时轮询（Round-Robin），不保证顺序
            return counter.getAndIncrement() % numPartitions;
        }
        // 有 key 时 hash 取模，相同 key 进同一分区
        return Math.abs(Utils.murmur2(keyBytes)) % numPartitions;
    }
}

// 按订单 ID 选择队列
MessageQueueSelector selector = (mqs, msg, arg) -> {
    Long orderId = (Long) arg;
    int index = (int) (orderId % mqs.size());
    return mqs.get(index);
};

// 发送时传入订单 ID 作为路由键
producer.send(msg, selector, orderId);

// RabbitMQ 使用一致性哈希交换器
Map<String, Object> args = new HashMap<>();
args.put("hash-header", "orderId");
channel.exchangeDeclare("order-exchange", "x-consistent-hash", true, false, args);

// 发送时设置 header
AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()
    .headers(Map.of("orderId", orderId))
    .build();
channel.basicPublish("order-exchange", "100", props, body);

// 错误做法：直接用商品 ID 做路由键
// 爆款商品的所有消息都会进入同一个分区
int partition = Math.abs(skuId.hashCode()) % numPartitions;

// 正确做法：引入随机盐值分散热点
String routingKey = skuId + "_" + (random.nextInt(10)); // 10 路分散

扩容前：orderId=100 → hash(100) % 3 = 1 → Partition 1
扩容后：orderId=100 → hash(100) % 4 = 2 → Partition 2

// 没有指定 key 的消息使用轮询策略
producer.send(new ProducerRecord<>("topic", null, message));
// 消息 1 → Partition 0
// 消息 2 → Partition 1
// 消息 3 → Partition 2
// 完全无序

flowchart TD
    subgraph Concurrent["并发消费（Concurrently）"]
        CQ1["Queue 0"] --> CT1["线程 1"]
        CQ1 --> CT2["线程 2"]
        CQ1 --> CT3["线程 3"]
    end
    
    subgraph Orderly["顺序消费（Orderly）"]
        OQ1["Queue 0"] --> OT1["线程 1（独占）"]
        OQ2["Queue 1"] --> OT2["线程 2（独占）"]
        OQ3["Queue 2"] --> OT3["线程 3（独占）"]
    end

// 方式一：并发消费（MessageListenerConcurrently）
// 一个 Queue 的消息可以被多个线程并行消费
// 不保证顺序，但吞吐量高
consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
    // 可以在这里用线程池并行处理
    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
});

// 方式二：顺序消费（MessageListenerOrderly）
// 一个 Queue 的消息被单个线程串行消费
// 保证顺序，但该 Queue 的吞吐量受限于单线程
consumer.registerMessageListener((MessageListenerOrderly) (msgs, context) -> {
    // 串行处理，处理完一条才能处理下一条
    return ConsumeOrderlyStatus.SUCCESS;
});

// 关键约束：消费者数 ≤ 分区数
// Kafka 消费者组中，多余的消费者会处于空闲状态
// 所以并行度上限 = min(消费者数, 分区数)

// 顺序消费的伪代码
public class OrderlyConsumeService {
    // 每个 Queue 绑定一个线程
    private final Map<MessageQueue, Thread> queueThreadMap = new HashMap<>();
    
    public void consume(MessageQueue mq, List<MessageExt> msgs) {
        Thread worker = queueThreadMap.get(mq);
        synchronized (worker) {
            for (MessageExt msg : msgs) {
                // 处理消息——此时该 Queue 的其他消息必须等待
                handleMessage(msg);
            }
        }
    }
}

// 危险：manual ack + requeue 会破坏顺序
channel.basicConsume("order-queue", false, (consumerTag, delivery) -> {
    try {
        processMessage(delivery);
        channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
    } catch (Exception e) {
        // requeue=true 会把消息放回队列头部
        // 如果后面的消息已经被处理了，就会乱序
        channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, true);
    }
}, consumerTag -> {});

// 消息体设计
public class OrderEvent {
    private String orderId;
    private String eventType;   // CREATED, PAID, SHIPPED, DELIVERED
    private long version;       // 递增版本号
    private long timestamp;     // 事件时间
    
    // 消费端处理
    public void handle(OrderEvent event) {
        Order current = orderRepository.findById(event.getOrderId());
        // 忽略旧版本消息
        if (current.getVersion() >= event.getVersion()) {
            return;
        }
        // 应用状态变更
        current.apply(event);
        orderRepository.save(current);
    }
}

// 关键监控指标
public class OrderingMetrics {
    // 1. 消息延迟（从生产到消费的时间差）
    // 如果某个队列的延迟突然飙升，说明该队列有慢消息
    private long consumerLag;
    
    // 2. 队列消费阻塞次数
    // 顺序消费中，一条消息处理慢会导致后续消息全部阻塞
    private long blockedCount;
    
    // 3. 热点分区检测
    // 某个分区的消息量远超其他分区
    private double partitionLoadStdDev;
    
    // 4. 乱序检出率
    // 业务层检测到的乱序消息比例
    private double outOfOrderRate;
}