关于后端:消息队列系统学习

博客:cbb777.fun

全平台账号: 安妮的心动录

github: https://github.com/anneheartrecord

下文中我说的可能对，也可能不对，鉴于笔者程度无限，请君自辨。有问题欢送大家找我探讨

什么是音讯队列

音讯队列，咱们个别会称为 MQ（Message Queue）, 也就是说音讯队列的实质就是一个队列，而队列是一种先进先出的数据结构，提供 消息传递和音讯排队 模型，能够在分布式环境下提供利用解耦、弹性伸缩、流量削峰、异步通信、数据同步、微服务之间通信等性能，作为分布式系统架构中的一个重要组件，有着无足轻重的位置
咱们会将要传输的数据、音讯放在音讯队列中
其中，往 MQ 里放货色的叫做生产者
从 MQ 外面取音讯的叫做消费者

为什么要用音讯队列

解耦

当初有一个零碎 A，A 能够产生一个 UserId
而后有零碎 B 和零碎 C 都须要这个 UserId 去做相干的操作

这样就会有一个问题，每当呈现改变的时候，我都要改变整个零碎，比方 B 不须要 A 的音讯了，那么就要改 A、B，新来的一个 D 服务，须要用到 A 的音讯，那么又须要批改，整个零碎的冗余度很高
引入音讯队列之后：零碎 A 将 id 写到音讯队列中，BC 服务从音讯队列中拿数据

零碎 A 只负责写数据，不关怀数据的生产、解决，而 BC 服务只负责拿数据，即便 BC 服务挂了，也和零碎 A 没关系，只和音讯队列无关，这样就能做到多个服务之间的解耦

异步

如果是同步的话，A 必须要期待 B C D 解决完之后能力返回，整个工夫比拟久

如果是异步的话，A（比方注册服务）将音讯写道音讯队列之后就能够返回，之后再发给邮件服务和短信服务生产

限流、削峰

当申请来的时候，先把申请放在音讯队列中，而后零碎再依据本人可能解决的申请数去音讯队列外面拿数据，这样即便每秒申请数很大，也不会把零碎打崩

流量管制

音讯队列通常有很多种形式来实现流量管制

1. 配额管制：通过为每个生产者或者消费者调配配额，限度它们能够发送或者承受的音讯数量。这能够保障音讯队列中的资源不会被适度应用，并确保零碎在高负载状况下的稳定性

2. 窗口机制：当生产者将音讯发送到音讯队列的时候，音讯队列会给每个生产和调配一个发送窗口，当消费者确认之后把对应的音讯从窗口外面删除

3. 缓冲区：音讯队列把音讯往缓冲区里丢，消费者从缓冲区里去取

4. 速率限度：这个个别是在客户端实现的，能够实现生产者在每秒、每分钟生产多少条音讯

如何保障音讯不被反复生产

常见的音讯队列都有确认机制（ACK 机制），当消费者生产数据之后会给音讯队列中间件发送一个确认音讯，音讯队列收到之后就会把这条音讯从队列中删除。

当呈现网络传输等故障，ACK 没有传送到音讯队列，导致音讯队列不晓得消费者曾经生产过该音讯了，再次将音讯分发给其余的消费者

解决：

1. 看场景，如果场景不须要幂等，那么能够不论，比方这条音讯拿去插入数据库，反复插入主键雷同的数据是会主动出错的，再比方做 redis 的 set 操作，也不须要管，屡次 set 之后依然是幂等的

2. 筹备第三方介质做生产记录，比方加个 redis，给音讯调配一个全局 id，只有生产过该音讯就将 <id,message> 写入 redis。消费者生产之前先去 redis 中查有没有生产记录即可

如何保障音讯的可靠性传输

音讯队列丢数据次要有三个可能

1. 生产者丢数据

2. 音讯队列组件丢数据

3. 消费者丢数据

生产者：能够采纳 transaction 机制，开启事务来发送音讯，如果发送失败就回滚。然而生产中用的不多，因为会导致吞吐量的降落，个别都是用 confirm 机制，如果生产者胜利把音讯发送给队列，队列会回一个 ack，否则回一个 nack

音讯队列：能够开启长久化，而且个别是集群部署的，有 master 和 slave 节点，个别都是同步复制，只有主节点和从节点都写入胜利才返回 ack 给生产者

