Day21 | 消息队列基础面试真题解析

// 开启Confirm模式channel.confirmSelect();channel.basicPublish("exchange", "routingKey", null, message.getBytes());// 等待确认（同步方式，性能较差）boolean confirmed = channel.waitForConfirms();if (confirmed) {    System.out.println("✅ 消息发送成功");} else {    System.out.println("❌ 消息发送失败，需要重试");}// 异步Confirm（推荐）channel.addConfirmListener(    (deliveryTag, multiple) -> {        System.out.println("✅ 消息已确认: " + deliveryTag);    },    (deliveryTag, multiple) -> {        System.out.println("❌ 消息未确认: " + deliveryTag);    });

// 关闭自动确认，开启手动Ackchannel.basicConsume("queue", false, (consumerTag, delivery) -> {    try {        // 处理消息业务逻辑        processMessage(delivery.getBody());        // 手动确认消息        channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);    } catch(Exception e) {        // 处理失败，拒绝消息并可选择是否重新入队        channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, true);    }}, consumerTag -> {});

-- 去重表设计  CREATE TABLE idempotent_log (      msg_id VARCHAR(64) PRIMARY KEY,      biz_type VARCHAR(32),      create_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP  );  // 消费时先插入去重表，利用唯一索引保证幂等性  @Transactional  public void consumeMessage(String msgId, OrderMessage msg) {      try {          // 插入去重记录，重复消息会抛出DuplicateKeyException          idempotentLogMapper.insert(new IdempotentLog(msgId, "ORDER"));          // 执行业务逻辑          orderService.process(msg);      } catch (DuplicateKeyException e) {          // 已处理过，直接返回          log.info("消息 {} 已处理，跳过", msgId);      }  }

// 使用Redis SetNX命令实现幂等  public boolean idempotentConsume(String msgId, Consumer<String> consumer) {      String key = "mq:idempotent:" + msgId;      // SET key value NX EX 600      Boolean success = redisTemplate.opsForValue()          .setIfAbsent(key, "1", Duration.ofMinutes(10));      if (Boolean.TRUE.equals(success)) {          consumer.accept(msgId);          return true;      }      log.info("消息 {} 重复消费，已跳过", msgId);      return false;  }

@Transactional  public void payOrder(Long orderId) {      Order order = orderMapper.selectById(orderId);      // 只有待支付状态才执行支付      if (order != null && "PENDING".equals(order.getStatus())) {          orderMapper.updateStatus(orderId, "PAID");          // 其他业务逻辑...      }  }

// 自定义Partitioner，按userId取模  public class UserIdPartitioner implements Partitioner {      @Override      public int partition(String topic, Object key, byte[] keyBytes,                           Object value, byte[] valueBytes, Cluster cluster) {          int partitionCount = cluster.partitionCountForTopic(topic);          return Math.abs(key.hashCode()) % partitionCount;      }  }  // 发送消息时指定key  producer.send(new ProducerRecord<>("order-topic", userId, orderMessage));

 // 消费者配置：关闭自动提交，单线程消费  properties.put("max.poll.records", "1");  properties.put("enable.auto.commit", "false");  // 同步处理完一条再拉取下一条  while (true) {      ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));      for (ConsumerRecord<String, String> record : records) {          process(record);          consumer.commitSync(); // 同步提交offset      }  }

 // RabbitMQ 死信队列配置示例  @Bean  public Queue mainQueue() {      return QueueBuilder.durable("main.queue")          .withArgument("x-dead-letter-exchange", "dlx.exchange")          .withArgument("x-dead-letter-routing-key", "dlx.routing.key")          .withArgument("x-message-ttl", 30000)  // 30秒过期          .build();  }

// 创建延迟队列（设置TTL为30分钟）  @Bean  public Queue delayQueue() {      return QueueBuilder.durable("delay.queue")          .withArgument("x-dead-letter-exchange", "normal.exchange")          .withArgument("x-dead-letter-routing-key", "normal.routing.key")          .withArgument("x-message-ttl", 30 * 60 * 1000)  // 30分钟          .build();  }  // 消息先发送到延迟队列，过期后自动转发到正常队列  rabbitTemplate.convertAndSend("delay.exchange", "delay.routing.key", message);

 // RocketMQ 延迟消息示例  Message msg = new Message("TopicTest", "TagA", "OrderID188", "Hello".getBytes());  // 设置延迟级别（RocketMQ预设18个级别）  // 1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h  msg.setDelayTimeLevel(5);  // 延迟1分钟  producer.send(msg);