1
// 保证 Key 相同的消息进入同一分区
2
ProducerRecord<String, String> record = new ProducerRecord<>(
3
    "orders",
4
    orderId,    // Key：相同 orderId 路由到同一分区
5
    orderJson   // Value
6
);
7
producer.send(record);
8

9
// Kafka 默认分区器：
10
// 1. Key != null → hash(Key) % numPartitions
11
// 2. Key == null → 轮询 / 粘性分区

1
// 保证 FIFO 有序的完整配置
2
Properties props = new Properties();
3
props.put("bootstrap.servers", "localhost:9092");
4
// 幂等生产者：防止重试导致消息重复或乱序
5
props.put("enable.idempotence", "true");
6
props.put("acks", "all");
7
// max.in.flight.requests.per.connection ≤ 5 时幂等仍保证有序
8
props.put("max.in.flight.requests.per.connection", "5");
9
// 消费者：单分区单消费者保证 FIFO
10
props.put("group.id", "order-processor");
11
// 禁止自动提交，处理完再提交，防止消息丢失后跳过
12
props.put("enable.auto.commit", "false");

1
// 幂等生产者：防止重试导致乱序
2
props.put("enable.idempotence", "true");
3
props.put("acks", "all");
4
props.put("max.in.flight.requests.per.connection", "5");
5
// 注意：启用幂等后，max.in.flight.requests.per.connection <= 5 仍可保证有序
6

7
// 不启用幂等时，必须设置 max.in.flight.requests.per.connection=1
8
// 否则重试可能导致消息乱序

1
// Producer：按 Key 选择队列
2
SendResult result = producer.send(msg, new MessageQueueSelector() {
3
    @Override
4
    public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
5
        Long orderId = (Long) arg;
6
        int index = (int) (orderId % mqs.size());
7
        return mqs.get(index);
8
    }
9
}, orderId);
10

11
// Consumer：顺序消费
12
consumer.registerMessageListener(new MessageListenerOrderly() {
13
    @Override
14
    public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
15
        for (MessageExt msg : msgs) {
16
            processOrder(msg);
17
        }
18
        return ConsumeOrderlyStatus.SUCCESS;
19
    }
20
});

1
// 单队列：消息按入队顺序消费
2
channel.basicConsume("order-queue", false, consumer);
3

4
// 问题：多个消费者消费同一队列时，消息可能乱序
5
// Consumer A 处理 msg1（慢）
6
// Consumer B 处理 msg2（快）→ msg2 先完成

1
// 策略：一致性哈希 Exchange
2
// 安装插件：rabbitmq-plugins enable rabbitmq_consistent_hash_exchange
3

4
channel.exchangeDeclare("order-hash", "x-consistent-hash");
5
channel.queueDeclare("order-queue-0", true, false, false, null);
6
channel.queueDeclare("order-queue-1", true, false, false, null);
7

8
// 绑定：权重为 1
9
channel.queueBind("order-queue-0", "order-hash", "1");
10
channel.queueBind("order-queue-1", "order-hash", "1");
11

12
// 发送：按 orderId 路由
13
channel.basicPublish("order-hash", orderId, null, data.getBytes());

1
// 消费者端：按时间戳排序
2
public class OrderedProcessor {
3
    private final TreeMap<Long, Message> buffer = new TreeMap<>();
4
    private long lastProcessedTimestamp = 0;
5
    private static final long MAX_OUT_OF_ORDER_MS = 5000; // 5 秒窗口
6

7
    public void process(Message msg) {
8
        long timestamp = msg.getTimestamp();
9
        buffer.put(timestamp, msg);
10

11
        // 处理窗口内的消息
12
        while (!buffer.isEmpty()) {
13
            long oldest = buffer.firstKey();
14
            if (lastProcessedTimestamp == 0 || oldest <= lastProcessedTimestamp + MAX_OUT_OF_ORDER_MS) {
15
                Message toProcess = buffer.remove(oldest);
16
                handleMessage(toProcess);
17
                lastProcessedTimestamp = oldest;
18
            } else {
19
                break; // 等待更多消息
20
            }
21
        }
22
    }
23
}

1
// Producer：为消息分配单调递增的序列号
2
public class SequencedProducer {
3
    private final AtomicLong sequence = new AtomicLong(0);
4

5
    public void send(String topic, String key, byte[] value) {
6
        long seq = sequence.incrementAndGet();
7
        // 将序列号放入消息头
8
        RecordHeaders headers = new RecordHeaders();
9
        headers.add("sequence", ByteBuffer.allocate(8).putLong(seq).array());
10
        ProducerRecord<String, byte[]> record = new ProducerRecord<>(
11
            topic, null, key, value, headers);
12
        producer.send(record);
13
    }
14
}
15