消费者：勾销主动确认，主动确认后音讯队列收到 ACK 会立马把音讯从队列中删掉，而是手动确认（即解决后才回 ACK）。

音讯队列须要思考的问题

高可用

音讯队列必定不是单机的，这样可用性和健壮性都十分差，所以我的项目中应用的音讯队列都得集群或者分布式

数据问题

音讯失落：当消费者拿了数据还没应用的时候，服务就挂掉了，就会导致音讯的失落，个别会应用 ACK 应答机制，当消费者拿到音讯发送确认 ACK 信号，音讯队列才会把对应的音讯删掉

音讯沉积：音讯沉积分为客户端沉积和服务端沉积
个别都会设置告警规定来告诉开发者音讯沉积的问题

如果是客户端音讯沉积，那能够思考扩充生产线程或者节点来解决，针对于某些非凡场景，如果音讯沉积曾经影响到业务，并且沉积的音讯能够跳过不生产，那么能够重置生产音讯地位为最新地位开始生产，疾速复原业务。

如果是服务端音讯沉积，思考服务端宕机的状况，疾速复原之后从新可用

消费者取数据

两种办法

1.push 音讯队列有新音讯的时候被动叫消费者去拿，实时性强。如果消费者故障，服务端沉积音讯。

2.pull 消费者一直的轮询音讯队列，看看有没有新数据，如果有就生产，实时性弱。

音讯队列的传输模式

点对点模型

用于音讯生产者和音讯消费者之间点到点的通信，音讯生产者将音讯发送到某个特定的消费者
特点：
1. 每个音讯只有一个消费者
2. 发送者和接收者都没有工夫依赖
3. 接受者确认音讯承受和解决胜利

公布订阅模型

公布订阅模型反对向一个特定的音讯主题产生音讯，在这种模型下，公布和订阅者彼此不晓得对方的存在，多个消费者能够取得音讯，在发布者和订阅者之间存在 工夫依赖性 ， 发布者 publish须要建设一个 订阅 subscription，以便消费者可能订阅。订阅者必须 放弃继续的活动状态 并承受音讯

主题、订阅、消费者（组）之间的关系为 M：N：O

在这种状况下，订阅者未连贯时，公布的音讯将 在订阅者从新连贯的时候从新公布
特点：
1. 每个音讯能够有多个订阅者
2. 客户端只有订阅之后能力收到音讯
3. 长久订阅和非长久订阅

长久订阅：订阅关系建设之后，音讯就不会隐没，不论订阅者是否在线
非长久订阅：订阅者为了承受音讯，必须始终在线，当只有一个订阅者的时候等于点对点模式

pub-sub vs queue

公布订阅和队列模式是音讯队列中的两种不同的音讯模式

公布 - 订阅模式：发布者将音讯发送到特定的主题（topic）上，订阅者能够抉择订阅感兴趣的主题，从而承受与该主题相干的音讯。在该模式中，音讯被播送给所有订阅者，每个订阅者能够独立解决音讯，订阅者之间不会互相烦扰。公布 - 订阅模式通常用于播送音讯或者告诉

队列模式：音讯发送到队列中，而后一个或者多个消费者从队列中收取并解决音讯。在该模式中，每条音讯只能被一个消费者接管和解决。如果有多个消费者，音讯将被平均分配给它们。队列模式通常用于实现任务分配或者负载平衡等场景

市面上音讯队列比照

记住两个最常见的 MQ 的比照就能够，硬盘 MQ 代表是 Kafka 内存 MQ 代表是 RabbitMQ

Kafka 的长处是客户端反对多语言、应用 pull 模式，反对音讯批量操作，反对 replica 机制，Zookeeper 主动选举 leader 恢复能力，数据牢靠，有容错容灾的能力，单机吞吐量为 10W 级，提早毫秒，数据基于硬盘层面存储，多 Client 反对有序，不反对事务，然而能够通过 LOW LEVEL API 的形式保障音讯只反对生产一次

RabbitMQ 的长处是客户端反对多语言，多协定反对，不反对音讯批量操作，有 pull 和 push 两种模式，应用的主从模式 master/slave，master 提供服务，slave 做备份，数据牢靠（因为有备份），单机吞吐量为万级别，音讯提早为微秒级，内存级别，能够被动开启长久化，反对集群和负载平衡，不保障多 Client 音讯有序

