消息队列顺序消息与延迟消息实战

// 按订单ID作为分区键，同一订单的所有消息进入同一分区
ProducerRecord<String, OrderMessage> record = new ProducerRecord<>(
    "order-topic",        // topic
    orderId,               // key，作为分区键
    orderMessage           // value
);

producer.send(record, (metadata, exception) -> {
    if (exception != null) {
        log.error("消息发送失败 orderId={}", orderId, exception);
    } else {
        log.info("消息发送成功 topic={} partition={} offset={}",
            metadata.topic(), metadata.partition(), metadata.offset());
    }
});

Properties props = new Properties();
props.put("bootstrap.servers", "kafka1:9092,kafka2:9092,kafka3:9092");
// 可靠性优先：所有副本确认
props.put("acks", "all");
// 重试次数
props.put("retries", 3);
// 启用幂等性，避免生产者重试导致消息重复
props.put("enable.idempotence", true);
// 批量发送，减少网络开销
props.put("batch.size", 16384);
props.put("linger.ms", 5);

// 错误方式：多线程并发消费同一分区会打乱顺序
@KafkaListener(topics = "order-topic", groupId = "order-consumer-group")
public void consumeOrderMultiThread(OrderMessage message) {
    executor.submit(() -> processMessage(message)); // ❌ 线程池并发处理
}

// 正确方式：保持单线程消费
@KafkaListener(topics = "order-topic", groupId = "order-consumer-group",
               concurrency = "1")
public void consumeOrderSingle(OrderMessage message) {
    processMessage(message); // ✅ 单线程串行消费
}

public class OrderMessageProducer {

    public void sendOrderMessage(OrderMessage orderMessage) throws Exception {
        DefaultMQProducer producer = new DefaultMQProducer("order-producer-group");
        producer.setNamesrvAddr("namesrv1:9876;namesrv2:9876");
        producer.start();

        // 关键：根据订单ID选择队列，保证同一订单的消息进入同一队列
        Message msg = new Message(
            "order-topic",           // topic
            "order",                 // tag
            orderMessage.getOrderId(), // 订单ID作为消息key
            JSON.toJSONBytes(orderMessage)
        );

        producer.send(msg, new MessageQueueSelector() {
            @Override
            public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                // arg 就是send()方法传入的第三个参数：orderId
                String orderId = (String) arg;
                // 取模算法：同一订单ID始终路由到同一个队列
                long index = Math.abs(orderId.hashCode()) % mqs.size();
                return mqs.get((int) index);
            }
        }, orderMessage.getOrderId()); // 第三个参数会传递给select()方法的arg

        producer.shutdown();
    }
}

public class OrderConsumer {

    public void consumeOrderly() throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("order-consumer-group");
        consumer.setNamesrvAddr("namesrv1:9876;namesrv2:9876");
        consumer.subscribe("order-topic", "*");

        // 注册顺序消息监听器，内部会对每个MessageQueue加分布式锁
        consumer.registerMessageListener(new MessageListenerOrderly() {
            @Override
            public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs,
                                                        ConsumeOrderlyContext context) {
                for (MessageExt msg : msgs) {
                    try {
                        String orderId = msg.getKeys();
                        String body = new String(msg.getBody(), "UTF-8");
                        log.info("消费消息 orderId={}, body={}", orderId, body);

                        // 执行业务逻辑：下单→支付→发货
                        processBusiness(orderId, body);

                        return ConsumeOrderlyStatus.SUCCESS;
                    } catch (Exception e) {
                        log.error("消费失败 orderId={}", orderId, e);
                        // 顺序消息失败时不能跳过，否则会导致后续消息乱序
                        // 应返回SUSPEND_CURRENT_QUEUE_A_MOMENT进行短暂暂停后重试
                        return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
                    }
                }
                return ConsumeOrderlyStatus.SUCCESS;
            }
        });

        consumer.start();
    }
}

public class DelayMessageProducer {

