📨 消息队列场景决策

@Component @RequiredArgsConstructor public class OrderConsumer { private final DeduplicationMapper dedupMapper; private final OrderMapper orderMapper; @RabbitListener(queues = "order.queue") public void handleOrder(Message message) { String msgId = message.getMessageProperties().getMessageId(); // 1. 检查是否已处理 if (dedupMapper.existsById(msgId)) { log.info("重复消息，跳过: {}", msgId); return; // 幂等：重复消息不重复执行 } // 2. 业务处理 OrderEvent event = JSON.parseObject( new String(message.getBody()), OrderEvent.class); orderMapper.updateStatus(event.getOrderId(), "PAID"); // 3. 记录已处理（同一事务） dedupMapper.insert(new DedupRecord(msgId, LocalDateTime.now())); } } // 去重表DDL // CREATE TABLE msg_dedup ( // msg_id VARCHAR(64) PRIMARY KEY, -- 消息唯一ID // created_at DATETIME // );

@Component public class RedisDedupService { @Autowired private StringRedisTemplate redis; /** * 幂等检查：SETNX + 过期时间 * @return true=首次处理 false=重复消息 */ public boolean tryAcquire(String msgId) { Boolean success = redis.opsForValue() .setIfAbsent( "dedup:" + msgId, // key "1", // value Duration.ofDays(7) // 7天过期自动清理 ); return Boolean.TRUE.equals(success); } }

// 发送时指定队列选择器 — 同一订单ID的消息进同一Queue sendResult = producer.send(msg, new MessageQueueSelector() { @Override public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) { Long orderId = (Long) arg; int index = (int) (orderId % mqs.size()); return mqs.get(index); // 同一orderId永远进同一Queue } }, orderId); // 消费时用 OrderlyMessageListener — 单Queue单线程消费 @RocketMQMessageListener( topic = "order-topic", consumerGroup = "order-consumer", consumeMode = ConsumeMode.ORDERLY // ← 顺序消费 ) public class OrderConsumer implements RocketMQListener<String> { ... }

// 1. 定义死信交换机和队列 @Bean public DirectExchange dlxExchange() { return new DirectExchange("order.dlx"); } @Bean public Queue dlqQueue() { return QueueBuilder.durable("order.dlq").build(); } // 2. 业务队列绑定死信 @Bean public Queue orderQueue() { return QueueBuilder.durable("order.queue") .withArgument("x-dead-letter-exchange", "order.dlx") .withArgument("x-dead-letter-routing-key", "order.dlq") .build(); } // 3. 消费重试3次后进死信 // application.yml // spring: // rabbitmq: // listener: // simple: // retry: // max-attempts: 3 // enabled: true // 4. 死信消费者 — 人工处理或告警 @RabbitListener(queues = "order.dlq") public void handleDeadLetter(Message message) { log.error("死信消息: {}", new String(message.getBody())); // 发送告警 / 存入数据库 / 人工处理界面 }

// 延迟交换机 + 延迟队列 @Bean public Queue delayQueue() { return QueueBuilder.durable("order.delay") .withArgument("x-message-ttl", 30 * 60 * 1000) // 30分钟TTL .withArgument("x-dead-letter-exchange", "order.exchange") .withArgument("x-dead-letter-routing-key", "order.queue") .build(); } // 发送延迟消息 rabbitTemplate.convertAndSend("order.delay.exchange", "order.delay", orderEvent); // 30分钟后消息过期 → 进入死信 → 被order.queue消费 // 消费者检查订单状态：还是"待支付"→ 取消订单 @RabbitListener(queues = "order.queue") public void handleOrderTimeout(OrderEvent event) { Order order = orderMapper.selectById(event.getOrderId()); if (order != null && order.getStatus().equals("PENDING")) { order.setStatus("CANCELLED"); orderMapper.updateById(order); // 恢复库存 } }

// RocketMQ 支持固定延迟等级 // 1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h Message msg = new Message("order-topic", "tag", JSON.toJSONBytes(orderEvent)); msg.setDelayTimeLevel(16); // 16 = 30分钟 producer.send(msg); // 30分钟后消费者才会收到这条消息

// Step 1: 先看积压量 // RabbitMQ: rabbitmqctl list_queues name messages // Kafka: kafka-consumer-groups --describe --group xxx // Step 2: 快速扩容消费者 // K8s: kubectl scale deployment order-consumer --replicas=10 // Step 3: 如果扩容也来不及 → 转发到新Topic批量消费 // 创建临时消费者，快速消费积压消息，转发到新Topic @RabbitListener(queues = "order.queue.backlog", concurrency = "20") public void fastConsume(Message message) { // 转发到新Topic，用批量消费者慢慢处理 rabbitTemplate.convertAndSend("order.batch.queue", message.getBody()); } // Step 4: 积压清完后恢复正常消费者数 // kubectl scale deployment order-consumer --replicas=2 // ⚠️ 预防措施 // 1. 监控队列深度，积压超阈值告警 // 2. 设置队列最大长度，防止磁盘打满 // 3. 压测确认消费能力