pulsar kafka rabbitmq nsq 的异同

1.pulsar 和 kafka 基于公布订阅模式，而 rabbitmq 和 nsq 基于的是队列模式

2.pulsar 和 kafka 都采纳了长久化机制，以反对高吞吐量和高可靠性，而 rabbitmq 和 nsq 则将数据存储在内存中，以反对更低的提早和更高的吞吐量

3.pulsar 和 kafka 都反对 多租户和多数据中心部署，能够轻松地在多个数据中心或者云平台上进行扩大，而 rabbitmq 和 nsq 则更加适宜单个数据中心的部署

4.pulsar 和 kafka 都具备杰出的可伸缩性和高可用性，解决数据单位是百万级别的；而 rabbitmq 和 nsq 则更加适宜小规模的应用程序，具备更低的提早和更高的性能

5.pulsar 和 kafka 都提供了音讯流的处理程序，使用户能够对音讯进行实时剖析和解决

总的来说：pulsar 和 kafka 更适宜解决大量音讯和数据流，rabbitmq 和 nsq 则更适宜小规模应用程序，具备更低的提早和更高的性能

Pulsar 底层实现

Pulsar 是 Apache 基金会的顶级我的项目，是 云原生的分布式音讯队列，集音讯、存储、轻量化函数式计算为一体，采纳计算与存储拆散架构设计，反对多租户、长久化存储、多机房跨区域数据恢复机制，具备强一致性、高吞吐、低时延及高可扩展性等流数据存储个性

• 云原生 MQ
• 单个 Pulsar 实例原生反对多集群，可跨机房实现音讯复制
• 反对超过一百万个 topic
• 反对多语言客户端
• 主题多种订阅模式（独占、共享和故障转移）
• 通过 Book Keeper 来实现长久化存储，保障消息传递

• 分层式存储，可在数据古老时将数据从热存储卸载到冷存储中

  音讯

  Messages 有很多的局部组成，上面是几个值得注意的

• value/payload 音讯的数据
• properties 可选的属性，是一个 key->value 的键值对
• producer name 生产者名称
• publish time 公布工夫戳
• sequence id 在 topic 中 每个 msg 属于一个有序的队列 sequence id 是它在序列中的秩序

Pulsar 和其余的 MQ 一样，会对音讯的大小做出限度

这个限度通过 broker.conf 中的 maxMessageSize 决定

不设置的话，默认为 5MB

生产者

生产者是关联到 topic 的程序，它公布音讯到 Pulsar 的 broker 上

发送模式

• 异步发送：生产者发送音讯之后会期待 broker 的确认，如果没有收到确认则认为是发送失败
• 同步发送：会把音讯放在阻塞队列中，而后立马返回，而后这个阻塞队列会往 broker 中发消息

主题拜访模式

• Shared（共享）多个生产者能够公布一个主题，这是默认设置
• Exclusive（独占）一个主题只能由一个生产者公布，如果曾经有生产者链接，其余生产者试图公布该主题将立刻失去谬误。如果 ” 老 ” 的生产者与 broker 产生网络分区，” 老 ” 生产者将被驱赶，” 新 ” 生产者将被选为下一个惟一的生产者
• WaitForExclusive（独占期待）如果曾经有一个生产者连贯，生产者的创立是未决的，直到生产者取得独占拜访。胜利成为排他性的生产者被视为领导者，因而，如果想实现 leader 选举计划，能够应用这种模式

压缩、批处理与分块
压缩：咱们能够被动压缩生产者在传输期间公布的音讯，Pulsar 目前反对以下类型的压缩

• LZ4
• ZLIB
• ZSTD
• SNAPPY

批处理：如果批处理开启，producer 将会积攒一批音讯，而后通过一次申请发送进来。批处理的大小取决于最大的音讯数量及最大的提早公布

分块：分块和批处理不能同时启用，要启用分块，必须提前禁用批处理。Chunking 只反对长久化的主题

一个 producer 与一个订阅 consumer 的分块音讯
当生产者向主题发送一批大的分块音讯和一般的非分块音讯时。将 M1 切成分块 M1-C1、M1-C2、M1-C3。这个 broker 在其治理的 ledger 外面保留所有的三个块音讯，而后以雷同的程序分发给消费者（独占 / 灾备）。消费者将在内存缓存所有的块音讯，直到收到所有的音讯快。将这些音讯合并成为原始的音讯 M1，发送给解决过程

