一文读懂消息队列的一些设计

ゝ一纸荒年。 2022-01-14 06:15 257阅读 0赞

[2019独角兽企业重金招聘Python工程师标准>>> ][2019_Python_] ![hot3.png][]

## 高可用 ##

常用的消息队列的高可用是怎么设计的呢？

消息队列一般都有一个nameserver服务，用来检测broker是否存活，或者处理能力上是否存在延迟。这样在发送消息时就可以规避将消息发送到宕机的broker上，也避免因为网络等原因消息处理失败。

那么针对于以上两种情况，消息队列如何保证高可用方案的呢？

## 多副本 ##

每个topic可以设置几个partition，每个partition负责存储一部分数据。kafka的broker集群中，每台机器存储一些partition，存放一部分topic数据，这就实现了topic数据分布在一个broker集群上。

任何一个分布式系统，内部都有一套多副本冗余机制，多副本冗余是任何一个分布式系统具备的基本能力。

kafka中每个partition都有多个副本，其中一个副本是leader，其他副本为follower，leader和follower分布在不同机器上。leader对外统一提供写服务，leader接收到消息后follower副本会不停的和leader通信，尝试拉去最新数据，并持久化到本地磁盘。

说到这里不得不说下ISR，也就是保持同步的副本，表示了和leader始终保持同步的follower有哪些。

比如follower由于fullgc造成自己卡顿，使得无法及时从leader拉取数据，会导致这个follower数据比leader落后很多。 只要follower一直和leader保持同步关系，他们就处于同步关系。

每个partition都有一个ISR，这个ISR一定有leader，因为leader的数据永远是最新的，然后就是和leader保持同步的follower，也会在ISR里面。

kafka中有个acks参数。是在producer里面设置的，也就是客户端设置的。

在向fafka集群写数据时，可以设置这个acks参数，这个参数值有：0,1,all。

0： 意思是proucer在客户端只要把消息发送出去，不管消息有没有在partition leader上落盘就不管了。就认为消息发送成功了。

1： 意思是producer生产的消息要确保partition leader写入本地磁盘，就认为成功了，而不管follower有没有同步这条消息。 当然这个是kafka的默认设置。

all： 意思是partition leader接收到消息后，持久化到本地，还要求ISR列表中跟leader保持同步的那些follower要把消息持久了，才算写入成功。一般要求acks=all时，必须isr列表里面有两个以上的副本配合使用，起码每个leader有一个follower才行。

当broker回复客户端消息没有写入成功时，需要客户端进行消息重发。

## 重试 ##

消息发送时，一般存在这样的方法：

for(; times < timesTotal; times++){
    // send message
    }

这里是client发送消息时决定的重试次数，默认值为3。重试可以提高消息发送的成功率。

## 消息发送 ##

默认的消息发送采用对消息队列进行取模，确定队列。 其他的方式比如轮训方式等。

Kafka 有两个默认的分配策略：

*  Range：该策略会把主题的若干个连续的分区分配给消费者。
 *  RoundRobin：该策略把主题的所有分区逐个分配给消费者。

## 消费者 ##

消费者向kafka订阅topic，并从topic上接收消息。

消费者属于消费者组，一个消费组的消费组订阅的是同一个topic，每个消费者接收topic一个partition的消息。

kafka默认的规则中，每个分区只能被同一个消费组里面的一个消费者消费。

1个消费者接收4个分区的消息： ![80e530f223261e81914bd412b7d0c50f384.jpg][]

2个消费者接收4个分区的消息： ![ee1da089fb215eefe20995b087b114b45b4.jpg][]

4个消费者接收4个分区的消息： ![1ed6e9f495617dfffe9cfb22376c1971aa6.jpg][]

5个消费者接收4个分区的消息： ![25dc30616975de9de888f69ab9a3680ff26.jpg][]

如果消费者群组的消费者超过主题的分区数量，那么有一部分消费者就会被闲置，不会接收到任何消息。

两个消费者群组对应一个主题： ![f1646f94a545d88c1b1ade98ad9b79fb1a8.jpg][]

当一个消费者被关闭或发生崩溃时，它就离开群组，原本由它读取的分区将由群组里的其他消费者来读取。分区的所有权从一个消费者转移到另一个消费者，这样的行为被称为再均衡。在再均衡期间，消费者无法读取消息，造成整个群组一小段时间的不可用。