16
// Consumer：按序列号排序
17
public class SequencedConsumer {
18
    private final TreeMap<Long, ConsumerRecord<String, byte[]>> buffer = new TreeMap<>();
19
    private long expectedSequence = 1;
20

21
    public void process(ConsumerRecord<String, byte[]> record) {
22
        long seq = ByteBuffer.wrap(record.headers().lastHeader("sequence").value()).getLong();
23
        buffer.put(seq, record);
24

25
        // 按序处理
26
        while (!buffer.isEmpty() && buffer.firstKey() == expectedSequence) {
27
            ConsumerRecord<String, byte[]> toProcess = buffer.remove(buffer.firstKey());
28
            handleMessage(toProcess);
29
            expectedSequence++;
30
        }
31

32
        // 检测缺失
33
        if (!buffer.isEmpty() && buffer.firstKey() > expectedSequence + 1) {
34
            // 有消息缺失，等待或请求重发
35
            log.warn("Gap detected: expected {}, got {}", expectedSequence, buffer.firstKey());
36
        }
37
    }
38
}

1
// 订单状态机：只允许合法的状态转换
2
public class OrderStateMachine {
3
    private static final Map<OrderStatus, Set<OrderStatus>> TRANSITIONS = Map.of(
4
        OrderStatus.CREATED,   Set.of(OrderStatus.PAID, OrderStatus.CANCELLED),
5
        OrderStatus.PAID,      Set.of(OrderStatus.SHIPPED, OrderStatus.REFUNDING),
6
        OrderStatus.SHIPPED,   Set.of(OrderStatus.DELIVERED),
7
        OrderStatus.DELIVERED, Set.of(OrderStatus.COMPLETED, OrderStatus.REFUNDING),
8
        OrderStatus.REFUNDING, Set.of(OrderStatus.REFUNDED),
9
        OrderStatus.CANCELLED, Set.of(),
10
        OrderStatus.COMPLETED, Set.of(),
11
        OrderStatus.REFUNDED,  Set.of()
12
    );
13

14
    public boolean canTransit(OrderStatus from, OrderStatus to) {
15
        return TRANSITIONS.getOrDefault(from, Set.of()).contains(to);
16
    }
17

18
    public void handleEvent(OrderEvent event) {
19
        Order order = orderRepository.findById(event.getOrderId());
20
        if (!canTransit(order.getStatus(), event.getNewStatus())) {
21
            // 乱序事件：丢弃或延迟处理
22
            log.warn("Invalid transition: {} → {}, orderId={}",
23
                order.getStatus(), event.getNewStatus(), event.getOrderId());
24
            return;
25
        }
26
        order.setStatus(event.getNewStatus());
27
        orderRepository.save(order);
28
    }
29
}

1
// 向量时钟实现因果追踪
2
public class VectorClock {
3
    private final Map<String, Long> clock = new ConcurrentHashMap<>();
4

5
    public void increment(String processId) {
6
        clock.merge(processId, 1L, Long::sum);
7
    }
8

9
    public void merge(VectorClock other) {
10
        other.clock.forEach((processId, timestamp) ->
11
            clock.merge(processId, timestamp, Long::max));
12
    }
13

14
    // 判断因果关系
15
    public Causality compare(VectorClock other) {
16
        boolean thisLessOrEqual = true;
17
        boolean otherLessOrEqual = true;
18

19
        Set<String> allKeys = new HashSet<>(clock.keySet());
20
        allKeys.addAll(other.clock.keySet());
21

22
        for (String key : allKeys) {
23
            long thisVal = clock.getOrDefault(key, 0L);
24
            long otherVal = other.clock.getOrDefault(key, 0L);
25
            if (thisVal > otherVal) otherLessOrEqual = false;
26
            if (thisVal < otherVal) thisLessOrEqual = false;
27
        }
28

29
        if (thisLessOrEqual && !otherLessOrEqual) return Causality.BEFORE;
30
        if (otherLessOrEqual && !thisLessOrEqual) return Causality.AFTER;
31
        if (thisLessOrEqual && otherLessOrEqual) return Causality.EQUAL;
32
        return Causality.CONCURRENT; // 并发，无因果关系
33
    }
34
}