Java - 消息中间件篇

5.1 RabbitMQ 核心概念

MQ（Message Queue）= 跨进程、异步通信的中间件。类比智能快递中转站。

三大核心作用：系统解耦、异步处理、流量削峰。

5.2 RabbitMQ 消息可靠性

如何保证消息不丢失？

① 生产者确认机制

基于 Publisher Confirm 机制，根据操作结果返回不同状态。

② 消息持久化

MQ 默认内存存储，开启持久化可确保消息不丢失。

③ 消费者确认机制

消费者处理成功后发送 ACK 回执，MQ 收到 ACK 后才删除消息。

四重保障总结：

1. 开启生产者确认机制
2. 开启持久化功能
3. 开启消费者确认机制为 auto
4. 开启消费者失败重试，多次失败后投递到异常交换机，人工处理

5.3 RabbitMQ 重复消费 / 死信队列 / 消息堆积

重复消费

解决方案：

• 每条消息设置唯一标识 id，已消费的不再处理
• 幂等方案：分布式锁、数据库锁（悲观锁/乐观锁）

死信交换机（延迟队列）

死信（Dead Letter）产生的条件：

1. 消费者 basic.reject / basic.nack 且 requeue = false
2. 消息超时无人消费
3. 队列满了，最早的消息被挤出

// 发送带 TTL 的消息
Message message = MessageBuilder
    .withBody("hello, ttl message".getBytes(StandardCharsets.UTF_8))
    .setExpiration("5000")
    .build();
CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
rabbitTemplate.convertAndSend("ttl.direct", "ttl", message, correlationData);

理解：过期 ≠ 销毁。就像支付取消 — 过期后不是直接销毁，而是经由死信交换机转发到业务队列执行取消逻辑。

消息堆积

生产者发送速度 > 消费者处理速度 → 队列积压 → 超出上限 → 消息成为死信

三种解决思路：

1. 增加更多消费者，提高消费速度
2. 消费者内开启线程池加速处理
3. 扩大队列容积

惰性队列：

• 收到消息后直接存入磁盘（非内存）
• 消费者消费时才从磁盘读取并加载到内存
• 支持数百万条消息存储

⚠️ 缺点：受限于磁盘 IO，时效性会降低。

RabbitMQ 高可用

仲裁队列：基于 Raft 算法变种，多数派写入，强一致性，避免脑裂。

5.4 Kafka 消息可靠性

基本架构

• Broker：服务器节点
• Topic：消息分类的逻辑概念
• Partition：Topic 被切分成多个分区分布在不同的 Broker 上，实现负载均衡和水平扩展

① 生产者发送阶段

解决：异步发送 + 重试机制

// 异步发送
kafkaProducer.send(record, new Callback() {
    @Override
    public void onCompletion(RecordMetadata recordMetadata, Exception e) {
        if (e != null) {
            System.out.println("消息发送失败 | 记录日志");
        }
    }
});

// 设置重试
prop.put(ProducerConfig.RETRIES_CONFIG, 10);

② Broker 存储阶段

解决：设置 acks=all，让所有副本都参与保存数据后确认。

③ 消费者接收阶段

解决：关闭自动提交偏移量，改为手动提交（同步 + 异步）。

Kafka 保证消息不丢失总结

5.5 Kafka 顺序性 / 高可用 / 存储 / 高性能

消息顺序性

Topic 分区内消息按先后顺序执行，但不能保证跨分区顺序。

解决方案：发送时指定分区号 / 按照相同业务设置相同 key，确保同类消息进同一分区。

高可用机制

普通 Follower 异步更新，ISR（In-Sync Replicas）同步更新。

文件存储机制

数据清理两种策略：

1. 按时间（默认 168 小时）
2. 按大小（Topic 日志文件超过阈值则删除最旧消息，需手动开启）

高性能设计

零拷贝：

原始：用户空间 → 硬件 → 用户空间 → 网卡 → 消费者（4 次拷贝）
优化后：页缓存 → 网卡 → 消费者（直接传输，无需经过用户空间）