关于消息队列:MQ2消息队列

引言

由 MQ(1)音讯队列文章咱们晓得 Kafka 采纳公布 / 订阅队列，区别于 RabbitMQ 的队列模式，队列模型每条音讯只能被一个消费者生产，而公布 / 订阅模型就是为让一条音讯能够被多个消费者生产而生的，当然队列模型也能够通过音讯全量存储至多个队列来解决一条音讯被多个消费者生产问题，然而会有数据的冗余。
接下来的内容都基于公布 / 订阅模型 Kafka。

Kafka

个别咱们称发送音讯方为生产者 Producer，承受生产音讯方为消费者 Consumer，音讯队列服务端为 Broker。音讯从 Producer 发往 Broker，Broker 将音讯存储至本地，而后 Consumer 从 Broker 拉取音讯，或者 Broker 推送音讯至 Consumer，最初生产。

1. Broker
Kafka 集群蕴含一个或多个服务器，服务器节点称为 broker。broker 存储 topic 的数据。如果某 topic 有 N 个 partition，集群有 N 个 broker，那么每个 broker 存储该 topic 的一个 partition。刚好散布平均。
如果某 topic 有 N 个 partition，集群有 (N+M) 个 broker，那么其中有 N 个 broker 存储该 topic 的一个 partition，剩下的 M 个 broker 不存储该 topic 的 partition 数据。
如果某 topic 有 N 个 partition，集群中 broker 数目少于 N 个，那么一个 broker 存储该 topic 的一个或多个 partition。在理论生产环境中，尽量避免这种状况的产生，这种状况容易导致 Kafka 集群数据不平衡。
2. topic
每条公布到 Kafka 集群的音讯都有一个类别，这个类别被称为 Topic。
3. Partition(分区)
为了进步并发度，Kafka 引入了分区 Partition 的概念，在 RocketMQ 中也叫队列，实质一样。即音讯是发往一个主题下的某个分区中。例如某个主题下有 5 个分区，那么这个主题的并发度就进步为 5，同时能够有 5 个消费者并行生产该主题的音讯, 每个 topic 至多有一个 partition。。个别能够采纳轮询或者 key hash 取余等策略来将同一个主题的音讯调配到不同的队列中。每个 partition 中的数据应用多个 segment 文件存储。partition 中的数据是有序的，不同 partition 间的数据失落了数据的程序。如果 topic 有多个 partition，生产数据时就不能保证数据的程序。在须要严格保障音讯的生产程序的场景下，须要将 partition 数目设为 1。

3. Consumer Group
与之对应的消费者个别都有组的概念 Consumer Group, 即消费者都是属于某个生产组的。一条音讯会发往多个订阅了这个主题的生产组。假如当初有两个生产组别离是 Group 1 和 Group 2，它们都订阅了 Topic-a。此时有一条音讯发往 Topic-a，那么这两个生产组都能接管到这条音讯。
而后这条音讯理论是写入 Topic 某个分区中，生产组中的某个消费者对应生产一个分区的音讯。
在物理上除了正本拷贝之外，一条音讯在 Broker 中只会有一份，每个生产组会有本人的 offset 即生产点位来标识生产到的地位。在生产点位之前的音讯表明曾经生产过了。当然这个 offset 是队列级别的。每个生产组都会保护订阅的 Topic 下的每个队列的 offset。
5. Producer(生产者)
生产者即数据的发布者，该角色将音讯公布到 Kafka 的 topic 中。broker 接管到生产者发送的音讯后，broker 将该音讯追加到以后用于追加数据的 segment 文件中。生产者发送的音讯，存储到一个 partition 中，生产者也能够指定数据存储的 partition。
6. Leader and Follower
每个 partition 有多个正本，其中有且仅有一个作为 Leader，Leader 是以后负责数据的读写的 partition。Follower 追随 Leader，所有写申请都通过 Leader 路由，数据变更会播送给所有 Follower，Follower 与 Leader 保持数据同步。如果 Leader 生效，则从 Follower 中选举出一个新的 Leader。当 Follower 与 Leader 挂掉、卡住或者同步太慢，leader 会把这个 follower 从“in sync replicas”（ISR）列表中删除，从新创立一个 Follower。

