📨 消息队列选型实战

// ===== Kafka 生产者:发送用户行为日志 ===== import org.apache.kafka.clients.producer.*; public class LogProducer { public static void main(String[] args) { // 1. 生产者配置(关键参数都加了注释) Properties props = new Properties(); props.put("bootstrap.servers", "kafka1:9092,kafka2:9092,kafka3:9092"); props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer"); // ⚠️ acks=all:所有副本都写入才返回成功,保证消息不丢 props.put("acks", "all"); // ⚠️ 重试次数:网络抖动时自动重试 props.put("retries", 3); // ⚠️ 批量发送:提升吞吐,攒一批再发 props.put("batch.size", 16384); props.put("linger.ms", 10); // ⚠️ 幂等生产者:防止重试导致消息重复 props.put("enable.idempotence", "true"); try (KafkaProducer<String, String> producer = new KafkaProducer<>(props)) { for (int i = 0; i < 100; i++) { // 2. 构造日志消息:JSON 格式 String log = "{\"userId\":\"u" + i + "\",\"action\":\"click\",\"ts\":1700000000}"; // 3. 用 userId 做 key → 相同用户进同一分区,保证用户级行为有序 ProducerRecord<String, String> record = new ProducerRecord<>("user-behavior-log", "u" + i, log); // 4. 异步发送 + 回调(失败要记录,不能吞掉) producer.send(record, (metadata, e) -> { if (e != null) { System.err.println("发送失败: " + e.getMessage()); // TODO: 落库重发 / 死信队列 } else { System.out.printf("offset=%d, partition=%d%n", metadata.offset(), metadata.partition()); } }); } producer.flush(); // 强制把缓冲区消息发出去 } } } // ===== Kafka 消费者:消费日志写入 HDFS/Flink ===== public class LogConsumer { public static void main(String[] args) { Properties props = new Properties(); props.put("bootstrap.servers", "kafka1:9092"); props.put("group.id", "log-hdfs-consumer-group"); props.put("enable.auto.commit", "false"); // ⚠️ 关自动提交 props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); // ⚠️ earliest:从头消费;latest:只消费新消息(默认) props.put("auto.offset.reset", "earliest"); try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props)) { consumer.subscribe(Arrays.asList("user-behavior-log")); while (true) { // 1. 拉取消息(批量),超时 1 秒 ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000)); for (ConsumerRecord<String, String> record : records) { try { // 2. 处理消息:写 HDFS / 喂 Flink processLog(record.value()); } catch (Exception e) { // ⚠️ 处理失败不能提交 offset,下次会重投 System.err.println("处理失败,稍后重试: " + e.getMessage()); return; } } // 3. 手动提交 offset(处理成功后才提交) consumer.commitSync(); } } } }

// ===== 下单服务:用事务消息保证"扣库存 + 发消息"原子性 ===== @Service public class OrderService { @Autowired private TransactionMQProducer transactionProducer; public void createOrder(OrderDTO order) throws Exception { // 1. 构造消息:Topic = order-topic, Tag = create Message msg = new Message( "order-topic", // Topic "create", // Tag(子分类,可按 Tag 订阅) order.getUserId(), // Key(用于查询/幂等) JSON.toJSONString(order).getBytes() // body ); // 2. 发送事务消息(半消息)— 这一步会先发到 MQ,但消费者看不到 TransactionSendResult result = transactionProducer.sendMessageInTransaction( msg, new OrderTransactionListener() // 事务监听器 ); if (result.getLocalTransactionState() == LocalTransactionState.COMMIT_MESSAGE) { System.out.println("事务提交,消息可被消费"); } else { System.out.println("事务回滚,消息会被删除"); } } } // ===== 事务监听器:执行本地事务 + 回查 ===== public class OrderTransactionListener implements TransactionListener { @Autowired private OrderMapper orderMapper; // 本地数据库 // ① 执行本地事务(半消息发送成功后回调) @Override public LocalTransactionState executeLocalTransaction(Message msg, Object arg) { try { OrderDTO order = JSON.parseObject(new String(msg.getBody()), OrderDTO.class); // 执行本地事务:写订单表 + 扣库存 orderMapper.insert(order); stockService.deduct(order.getProductId(), order.getQty()); // 本地事务成功 → 提交消息,消费者能看到 return LocalTransactionState.COMMIT_MESSAGE; } catch (Exception e) { // 本地事务失败 → 回滚消息,消息删除 return LocalTransactionState.ROLLBACK_MESSAGE; } } // ② 事务回查(MQ 主动问:你那个事务到底成没成?) @Override public LocalTransactionState checkLocalTransaction(MessageExt msg) { // ⚠️ 如果 executeLocalTransaction 超时/挂了,MQ 会回查 String orderId = msg.getKeys(); Order order = orderMapper.selectById(orderId); if (order != null) { // 订单已落库 → 事务成功 → 提交 return LocalTransactionState.COMMIT_MESSAGE; } // 订单不存在 → 事务未完成 → 回滚 return LocalTransactionState.ROLLBACK_MESSAGE; } } // ===== 库存消费者:消费订单消息,扣库存 ===== @Component @RocketMQMessageListener(topic = "order-topic", consumerGroup = "stock-consumer-group") public class StockConsumer implements RocketMQListener<OrderDTO> { @Override public void onMessage(OrderDTO order) { // ⚠️ 幂等性:防止重复消费扣两次库存 boolean isNew = redis.setIfAbsent( "order:processed:" + order.getId(), "1", 24, HOURS ); if (!isNew) return; // 已处理过,跳过 stockService.deduct(order.getProductId(), order.getQty()); } }

