丁丁书屋

Appearance

消息队列 ​

消息中间件概述 ​

什么是消息队列 ​

MQ（消息队列，MessageQueue） 是基于队列的存放消息的容器，用于实现应用程序和应用程序之间的通信。

队列，是一种 先进先出（FIFO） 的数据结构。

消息则是跨进程传递的数据。

—个典型的 MQ 系统，由 生产者 将消息发送到 MQ 进行排队，然后根据一定的顺序交由消息的 消费者 进行处理。构成典型的异步通信模型。

通过提供 消息传递 和 消息排队 模型在分布式环境下提供应用解耦、弹性伸缩、冗余存储、流量削峰、异步通信、数据同步等功能。

发送者把消息发送给消息服务器，消息服务器将消息存放在若干 队列/主题 中，在合适的时候，消息服务器会将消息转发给接受者。 在这个过程中，发送和接受是 异步 的，也就是发送无需等待，而且发送者和接受者的生命周期也没有必然关系。 image-20211224203701964

为什么使用消息队列 ​

数据交换的角度 ​

线程与线程之间，无需外部中间件，通过内存即可实现数据交换，详见 [Java 内存模型](../../Java/JVM/index.md#Java 内存模型)，通常采用 [阻塞队列](../../Java/Java并发编程/index.md#阻塞队列 BlockingQueue) 来作为数据缓存区。

image-20260911152701450

由于 JVM 启动时会向操作系统申请独立的内存空间，所以进程与进程之间无法通过内存进行数据交互，通常通过网络相互传输数据。

image-20260911153743913

消息中间件，可以解决线程之间交互缓存堆积导致内存空间不足的问题，也可以解决进程之间多次传输的网络 I/O 成本问题和无法多路传输只能与单节点耦合的问题。

image-20260911154502793

分布式的角度 ​

分布式系统中，微服务使用 RPC 远程过程调用工具 （如 Feign）调用其他微服务，属于 同步操作，微服务请求后必须等待响应，才能继续工作。

使用 RPC 存在的问题

  • 系统之间接口耦合比较严重
  • 面对大流量并发时，容易被冲垮
  • 等待同步存在性能问题

image-20260814153459487

如果使用消息队列，发送消息和接收消息变成异步操作，单个服务无需等待，消息队列使得微服务架构设计和组件之间的访问有了更灵活的实现方式。

使用消息队列的好处：

  • 异步处理

    • 快速响应，提高吞吐
    • 将耗时操作解耦并异步执行，提升响应速度与用户体验，如订单提交后立即返回。

    img

  • 功能解耦。

    • 模块间无直接调用依赖，新增或修改模块不影响其他系统，显著增强可扩展性。
    • 开发者专注完成当前业务开发，无需关心后续操作。

    img

  • 削峰限流。

    • 缓冲突发高并发请求，防止后端服务被瞬时流量压垮，适用于秒杀、促销等场景。

    • 通过消息队列暂存事务消息，由消费者按自身处理能力逐步消化。

    img

  • 便于扩展。

并不是把所有交互方式都改成异步

  • 强关联调用还是通过 OpenFeign 进行同步调用
  • 弱关联、可独立拆分出来的功能使用消息队列进行异步调用

消息队列的特点 ​

  • 采用异步处理模式
  • 应用系统之间解耦
    • 发送者和接收者不必了解对方，只需确认消息
    • 发送者和接收者不必同时在线

image-20211224204522217

消息队列的应用场景 ​

  • 顺序保证和延时处理
    • 消息队列保证数据按照特定的顺序被处理，适用于那些对数据顺序有严格要求的场景。多数主流 MQ（RocketMQ/RabbitMQ/Kafka）支持严格顺序消息，满足金融等强序场景。
    • 消息发送后不会立即被消费，而是指定一个时间，到时间后再消费。支持定时/延时消息投递，如订单超时取消、预约提醒等业务逻辑。
  • 分布式事物
    • MQ 事务是分布式事务解决方案之一，RocketMQ/Kafka/Pulsar 等均提供原子性保障。支持消费、处理、再生产消息全过程作为单一事务单元执行。
    • 2PC（二阶段提交）
    • TCC（Try-Confirm-Cancel）
  • 即时通讯与数据流处理
    • RabbitMQ 通过 MQTT 插件支持物联网低带宽环境下的稳定消息传递
    • 针对分布式系统产生的海量数据流，如业务日志、监控数据、用户行为等，消息队列可以实时或批量收集这些数据，并将其导入到大数据处理引擎中，实现高效的数据流管理和处理。

消息队列技术选型 ​

特性ActiveMQRabbitMQRocketMQKafka
语言协议基于Erlang 实现 AMQP 协议Java 实现 架构极简类 JMS 的基于 TCP 的二进制协议
单机吞吐量万级，比 RocketMQ, Kafka 低一个数量级万级，比 RocketMQ, Kafka 低一个数量级10 万级，支持 高吞吐10 万级，支持 高吞吐
Topic 数量对吞吐量的影响Topic 可以达到几百/几千量级Topic 可以达到几百量级，如果更多的话，吞吐量会大幅度下降
时效性ms 级微秒级别，延迟最低ms 级ms 级
可用性高，基于主从架构实现高可用高，基于主从架构实现高可用非常高，分布式架构非常高，分布式架构
消息可靠性有较低的概率丢失数据基本不丢经过参数优化配置，可以做到 0 丢失经过参数优化配置，可以做到 0 丢失
功能支持MQ 领域的 功能极其完备并发能力强，性能极好，延时很低MQ 功能较为完善，分布式，扩展性好功能较为简单，支持简单的 MQ 功能，在大数据领域被广泛使用
社区支持很早的软件，社区不是很活跃开源，稳定，社区活跃度高阿里开发，社区活跃度不高Apache 开源，高吞吐量，社区活跃度极高
常用场景ActiveMQ因性能差、迭代慢已被淘汰，不推荐学习或用于新项目。适合路由复杂、接入简单、中小规模吞吐且需强可靠性的企业内部系统。交易、订单等强一致性场景日志采集、实时计算等大数据场景、 JavaEE 与大数据的融合场景

消息队列使用的通信协议标准 ​

目前主流的消息队列通信协议标准包括：

  • AMQP（Advanced Message Queuing Protocol）：通用协议，IBM 公司研发。
    • 传输层与会话层
    • 支持多种消息模型，包括点对点（P2P）和发布/订阅（Pub/Sub）。
    • 二进制协议进行消息传递，提高了高效、可靠的消息投递机制。
    • 典型代表——RabbitMQ。
  • MQTT（Message Queuing Telemetry Transport）：物联网小型设备之间传输消息的轻量级通讯协议。
  • JMS（Java Message Service）：专门为 Java 语言服务，SUN 公司研发，一组由 Java 接口组成的 Java 标准。
    • 应用层
    • 主要支持点对点和发布/订阅两种 消息模型。
    • 种面向文本的协议（如 HTTP 或 TCP），消息的传输效率可能较低。
    • 典型代表——ActiveMQ。

AMQP 是更底层的、跨语言的“通信协议”，而 JMS 是更高层的、Java 专用的“编程接口”

AMQP 和 MQTT 都可以作为 JMS 的底层协议。

JMS ​

JMS 概述 ​

Java Message Service 简称 JMS，即 Java 消息服务应用程序接口，是一个 Java 平台中关于面向消息中间件（Message-Oriented Middleware，MOM）的 API，用于在两个应用程序之间，或分布式系统中发送消息，进行 异步通信。它提供了一个可以通过网络访问的抽象消息队列，与具体的消息队列平台无关，绝大多数 MOM 提供商都对 JMS 提供支持，例如 RabbitMQ、ActiveMQ、RocketMQ、Kafka。

JMS 之于 消息中间件，就像 JDBC 之于 数据库。

JMS 的组成 ​

JMS 由 JMS provider、JMS producer、JMS consumer、JMS message 组成。

  • JMS provider：实现 JMS 接口和规范的消息中间件 ，也就是我们的 MQ 服务。
  • JMS producer：消息生产者，创建和发送 JMS 消息的客户端应用。
  • JMS consumer：消息消费者，接收和处理 JMS 消息的客户端应用。
  • JMS message 由消息头、消息体、消息属性 3 部分组成。
    • 消息头有 5 个比较常见的属性，分别是 JMSDestination、JMSDeliveryMode、JMSExpiration、JMSPriority、JMSMessageID。
    • 消息体：封装了具体的消息数据，消息体有五种消息格式，分别是 TextMessage、MapMessage、BytesMessage、StreamMessage、ObjectMessage，发送和接收的消息体类型必须一致对应。
    • 消息属性：消息属性是以属性名和属性值对的形式制定的，可以将属性视为消息头的扩展，属性指定一些消息头没有包括的附加信息。

JMS 标准消息模型 ​

P2P 模型，即点对点模型，不管消费者有多少个，消息队列中的每一条消息只能被一个消费者消费掉，通常还支持消息反馈机制

image-20260911161050312

Public/Subscribe 模型，即发布/订阅模型，多个消费者可以订阅一个主题，并且可以同时消费一样的数据。

image-20260911161327698

AMQP ​

AdvancedMessageQueuingProtocol 是跨语言、跨平台的二进制线路层协议，RabbitMQ 为其代表实现。

支持 direct/fanout/topic/headers/system 五种 exchange 路由模型，比 JMS 更细粒度控制消息分发。

消费者的推拉模式 ​

推拉模式特指 Consumer（消费者）与 Broker（服务端）之间的交互。Producer（生产者）默认都是推给 Broker，因为依赖成干上万的 Producer 自己存数据并等拉取，可靠性无法保证。

推模式 ​

推模式指的是消息从 Broker 推向 Consumer，即 Consumer 被动的接收消息，由 Broker 来主导消息的发送。

  • 优点：实时性极高，消息一来立即推送；且消费者使用简单（被动等待即可）。
  • 致命缺点：速率难匹配。Broker 无法感知消费者能力，推送过快会像 DDoS 一样压垮消费者（爆仓），且 Broker 需维护每个消费者的状态，复杂度高。

拉模式 ​

拉模式指的是主动向 Broker 请求拉取消息，即 Broker 被动的发送消息给 Consumer。

  • 优点：消费者自主可控，消费不过来就暂停拉取，能自我保护；Broker 职责轻（纯存储），且便于消费者按需批量拉取。
  • 致命缺点：存在延迟和空耗。消费者不知道消息何时到，需轮询。拉取间隔大则延迟高，间隔小则变成“空请求”攻击 Broker(忙请求)。

主流的选择 ​

RocketMQ 和 Kafka 都选择了拉模式，因为 Broker 作为中心节点应尽量轻量，不该依赖消费者状态。

为了弥补拉模式的缺点，它们都引入了 长轮询（LongPolling）：消费者请求后，若没消息，Broker 不立即返回，而是挂起等待（如等待 5s 或 30s）。一旦新消息到达或超时，再立即返回。这既实现了近实时的效果（有消息秒回），又大幅减少了无效的空轮询（没消息时请求不空跑）。

消息队列可靠性保证 ​

异常情况：Broker 宕机、网络分区

常见问题：重复消费、消息丢先、顺序错乱、事务一致性

全链路可靠性设计 ​

  1. 生产者阶段：启用 发送确认 ACK、失败重试，关键业务落本地消息表或使用事务消息。
  2. Broker 存储阶段：配置持久化、[多副本](./Kafka/index.md#Broker 和集群)、合理刷盘策略与副本确认机制。

消费与兜底策略 ​

  • 消费者须业务成功后再 ACK 或 提交 offset，失败则重试、进死信队列或触发补偿。
  • 业务层需通过对账、补偿任务、告警及人工介入兜底极端异常场景。

重复消费应对 ​

  • 幂等设计是必需项，常见方案包括唯一索引、状态机流转、消费记录表、Redis 短时去重。
  • 应基于业务唯一键（如订单号）而非 MQ 自动生成 ID，确保重试/补偿场景下仍有效。

消息积压应对 ​

  • 先定位根因：生产突增、消费者变慢、分区/队列不足或 Broker 异常。
  • 处理顺序为限流止血一扩容消费者与分区一优化慢消费逻辑一批处理历史积压。