1.Kafka 的分区策略有哪些？

所谓分区策略就是决定⽣产者将音讯发送到哪个分区的算法。

1. 轮询策略：默认的分区策略，⾮常优良的负载平衡体现，它总是能保障音讯最⼤限度地被平均分配到所有分区上；
2. 随机策略：实现随机策略版的 partition ⽅法；
3. 按音讯键保序策略：也称 Key-Ordering 策略，能够保障同⼀个 Key 的所有音讯都进⼊到雷同的分区⾥，因为每个分区下的音讯解决是有程序的，所以称之为音讯键保序策略；

2. 音讯队列中如何保证数据音讯不失落？

音讯失落是上游零碎没收到上游零碎发送的音讯，造成零碎间数据不统一。比方，订单零碎没有把胜利状态的订单音讯胜利发送到音讯队列里，造成上游的统计零碎没有收到下单胜利订单的音讯，于是造成零碎间数据的不统一，从而引起用户查看集体订单列表时跟理论不相符的问题。首先剖析音讯队列的流程，音讯失落的状况别离可能产生在生产端，Kafka 服务端，生产端。

（1）生产端须要保障不少生产音讯

• 应用带有回调办法的 API 时，咱们能够依据回调函数得悉音讯是否发送胜利，如果发送失败了，咱们要进行异样解决，比方把失败音讯存储到本地硬盘或近程数据库，等利用失常了再发送，这样能力保障音讯不失落。
• 设置参数 acks=-1。acks 这个参数是指有多少分区正本收到音讯后，生产者才认为音讯发送胜利了，可选的参数值有 0、1 和 -1。acks=0，示意生产者不期待任何服务器节点的响应，只有发送音讯就认为胜利。acks=1，示意生产者收到 leader 分区的响应就认为发送胜利。acks=-1，示意只有当 ISR 中的正本全副收到音讯时，生产者才会认为音讯生产胜利了。这种配置是最平安的，因为如果 leader 正本挂了，当 follower 正本被选为 leader 正本时，音讯也不会失落。然而零碎吞吐量会升高，因为生产者要期待所有正本都收到音讯后能力再次发送音讯。
• 第三个，设置参数 retries=3。参数 retries 示意生产者生产音讯的重试次数。这里 retries=3 是一个倡议值，个别状况下能满足足够的重试次数就能重试胜利。然而如果重试失败了，对异样解决时就能够把音讯保留到其余牢靠的中央，如磁盘、数据库、近程缓存等，而后等到服务失常了再持续发送音讯。
• 第四个，设置参数 retry.backoff.ms=300。retry.backoff.ms 指音讯生产超时或失败后重试的间隔时间，单位是毫秒。如果重试工夫太短，会呈现零碎还没复原就开始重试的状况，进而导致再次失败。联合我集体教训来说，300 毫秒还是比拟适合的。只有下面这四个要点配置对了，就能够保障生产端的生产者不少生产音讯了。

（2）服务端保障不丢音讯

• 第一个，设置 replication.factor >1。replication.factor 这个参数示意分区正本的个数，这里咱们要将其设置为大于 1 的数，这样当 leader 正本挂了，follower 正本还能被选为 leader 正本持续接管音讯。
• 第二个，设置 min.insync.replicas >1。min.insync.replicas 指的是 ISR 起码的正本数量，原理同上，也须要大于 1 的正本数量来保障音讯不失落。这里我简略介绍下 ISR。ISR 是一个分区正本的汇合，每个分区都有本人的一个 ISR 汇合。但不是所有的正本都会在这个汇合里，首先 leader 正本是在 ISR 汇合里的，如果一个 follower 正本的音讯没落后 leader 正本太长时间，这个 follower 正本也在 ISR 汇合里；可是如果有一个 follower 正本落后 leader 正本太长时间，就会从 ISR 汇合里被淘汰进来。也就是说，ISR 里的正本数量是小于或等于分区的正本数量的。