    public void sendDelayMessage() throws Exception {
        DefaultMQProducer producer = new DefaultMQProducer("delay-producer-group");
        producer.setNamesrvAddr("namesrv1:9876;namesrv2:9876");
        producer.start();

        Message msg = new Message("order-topic", "cancel", orderId,
            JSON.toJSONBytes(orderMessage).getBytes());

        // RocketMQ延迟级别：1s, 5s, 10s, 30s, 1m, 2m, 3m, 4m, 5m, 6m,
        // 7m, 8m, 9m, 10m, 20m, 30m, 1h, 2h
        // 延迟级别15对应30分钟，适用于订单超时取消场景
        msg.setDelayTimeLevel(15); // 30分钟

        producer.send(msg);
        log.info("延迟消息发送成功，订单ID={}，延迟30分钟", orderId);
        producer.shutdown();
    }
}

public class TimeoutCancelDemo {

    // 使用Netty的HashedWheelTimer实现订单超时取消
    private final HashedWheelTimer timer = new HashedWheelTimer(
        new NamedThreadFactory("timeout-timer"),
        100,              // 每个刻度的时间跨度：100ms
        TimeUnit.MILLISECONDS,
        512               // 时间轮槽数：512个槽
    );

    public void createOrder(String orderId) {
        // 1. 创建订单（状态：待支付）
        orderService.createOrder(orderId);

        // 2. 创建20分钟超时取消任务
        timer.newTimeout(timeout -> {
            Order order = orderService.getOrder(orderId);
            if (order != null && "UNPAID".equals(order.getStatus())) {
                // 双重检查：再次确认订单状态，防止并发支付成功后又取消
                orderService.cancelOrder(orderId, "超时自动取消");
                log.info("订单 {} 超时自动取消", orderId);
            }
        }, 20, TimeUnit.MINUTES); // 延迟20分钟执行

        log.info("订单 {} 已创建，20分钟后检查支付状态", orderId);
    }
}

// Kafka消费者扩容：增加消费者实例数（不超过分区数）
// 假设order-topic有16个分区，当前有4个消费者
// 扩容到8个消费者后，每个消费者平均处理2个分区，消费能力翻倍

// 消费者组内增加实例即可自动重平衡
@KafkaListener(
    topics = "order-topic",
    groupId = "order-consumer-group",
    containerFactory = "kafkaListenerContainerFactory"
)

// RocketMQ批量消费
consumer.registerMessageListener(new MessageListenerConcurrently() {
    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                                                       ConsumeConcurrentlyContext context) {
        // msgs 最多包含32条消息（可配置）
        List<OrderMessage> orders = msgs.stream()
            .map(msg -> JSON.parseObject(new String(msg.getBody()), OrderMessage.class))
            .collect(Collectors.toList());

        // 批量入库：一次数据库操作处理多条订单
        try {
            orderService.batchInsertOrders(orders);
            log.info("批量消费成功，数量={}", orders.size());
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        } catch (Exception e) {
            log.error("批量消费失败，将重试这批消息", e);
            return ConsumeConcurrentlyStatus.RECONSUME_LATER;
        }
    }
});

// 配置批量消费参数
consumer.setConsumeMessageBatchMaxSize(32);   // 一次最多拉取32条
consumer.setConsumeThreadMin(20);             // 最小消费线程
consumer.setConsumeThreadMax(100);             // 最大消费线程

// Kafka消费监控：使用AdminClient查询消费lag
public void monitorConsumerLag() {
    AdminClient admin = AdminClient.create(props);
    Map<TopicPartition, OffsetAndMetadata> committed = 
        admin.listConsumerGroupOffsets("order-consumer-group")
             .partitionsToOffsetAndMetadata()
             .get();

    committed.forEach((tp, offset) -> {
        // 计算每个分区的积压量
        long lag = getEndOffset(tp) - offset.offset();
        log.info("分区={}, offset={}, lag={}", tp.partition(), offset.offset(), lag);
        if (lag > 10000) {
            alertService.sendAlert("消费积压严重，lag=" + lag);
        }
    });
}