消费者

消费者通过订阅关联到主题，而后承受音讯的程序

接管模式

音讯能够通过同步或者异步的形式从 broker 承受

同步：同步接管将会阻塞，直到音讯可用

异步：异步接管立即返回 future 值，一旦新音讯可用，它将即可实现

监听

客户端类库提供了它们对于 consumer 的监听实现，在这个接口中，一旦承受到新的音讯，received 办法将被调用

确认

消费者胜利解决音讯之后须要发送确认（ack）给 broker，以让 broker 丢掉这条音讯（否则将始终存储）。音讯的确认能够一一进行，也能够累积到一起。累计确认的时候，消费者只须要确认最初一条它收到的音讯，所有之前的音讯都认为被胜利生产。累积确认不能用于 shared 模式，因为 shared 订阅为同一个订阅引入了多个消费者

主题

和其余的 MQ 一样，Pulsar 中的 topic 是带有名称的通道，用来从 producer 到 consumer 传输音讯，topic 的名称是合乎良好构造的 URL

{persistent|non-persistent}://tenant/namespace/topic

• peisistent|non-persistent 定义了 topic 的类型，Pulsar 反对两种不同的 topic：长久化和非长久化，默认是长久化类型，也就是会保留到硬盘上的类型
• tenant 实例中 topic 的租户，tenant 是 Pulsar 多租户的基本要素，能够被跨集群的流传
• namespace topic 的治理单元，与 topic 组的管理机制相干。大多数的 topic 配置在 namespace 层面失效，每个 tenant 能够有多个 namespace
• topic 主题的最初组成部分

Partitioned topics 分区主题

一般的主题只由单个 broker 提供服务，这限度了主题的最大吞吐量，分区主题是由多个 broker 解决的一种非凡类型的主题，因而容许更高的吞吐量
分区主题实际上实现为 N 个主题，N 是分区的数量。当音讯公布到分区主题的时候，每个音讯都被路由到几个 Broker 中的一个。分区在 broker 之间的散布由 Pulsar 主动的进行解决

如上图，Topic 有五个分区，划分在 3 个 broker 上，因为分区比 broker 多，前两个 broker 别离解决两个分区，而第三个 broker 只解决一个分区（Pulsar 主动解决分区的散布）
此主题的音讯将播送给两个消费者，路由模式决定将每个音讯公布到哪个分区，而订阅模式决定将哪些音讯发送到哪个消费者

路由模式

• RoundRobinPartition message 无 key 则轮询，有 key 则 hash 指定分区（默认模式）
• SinglePartition message 无 key 则 producer 将会随机抉择一个分区，将所有的音讯都发送给该分区。如果 message 有 key，那么会 hash 指定分区
• CustomParition 应用自定义音讯路由实现

程序保障

音讯的程序与路由模式和音讯的 key 无关

• Per-key-partition（按 key 分区）具备雷同 key 的所有音讯将被程序搁置在同一个分区中
• Per-producer（依照 Producer）来自同一生产者的所有音讯都是有序的

哈希计划

HashingScheme 是一个 enum，示意在抉择要为特定音讯应用的分区时可用的规范哈希函数集

有两种类型的规范哈希函数可用：JavaStringHash 和 Murmur3_32Hash，生产者的默认哈希函数是 Java，然而当生产者的客户端是多语言的时候，Java 是没用的

长久 / 非长久化主题

默认状况下,Pulsar 会保留所有没确认的音讯到 Book Keeper 中，长久 Topic 音讯会在 broker 重启或者 consumer 出问题的时候保留下来

Pulsar 也反对非长久化 Topic 这些 Topic 的音讯只存在于内存中，不会存储到磁盘

因为 Broker 不会对音讯进行长久化存储，当 Producer 将音讯发送给 Broker 时，Broker 能够立刻将 ack 返回给 Producer，所以非长久化的消息传递会比长久化的更快。绝对的，当 Broker 因为一些起因宕机、重启后，非长久化的 Topic 音讯都会隐没，订阅者将无奈收到这些音讯。