• 第三个，设置 unclean.leader.election.enable = false。unclean.leader.election.enable 指是否能把非 ISR 汇合中的正本选举为 leader 正本。unclean.leader.election.enable = true，也就是说容许非 ISR 汇合中的 follower 正本成为 leader 正本。如果设置成这样会有什么问题呢？假如 ISR 汇合内的 follower1 正本和 ISR 汇合外的 follower2 正本向 leader 正本拉取音讯（如下图 1），也就是说这时 ISR 汇合中就有两个正本，一个是 leader 正本，另一个是 follower1 正本，而 follower2 正本因为网络或本身机器的起因曾经落后 leader 正本很长时间，曾经被踢出 ISR 汇合。
  
  忽然 leader 和 follower1 这两个正本挂了，因为 unclean.leader.election.enable = true，而当初分区的副本能失常工作的仅仅剩下 follower2 正本，所以 follower2 最终会被选为新的 leader 正本并持续接管生产者发送的音讯，咱们能够看到它接管了一个新的音讯 5。
  
  如果这时 follower1 正本的服务复原，又会产生什么状况呢？因为 follower 正本要拉取 leader 正本同步数据，首先要获取 leader 正本的信息，并感知到当初的 leader 正本的 LEO 比本人的还小，于是做了截断操作，这时 4 这个音讯就丢了，这就造成了音讯的失落。
  
  因而，咱们肯定要把 unclean.leader.election.enable 设置为 false，只有这样非 ISR 汇合的正本才不会被选为分区的 leader 正本。然而这样做也升高了可用性，因为这个分区的正本没有 leader，就无奈收发音讯了，然而音讯会发送到别的分区 leader 正本，也就是说分区的数量实际上缩小了。

  （3）生产端不能少生产音讯

• 消费者生产音讯是有两个步骤的，首先拉取音讯，而后再解决音讯。向服务端提交音讯偏移量能够手动提交也能够主动提交。如果把参数 enable.auto.commit 设置为 true 就示意音讯偏移量是由生产端主动提交，由异步线程去实现的，业务线程无法控制。如果刚拉取了音讯之后，业务解决还没进行完，这时提交了音讯偏移量然而消费者却挂了，这就造成还没进行完业务解决的音讯的位移被提交了，下次再生产就生产不到这些音讯，造成音讯的失落。因而, 只有敞开主动提交 offset，在解决完生产的业务逻辑后手动提交 offset, 就能够保证数据不失落。

3. 如何保障音讯不被反复生产？或者说，如何保障音讯生产的幂等性？

要保障音讯不被反复复生产，其实就是要保障音讯生产时的幂等性。幂等性：⽆论你反复申请多少次，失去的后果都是⼀样的。例如：⼀条数据反复呈现两次，数据库⾥就只有⼀条数据，这就保障了零碎的幂等性。那么如何保障幂等性呢？

• 生产端写数据时，先依据主键查⼀下这条数据是否存在，如果曾经存在则 update；
• 数据库的唯⼀键束缚也能够保障不会反复插⼊多条，因为反复插⼊多条只会报错，不会导致数据库中呈现脏数据；
• 生产端 v 如果是写 redis，就没有问题，因为 set 操作是人造幂等性的。

4. 如何保障音讯的程序性？

• ⼀个 Topic，⼀个 Partition，⼀个 Consumer，外部单线程生产，单线程吞吐量太低，⼀般不会⽤这个。
• 写 N 个内存 Queue，具备雷同 key 的数据都到同⼀个内存 Queue；而后对于 N 个线程，每个线程别离生产⼀个内存 Queue 即可，这样就能保障程序性。