// ===== 场景:下单 30 分钟未支付,自动取消订单 ===== @Service public class OrderTimeoutService { @Autowired private DefaultMQProducer producer; public void sendDelayMessage(String orderId) throws Exception { Message msg = new Message( "order-timeout-topic", "timeout", orderId, orderId.getBytes() ); // ⚠️ RocketMQ 延迟级别:1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h // 30 分钟对应级别 16 msg.setDelayTimeLevel(16); // 30 分钟后投递 producer.send(msg); // 发送后,MQ 会在 30 分钟后才把消息投递给消费者 } } // ===== 消费者:30 分钟后收到,检查支付状态,未支付就取消 ===== @Component @RocketMQMessageListener(topic = "order-timeout-topic", consumerGroup = "order-cancel-group") public class OrderCancelConsumer implements RocketMQListener<String> { @Override public void onMessage(String orderId) { Order order = orderMapper.selectById(orderId); if (order.getStatus() == OrderStatus.UNPAID) { // 仍未支付 → 取消订单 + 回滚库存 orderMapper.updateStatus(orderId, OrderStatus.CANCELLED); stockService.rollback(order.getProductId(), order.getQty()); } // 已支付 → 忽略 } }

// ===== RabbitMQ 配置:Topic Exchange + 多队列绑定 ===== @Configuration public class RabbitConfig { public static final String ORDER_EXCHANGE = "order.exchange"; public static final String SMS_QUEUE = "order.sms.queue"; public static final String EMAIL_QUEUE = "order.email.queue"; public static final String ANALYSIS_QUEUE = "order.analysis.queue"; // 1. 声明 Topic Exchange(按 routing key 模式匹配) @Bean public TopicExchange orderExchange() { return new TopicExchange(ORDER_EXCHANGE, true, false); // durable=true 持久化 } // 2. 声明队列(durable=true 持久化,防重启丢消息) @Bean public Queue smsQueue() { return QueueBuilder.durable(SMS_QUEUE).build(); } @Bean public Queue emailQueue() { return QueueBuilder.durable(EMAIL_QUEUE).build(); } @Bean public Queue analysisQueue() { return QueueBuilder.durable(ANALYSIS_QUEUE).build(); } // 3. 绑定:不同 routing key 路由到不同队列 @Bean public Binding smsBinding() { // order.created.* → 匹配 order.created.sms / order.created.any return BindingBuilder.bind(smsQueue()).to(orderExchange()).with("order.created.*"); } @Bean public Binding emailBinding() { return BindingBuilder.bind(emailQueue()).to(orderExchange()).with("order.created.*"); } @Bean public Binding analysisBinding() { // order.# → 匹配 order.created / order.paid / order.cancelled 所有事件 return BindingBuilder.bind(analysisQueue()).to(orderExchange()).with("order.#"); } } // ===== 生产者:发送订单事件 ===== @Service public class OrderEventProducer { @Autowired private RabbitTemplate rabbitTemplate; public void sendOrderCreated(Order order) { // ⚠️ 关键配置 1:消息持久化 MessagePostProcessor persistence = msg -> { msg.getMessageProperties().setDeliveryMode(MessageDeliveryMode.PERSISTENT); return msg; }; // ⚠️ 关键配置 2:发布确认(确保消息到了 MQ) rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> { if (!ack) { System.err.println("消息未到达 Exchange: " + cause); // TODO: 重发 / 落库补偿 } }); // ⚠️ 关键配置 3:return 回调(Exchange 到不了 Queue 时触发) rabbitTemplate.setReturnsCallback(returned -> { System.err.println("消息无法路由: " + returned.getReplyText()); }); // 发送:routing key = order.created.sms rabbitTemplate.convertAndSend( RabbitConfig.ORDER_EXCHANGE, "order.created.sms", order, persistence ); } } // ===== 消费者:短信服务(手动 ACK) ===== @Component public class SmsConsumer { @RabbitListener(queues = RabbitConfig.SMS_QUEUE) public void handle(Order order, Channel channel, long tag) throws IOException { try { // 业务处理:发短信 smsService.send(order.getUserPhone(), "您的订单已创建"); // ⚠️ 手动 ACK:处理成功才确认 channel.basicAck(tag, false); } catch (Exception e) { // ⚠️ 处理失败:requeue=false 进死信队列,不重新排队 channel.basicNack(tag, false, false); } } }