Dead letter topic 死信主题

死信主题容许你在用户无奈胜利生产某些音讯时应用新音讯。在这种机制中，无奈应用的音讯存储在独自的主题中，成为死信主题。

死信主题依赖于音讯的从新投递，因为确认超时或者否定确认，音讯将被从新发送。如果要对音讯应用否定确认，请确保在确认超时之前对齐进行否定确认。

Retry letter topic 重试主题

对于许多在线业务零碎，因为业务逻辑解决中出现异常，音讯会被反复生产。

若要配置从新生产失败音讯的延迟时间，能够配置生产者将音讯发送到业务主题和重试主题，并在消费者上启用主动重试。当在消费者上启用主动重试的时候，如果音讯没有被生产，那么就会存储到重试主题中，在指定的延迟时间后，消费者会被动承受来自重试主题的失败音讯

订阅模式

Pulsar 反对 exclusive（独占）failover（灾备）shared（共享）和 key_shared（Key 共享）四种音讯订阅模式，示意图如下

独占模式
默认的音讯订阅模式。只能有一个消费者生产音讯

灾备模式
灾备模式下，一个 topic 也是只有单个 consumer 生产一个订阅关系的音讯，然而在这个模式下，每个消费者会被排序，当后面的消费者无奈连贯上 broker 后，音讯会由下一个消费者进行生产

共享模式
共享模式下，音讯可被多个 consumer 同时生产，无奈保障生产的程序，音讯通过 roundrobin 的形式投递到每一个消费者

key 共享模式
依照 key 对音讯进行投递，雷同的 key 的音讯会被投递到同一个 consumer 上，生产示意图如下

音讯保留与过期
默认策略

• 立刻删除所有被消费者确认过的音讯
• 以 backlog 的模式，长久化保留所有未被确认的音讯

两个个性

• 音讯保留能够让你保留 consumer 确认过的音讯
• 音讯过期能够让你给未被确认的音讯设置 ttl

音讯保留和过期是针对 namesapce 层面进行设置和治理的

音讯去重

实现音讯去重的一种形式是确保音讯只生产一次，即生产者幂等，这种形式的毛病在于把音讯去重的工作交给利用来做。
在 pulsar 中，broker 反对配置开启音讯去重，用户不须要被动在代码中保障 Producer 只生产一次，启动之后即便音讯被屡次发送到 topic 上，也只会被长久化到磁盘一次

原理：Producer 对每一个发送的音讯，都会采纳递增的形式生成一个惟一的 sequence ID，这个音讯会放在 message 的元数据中传递给 broker。

同时，broker 也会保护一个 pendingmessage 队列，当 broker 返回发送胜利 ack 之后，producer 会将 pendingmessage 队列中的 sequence id 删除，标识 producer 工作这个音讯生产胜利。

broker 会记录针对每个 producer 承受到的最大 sequence id 和曾经解决完的最大 sequence id

当 broker 开启音讯去重之后，Broker 会针对每个音讯申请进行是否去重的判断，如果音讯反复，则间接返回 ack，不走后续存储的流程

延时音讯

延时音讯性能容许 Consumer 可能在音讯发送到 topic 之后，过一段时间之后能力生产到这条音讯。在这种集中中，音讯在公布到 broker 之后，会被存储在 book keeper 中，当到音讯特定的延迟时间时，音讯就会传递给 consumer

broker 不会在存储的时候做非凡解决，而是会把设置了延迟时间的音讯退出到 DelayedDeliveryTracker 中，当到了指定的发送工夫时，Tracker 才会把这条音讯推送给消费者

原理：

在 Pulsar 中有两种形式实现提早音讯，别离为 deliverAfter 和 deliverAt

deliverAfter 能够指定在多长时间之后进行生产

deliverAt 能够指定具体的提早生产工夫戳

DelayedDeliveryTracker 会记录所有须要提早投递的音讯的 index，index 由 timestamp、ledger id、和 entry id 三局部组成，其中 ledger id 和 entry id 用来定位该音讯

timestamp 除了记录须要投递的工夫，还用于提早优先级队列排序。tracker 会依据延迟时间对音讯进行排序

多租户模式

Pulsar 的云原生架构人造反对多租户，每个租户下还反对多 Namespace，非常适合做共享大集群，不便保护。此外 Pulsar 人造反对租户之间的逻辑隔离，避免相互烦扰，还能实现大集群资源的充分利用