通过上面消费者实例数量变化思考一个问题。在消费者机器重启过程中，存在partition和消费者重新建立联系的情况，比如最开始有4个消费者，由于并行重启消费者，可能存在一段时间消费者数量变为2个，当重启完成后消费者数量有变成了4个。

这个过程存在消息可能重复发送到同一个消费者消费的情况，造成重复消费，如果是对消息重复敏感的应用场景，我司自研的消息队列组件会提供一个选项，消息在分区进行主动积压，默认积压30s等待消费者重启完成，达到稳定的消费者数量。

消费者通过向被指派为群组协调器的 broker 发送心跳来维持它们和群组的从属关系以及它们对分区的所有权关系。消费者会在轮训消息或提交偏移量时发送心跳。如果消费者停止发送心跳的时间足够长，会话就会过期，群组协调器认为它已经死亡，就会触发一次再均衡。

如果一个消费者发生崩溃，并停止读取消息，群组协调器会等待几秒钟，确认它死亡了才会触发再均衡。所以上面的延迟是由于再平衡期间不可用造成的。

当消费者要加入群组时，它会向群组协调器发送一个 JoinGroup 请求。

第一个加入群组的消费者将成为"群主"。群主从协调器那里获得群组的成员列表，并负责给每一个消费者分配分区。 分配完毕之后，群主把分配情况列表发送给群组协调器，协调器再把这些信息发送给所有消费者。 每个消费者只能看到自己的分配情况。这个过程会在每次再均衡时重复发生。

## 消息消费 ##

kafka消费者有自己消费偏移量，这个偏移量是从kafka中读取的量，和kafka提交的偏移量不一样。消费者一般需要第一次和rebalance的时候需要根据提交的偏移量来获取数据，剩下的时候根据自己本地的偏移量来获取。

当消费者使用了自动提交模式，当还没有提交的时候，有消费者加入或者移除，发送rebalance，再次消费时，消费者根据提交偏移量进行，可能产生重复消费数据。

## 选举设计 ##

先说分区leader的选举，就是当ISR中的leader副本挂了，再重新选举一个过程。

kafka中的选举大致可以分为三大类：

*  控制器选举
 *  分区leader选举
 *  消费组相关选举

控制器选举： kafka集群中有一个或多个broker，其中一个broker会被选举为kafka controller，负责管理整个集群中所有分区和副本状态。当检测到某个分区的leader副本出现故障，controller负责为该分区选举新的leader副本。 如果检测到某个分区ISR集合发生变化时，控制器负责通知所有的broker更新元数据信息。 kafka controller的实现是依赖于zk实现的，哪个broker成功在zk的/controller临时节点创建成功，就成为kafka controller。

分区leader选举： 在topic下增加分区或者分区下线时，都需要执行leader选举。 基本思路是按照AR集合中副本顺序查找第一个存活的副本，并且这个副本在ISR集合中。

消费者相关选举： 消费组协调器需要为消费组内的消费者选择一个消费组leader，这个选举算法比较简单。 如果消费组内没有leader，那么第一个加入消费组的消费者成为组leader。 如果由于某种原因leader消费者退出消费组，需要重新选举leader，消费者协调器维护一个map结构，key为消费组id，value为消费者元信息，默认选择第一个key作为leader。

转载于:https://my.oschina.net/u/1000241/blog/3044851

[2019_Python_]: https://my.oschina.net/u/2663968/blog/3061697
[hot3.png]: /images/20220114/82cc3e318f5c43a1bef608016a02d137.png
[80e530f223261e81914bd412b7d0c50f384.jpg]: /images/20220114/f79d05b37e7140aaac69f8325545bbfe.png
[ee1da089fb215eefe20995b087b114b45b4.jpg]: /images/20220114/a2e664436eb54d09b55bdf72a87119e5.png
[1ed6e9f495617dfffe9cfb22376c1971aa6.jpg]: /images/20220114/dfbfe7a721614eebb89c917cc2a52ced.png
[25dc30616975de9de888f69ab9a3680ff26.jpg]: /images/20220114/d78a0e186e4b4f7db18d789df6f13a56.png
[f1646f94a545d88c1b1ade98ad9b79fb1a8.jpg]: /images/20220114/475140ddfb1d4d4d8f5225b966980179.png