# 简介
# MQ
软件设计模式中,发布/订阅模式、生产/消费模式都是用于让多个程序相互通信,需要使用一个 broker 来暂存消息。
- 可以使用 MySQL 等数据库软件,担任 broker ,但这样功能少。
- 推荐使用专业的 broker 软件,比如 ActiveMQ、Kafka ,它们统称为 MQ(Message Queue ,消息队列)。
MQ 软件的主要难点:
- 如何保证不丢失消息?
- broker 会记录每个 consumer 消费到了第几条消息。即使 consumer 重启,也可以继续消费,不会丢失消息。
- 但 producer 生产消息时、broker 存储消息时,可能丢失消息。
- 如何保证消息不被重复消费?
- 可以给每个消息分配一个唯一 ID ,consumer 记录自己获得的所有消息 ID ,如果收到重复的消息就忽略。
- 或者 broker 记录每个 consumer 当前消费的消息 ID ,不发送重复的消息。
- 性能怎么样,能否处理高并发消息?
- 如何保证不丢失消息?
# 设计模式
# 观察者模式
假设程序 A 想发送一条消息(比如发送一个邮件),给程序 B 。普通的解决方案是,直接通信。
- 原理:
- 程序 A 直接调用程序 B 的 API ,发送消息。
- 比如程序 A 发送 POST 类型的 HTTP 请求,给程序 B ,并将消息放在 HTTP 请求报文的 body 中。
- 优点:
- 原理简单。
- 缺点:
- 两者耦合
- 两者同步工作
- 程序 A 每发送一条消息,需要等待程序 B 返回响应,证明它收到消息,然后才能发送下一条消息。
- 原理:
另一种解决方案是,轮询。
- 原理:
- 让程序 B 定期调用程序 A 的 API ,获取消息。
- 比如程序 B 发送 GET 类型的 HTTP 请求,给程序 A 。而程序 A 将消息,放在 HTTP 响应报文的 body 中,交给程序 B 。
- 优点:
- 两者异步工作
- 程序 A 新增消息时,不必立即发送给程序 B ,而是等程序 B 自己来拿。
- 两者异步工作
- 缺点:
- 两者耦合
- 如果轮询的间隔较短,则需要经常轮询,开销较大。
- 如果轮询的时间间隔较长,则延迟较大。程序 A 新增一条消息之后,等较长时间,才会被程序 B 获取。
- 原理:
另一种解决方案是,观察者模式。
- 原理:
- 一个程序,负责发送消息,称为发布者(publisher)。
- 一个或多个程序,负责接收消息,称为订阅者(subscriber)。
- publisher 提供一套关于订阅(subscribe)的 API 。
- subscriber 可以调用 publisher 的上述 API ,声明自己希望接收哪些消息。比如只接收关于某主题的消息,不接收其它消息。
- publisher 每次新增一条消息时,会检查哪些 subscriber 订阅了该消息,然后将该消息发送给这些 subscriber 。
- 如果不存在订阅机制,则 publisher 会将所有新消息都发给 subscriber ,而这些消息不一定对 subscriber 有用。
- 优点:
- 支持一对多通信
- 允许一个 publisher ,将同一消息,发送给多个 subscriber 。
- 同步工作
- publisher 每发送一条消息,需要等待 subscriber 返回响应,证明它收到消息,然后才能发送下一条消息。
- 支持一对多通信
- 缺点:
- 两者耦合
- 容易丢失消息
- 如果 publisher 向 subscriber 推送数据时,subscriber 暂时故障(比如正在重启),则 publisher 通常不会等待 subscriber 恢复,而是开始推送下一条消息。顶多记录一条日志,表示某条消息发送失败。
- 原理:
# 发布/订阅模式
发布/订阅模式,是在观察者模式的基础上,引入中介模式。不让 publisher 与 subscriber 直接通信,而是引入一个 broker 作为中介。
架构:
- 运行一个程序,担任发布者(publisher),将消息发送到 broker 。
- 运行另一个程序,担任订阅者(subscriber),从 broker 获取消息。
- 运行一个代理(broker),负责暂存、转发消息。
- subscriber 可以调用 broker 的 API 来订阅消息,声明该 subscriber 希望接收哪些消息,比如关于某主题的消息。
- broker 每次新增一条消息时,会检查哪些 subscriber 订阅了该消息,然后将该消息发送给这些 subscriber 。
优点:
- 支持多对多通信
- 允许多个 publisher 发送消息到 broker ,这些消息可以推送给多个 subscriber 。
- 解耦
- publisher 与 subscriber 之间没有直接调用,没有直接耦合。虽然它们分别与 broker 耦合,但 broker 一般很少故障。
- 异步工作
- publisher 与 subscriber 之间异步工作,不过它们分别与 broker 同步工作。
- 支持多对多通信
缺点:
- 容易丢失消息
# 生产/消费模式
- 生产/消费模式,与发布/订阅模式相似,都是用于多对多通信。
- 原理:
- 运行一个代理(broker),负责暂存、转发消息。
- 运行一个程序,担任生产者(producer),将消息发送到 broker 。
- 运行另一个程序,担任消费者(consumer),从 broker 获取消息。
- 发布/订阅模式中,是由 broker 主动向 subscriber 推送消息。如果 broker 忘了推送,则不会传输消息。
- 生产/消费模式中,是由 consumer 主动从 broker 拉取消息。如果 consumer 忘了拉取,则不会传输消息。
- 优点:
- 支持多对多通信
- 解耦
- 异步工作
- 如果 producer 与 consumer 直接通信,则属于同步工作。producer 每发送一条消息,需要等待 consumer 返回响应,证明它收到消息,然后才能发送下一条消息。
- 支持获取历史消息
- broker 通常会暂存最近的一连串消息,因此 consumer 既可以获取最新一条消息,也可以获取过去一段时间的消息。
- 削峰
- 即使 producer 突然发送了很多消息,consumer 依然会按照自己的速度从 broker 获取消息,不怕 consumer 的负载过大。
- 缺点:
- 增加了系统的复杂度,需要多考虑一个 broker 中间件。
- 处理消息的延迟可能很大。
- 如果 broker 堆积的消息很多,则 consumer 需要获取多条消息,才能拿到最新的那条消息。换句话说,最新的那条消息,在 broker 中等了一段时间,才被 consumer 获取。
- 有的软件追求实时性,希望一旦出现消息,就被立即处理。
# 通信协议
# JMS
JMS(Java Message Service ,Java 消息服务):一个 API 规范,描述了 Java 程序如何调用消息中间件。
于 1998 年发布。
JMS 只支持 Java 编程语言。不过很多 MQ 软件提供了兼容 JMS 规范的 MQ 协议,或者借鉴了 JMS 规范。
基本概念:
- 消息:程序之间的通信内容。
- 客户端:指连接到 MQ 服务器的程序。可细分为消息的发送方(Sender)、接收方(Receiver)。
JMS 定义了两种传输消息的模式:
- 发布/订阅模式(Publish/Subscribe)
- 每条消息可以存在多个接收方。
- 点对点模式(Point to Point ,P2P)
- 与生产/消费模式相似,但每条消息只能存在一个接收方。
- 发布/订阅模式(Publish/Subscribe)
# AMQP
AMQP(Advanced Message Queuing Protocol ,高级消息队列协议)
- 于 2003 年发布。最初由摩根大通主导,用于解决金融公司之间的消息传递问题。
- 属于应用层协议,基于 TCP 通信。
- 与 JMS 相比,AMQP 是一个通用的 MQ 协议,跨语言。
# MQTT
MQTT(Message Queuing Telemetry Transport,消息队列遥测传输):一个 MQ 通信协议,主要用于物联网。
- 于 1999 年发布。
- 属于应用层协议,基于 TCP 通信。
- 适用于低带宽、不可靠网络中的通信。
- 采用 C/S 架构。
- mqtt server 又称为 mqtt broker 。常见的服务器软件有 ActiveMQ、mosquitto、EMQ 等。
- mqtt client 与 mqtt broker 建立连接之后就构成了一个会话(Session)。
- 采用发布/订阅模式。
- client 可以担任 Publisher ,发布消息到 mqtt broker 的某个 topic 下。
- client 可以担任 Subscriber ,订阅 mqtt broker 的某个 topic 下的新增消息。
# 常见软件
ActiveMQ
- 是第一个实现 JMS 规范的开源 MQ 软件,后来也支持 AMQP、STOMP、MQTT 等协议。
- 于 2004 年发布,采用 Java 语言开发。后来该项目交给 ASF 基金会管理。
RabbitMQ
- 是第一个实现 AMQP 协议的开源 MQ 软件,后来也支持 JMS、STOMP、MQTT 等协议。
- 于 2007 年发布,采用 erlang 语言开发。后来该项目被 VMWare 公司收购。
- 默认监听 TCP 5672 端口。
- 并发能力强,延迟很低。
Kafka
- 2011 年由 LinkedIn 公司开源,后来交给 ASF 基金会管理。
- 主要开发者离职后创建了 Confluent 公司,提供功能更多的 Kafka 发行版,分为开源版、企业版。
- 采用 Scala 语言开发。
- 吞吐量比 ActiveMQ、RabbitMQ 高了一个数量级,因此更适合处理大数据。
- 2011 年由 LinkedIn 公司开源,后来交给 ASF 基金会管理。
RocketMQ
- 2012 年由阿里巴巴公司开源,后来交给 ASF 基金会管理。
- 采用 Java 语言开发,借鉴了 Kafka、RabbitMQ 。