• Tenant（租户）和 Namespace 是 Pulsar 反对多租户的两个外围概念
• 在租户级别，Pulsar 为特定的租户预留适合的存储空间、利用受权和认证机制
• 在 namespace 级别，Pulsar 有一系列的配置策略（policy），包含存储配额、流控、音讯过期策略等等

对立音讯模型

Pulsar 做了队列模型和流模型的对立，在 topic 级别只须要保留一份数据，同一份数据可屡次生产。以流式、队列等形式计算不同的订阅模型，大大的晋升了灵便度

同时 Pulsar 通过事务采纳 Exactly-Once 刚好一次的语义，在进行音讯传输过程中，能够确保音讯不重不丢

分片流

Pulsar 将无界的数据看作是分片的流，分片扩散存储在分层存储（tiered storage）、BookKeeper 集群和 Broker 节点上，而对外提供一个对立的、无界数据的视图

不须要用户显示迁徙数据，对用户无感知，缩小存储老本并放弃近似有限的存储

跨地区复制

Pulsar 中的跨地区复制是将 Pulsar 中长久化的音讯在多个集群之间备份
在 Pulsar2.4.0 中新增了复制订阅模式，在某个集群生效状况下，该性能能够在其余集群复原消费者的生产状态，从而达到热备模式下的音讯服务高可用

架构

单个 Pulsar 集群由以下三局部组成

• 一个或者多个 broker 用于负责解决和负载平衡 producer 收回的音讯 并将这些音讯分派给 consumer。Broker 和 Pulsar 配置存储交互来解决相应的工作，并将音讯存储在 BookKeeper 实例中（bookies）；Broker 底层依赖的是 Zookeeper 集群来解决特定的工作
• 蕴含一个或者多个 bookie 的 BookKeeper 负责音讯的长久化存储
• 一个 ZooKeeper 集群用来解决多个 Pulsar 集群之间的协调工作

Pulsar 拆散出 Broker 和 Bookie 两层架构，Broker 为无状态的服务，用于公布和生产音讯，而 BookKeeper 专一于存储，Pulsar 存储是分片的，这种架构能够防止扩容时受到限制，实现数据的独立拓展和疾速复原

Brokers
Pulsar 的 broker 是一个无状态的组件，次要负责运行另外的两个组件

• 一个 HTTP 服务器（service discovery）它裸露了 REST 系统管理接口以及在生产者和消费者之间进行 Topic 查找的 API
• 一个调度散发器（Dispatcher）它是一个异步的 TCP 服务器 通过自定义二进制协定利用与所有相干的数据传输

出于性能思考，音讯通常从 Managed Ledger 缓存中分派进来，除非积压超过缓存大小。如果积压的音讯对于缓存来说太大了，则 Broker 开始从 BookKeeper 中读取 Entries

为了反对全局 Topic 异地复制，Broker 会管制 Replicators 追踪本地公布的条目

ZooKeeper 元数据存储

Pulsar 应用 ZooKeeper 进行元数据存储、集群配置和协调

• 配置存储 Quorum 存储了租户、命名空间、和其余须要全局统一的配置项
• 每个集群有本人独立的本地 ZooKeeper 保留集群外部的配置，例如 broker 负责哪几个主题及所有权归属元数据、broker 负载报告 ledger 元数据等等

BookKeeper 长久化存储

Apache Pulsar 为应用程序提供有保障的信息传递，如果音讯胜利达到 broker，就认为其预期达到了目的地

为了提供这种保障，未确认送达的音讯须要长久化直到它们被确认送达。这种消息传递模式通常成为长久消息传递，在 Pulsar 外部，没分新音讯都被保留并同步 N 份

BookKeeper 是一个分布式的预写日志（WAL）零碎，有如下几个个性

• 使得 Pulsar 可能利用独立的日志，称为 ledgers，随着工夫的推移能够为 topic 创立多个 ledgers
• 保障多零碎挂掉时的 ledgers 的读取一致性
• 提供不同 Boookies 之间平均的 IO 散布的个性
• 容量和吞吐量都具备程度伸缩性，可能通过减少 bookies 立刻减少容量到集群中

本文由 mdnice 多平台公布