知识分享-江南app体育官方入口

什么场景用到了消息队列

说到消息队列你脑子就要想到异步、削峰、解耦，条件反射那种。

异步：

• 我们之前的场景里面有很多步骤都是在一个流程里面需要做完的，就比如说我的下单系统吧，本来我们业务简单，下单了付了钱就好了，流程就走完了。
• 但是后面来了个产品经理，搞了个优惠券系统，ok问题不大，流程里面多100ms去扣减优惠券。
• 后来产品经理灵光一闪说我们可以搞个积分系统啊，也行吧，流程里面多了200ms去增减积分。
• 再后来后来隔壁的产品老王说：下单成功后我们要给用户发短信，也将就吧，100ms去发个短信。
• 你们可以看到这才加了三个，我可以斩钉截铁的告诉你真正的下单流程涉及的系统绝对在10个以上（主流电商），越大的越多。
• 这个链路这样下去，时间长得一批，用户发现我买个东西你特么要花几十秒，垃圾电商我不在你这里买了，不过要是都像并夕夕这么便宜，真香！
• 但是我们公司没有夕夕的那个经济实力啊，那只能优化系统了。
• 那链路长了就慢了，但是我们发现上面的流程其实可以同时做的呀，你支付成功后，我去校验优惠券的同时我可以去增减积分啊，还可以同时发个短信啊。
• 那正常的流程我们是没办法实现的呀，怎么办，异步。
• 你对比一下是不是发现，这样子最多只用100毫秒用户知道下单成功了，至于短信你迟几秒发给他他根本不在意是吧。

解耦：

• 为啥我们不能用线程去做，因为用线程去做，你是不是要写代码？
• 你一个订单流程，你扣积分，扣优惠券，发短信，扣库存。。。等等这么多业务要调用这么多的接口，每次加一个你要调用一个接口然后还要重新发布系统，写一次两次还好，写多了你就说：老子不干了！
• 而且真的全部都写在一起的话，不单单是耦合这一个问题，你出问题排查也麻烦，流程里面随便一个地方出问题搞不好会影响到其他的点，小伙伴说我每个流程都try catch不就行了，相信我别这么做，这样的代码就像个定时炸弹，你不知道什么时候爆炸，平时不炸偏偏在你做活动的时候炸，你就领个p0故障收拾书包提前回家过年吧。
• 但是你用了消息队列，耦合这个问题就迎刃而解了呀。
• 你下单了，你就把你 支付成功的消息告诉别的系统 ，他们收到了去处理就好了，你只用走完自己的流程，把自己的消息发出去，那后面要接入什么系统简单，直接订阅你发送的支付成功消息，你支付成功了我 监听就好了 。

削峰：

平时流量很低，但是你要做秒杀活动00 ：00的时候流量疯狂怼进来，你的服务器，redis，mysql各自的承受能力都不一样，你直接全部流量照单全收肯定有问题啊，直接就打挂了。

• 简单，把请求放到队列里面，然后至于每秒消费多少请求，就看自己的服务器处理能力，你能处理5000qps你就消费这么多，可能会比正常的慢一点，但是不至于打挂服务器，等流量高峰下去了，你的服务也就没压力了。
• 你看阿里双十一12：00的时候这么多流量瞬间涌进去，他有时候是不是会慢一点，但是人家没挂啊，或者降级给你个友好的提示页面，等高峰过去了又是一条好汉了。

问题

系统复杂性

本来蛮简单的一个系统，我代码随便写都没事，现在你凭空接入一个中间件在那，我是不是要考虑去维护他，而且使用的过程中是不是要考虑各种问题，比如消息重复消费、消息丢失、消息的顺序消费等等，反正用了之后就是贼烦。

数据一致性

• 这个其实是分布式服务本身就存在的一个问题，不仅仅是消息队列的问题，但是放在这里说是因为用了消息队列这个问题会暴露得比较严重一点。
• 你下单的服务自己保证自己的逻辑成功处理了，你成功发了消息，但是优惠券系统，积分系统等等这么多系统，他们成功还是失败你就不管了？
• 保证自己的业务数据对的就好了，其实还是比较不负责任的一种说法，这样就像个渣男，没有格局，这样呀你的路会越走越窄的。

重复消费

一般消息队列的使用，我们都是有重试机制的，就是说我下游的业务发生异常了，我会抛出异常并且要求你重新发一次。

我这个活动这里发生错误，你要求重发肯定没问题。但是大家仔细想一下问题在哪里？

是的，不止你一个人监听这个消息啊，还有别的服务也在监听，他们也会失败啊，他一失败他也要求重发，但是你这里其实是成功的，重发了，你的钱不就加了两次了？

真实的情况其实重试是很正常的，服务的网络抖动，开发人员代码bug，还有数据问题等都可能处理失败要求重发的。

一般我们叫这样的处理叫接口幂等。

幂等（idempotent、idempotence）是一个数学与计算机学概念，常见于抽象代数中。

在编程中一个幂等操作的特点是其任意多次执行所产生的影响均与一次执行的影响相同。

幂等函数，或幂等方法，是指可以使用相同参数重复执行，并能获得相同结果的函数。这些函数不会影响系统状态，也不用担心重复执行会对系统造成改变。

例如，“settrue()”函数就是一个幂等函数,无论多次执行，其结果都是一样的.更复杂的操作幂等保证是利用唯一交易号(流水号)实现.

一般幂等，可以分场景去考虑，看是强校验还是弱校验，比如跟金钱相关的场景那就很关键呀，就做强校验，别不是很重要的场景做弱校验。

所有的服务都成功才能算这一次下单是成功的，那怎么才能保证数据一致性呢？

分布式事务：把下单，优惠券，积分。。。都放在一个事务里面一样，要成功一起成功，要失败一起失败。

可用性

你搞个系统本身没啥问题，你现在突然接入一个中间件在那放着，万一挂了怎么办？我下个单mq挂了，优惠券不扣了，积分不减了，这不是杀一个程序员能搞定的吧，感觉得杀一片。

技术选型

市面上比较主流的消息队列中间件主要有，kafka、activemq、rabbitmq、rocketmq 等这几种

activemq和rabbitmq这两着因为吞吐量还有github的社区活跃度的原因，在各大互联网公司都已经基本上绝迹了，业务体量一般的公司会是有在用的，但是越来越多的公司更青睐rocketmq和kafka这样的消息中间件了。

kafka和rocketmq一直在各自擅长的领域发光发亮，不过写这篇文章的时候我问了蚂蚁金服，字节跳动和美团的朋友，好像大家用的都有点不一样，应该都是各自的中间件，可能做过修改，也可能是自研的，大多没有开源。

就像我们公司就是是基于kafka和rocketmq两者的优点自研的消息队列中间件，吞吐量、可靠性、时效性等都很可观。

就拿吞吐量来说，早期比较活跃的activemq 和rabbitmq基本上不是后两者的对手了，在现在这样大数据的年代吞吐量是真的很重要。

比如现在突然爆发了一个超级热点新闻，你的app注册用户高达亿数，你要想办法第一时间把突发全部推送到每个人手上，你没有大吞吐量的消息队列中间件用啥去推？

再说这些用户大量涌进来看了你的新闻产生了一系列的附带流量，你怎么应对这些数据，很多场景离开消息队列基本上难以为继。

就部署方式而言前两者也是大不如后面两个天然分布式架构的哥哥，都是高可用的分布式架构，而且数据多个副本的数据也能做到0丢失。

rabbitmq这个中间件其实还行，但是这玩意开发语言居然是erlang，我敢说绝大部分工程师肯定不会为了一个中间件去刻意学习一门语言的，开发维护成本你想都想不到，出个问题查都查半天。

至于rocketmq（阿里开源的），git活跃度还可以。基本上你push了自己的bug确认了有问题都有阿里大佬跟你试试解答并修复的，我个人推荐的也是这个，他的架构设计部分跟同样是阿里开源的一个rpc框架是真的很像（dubbo）可能是因为师出同门的原因吧。

kafka我放到最后说，你们也应该知道了，压轴的这是个大哥，大数据领域，公司的日志采集，实时计算等场景，都离不开他的身影，他基本上算得上是世界范围级别的消息队列标杆了。

分享的问题

• kafka节点之间如何复制备份的？
• kafka消息是否会丢失？为什么？
• kafka最合理的配置是什么？
• kafka的leader选举机制是什么？
• kafka对硬件的配置有什么要求？
• kafka的消息保证有几种方式？

kafka硬件配置推荐

kafka对cpu和hd要求不高，因为他不是密集型计算框架，他只是一个消息队列，吞吐量才是真谛。

但他对内存要求很高，一般情况下jvm设置几个g就可以，但是页缓存会用的更多，内存大页缓存才大，吞吐量才能上去。

如何确定kafka的分区数、key和consumer线程数?

1. 怎么确定分区数？
“我应该选择几个分区？”——如果你在kafka中国社区的群里，这样的问题你会经常碰到的。不过有些遗憾的是，我们似乎并没有很权威的答案能够解答这样的问题。其实这也不奇怪，毕竟这样的问题通常都是没有固定答案的。kafka江南app体育官方入口官网上标榜自己是”high-throughput distributed messaging system”，即一个高吞吐量的分布式消息引擎。那么怎么达到高吞吐量呢？kafka在底层摒弃了java堆缓存机制，采用了操作系统级别的页缓存，同时将随机写操作改为顺序写，再结合zero-copy的特性极大地改善了io性能。但是，这只是一个方面，毕竟单机优化的能力是有上限的。如何通过水平扩展甚至是线性扩展来进一步提升吞吐量呢？ kafka就是使用了分区(partition)，通过将topic的消息打散到多个分区并分布保存在不同的broker上实现了消息处理(不管是producer还是consumer)的高吞吐量。
kafka的生产者和消费者都可以多线程地并行操作，而每个线程处理的是一个分区的数据。因此分区实际上是调优kafka并行度的最小单元。对于producer而言，它实际上是用多个线程并发地向不同分区所在的broker发起socket连接同时给这些分区发送消息；而consumer呢，同一个消费组内的所有consumer线程都被指定topic的某一个分区进行消费(具体如何确定consumer线程数目我们后面会详细说明)。所以说，如果一个topic分区越多，理论上整个集群所能达到的吞吐量就越大。

• 客户端/服务器端需要使用的内存就越多

  • 先说说客户端的情况。kafka 0.8.2之后推出了java版的全新的producer，这个producer有个参数batch.size，默认是16kb。它会为每个分区缓存消息，一旦满了就打包将消息批量发出。看上去这是个能够提升性能的设计。不过很显然，因为这个参数是分区级别的，如果分区数越多，这部分缓存所需的内存占用也会更多。假设你有10000个分区，按照默认设置，这部分缓存需要占用约157mb的内存。而consumer端呢？我们抛开获取数据所需的内存不说，只说线程的开销。如果还是假设有10000个分区，同时consumer线程数要匹配分区数(大部分情况下是最佳的消费吞吐量配置)的话，那么在consumer client就要创建10000个线程，也需要创建大约10000个socket去获取分区数据。这里面的线程切换的开销本身已经不容小觑了。
    服务器端的开销也不小，如果阅读kafka源码的话可以发现，服务器端的很多组件都在内存中维护了分区级别的缓存，比如controller，fetchermanager等，因此分区数越多，这种缓存的成本越久越大。

• 文件句柄的开销

  • 每个分区在底层文件系统都有属于自己的一个目录。该目录下通常会有两个文件： base_offset.log和base_offset.index。kafak的controller和replicamanager会为每个broker都保存这两个文件句柄(file handler)。很明显，如果分区数越多，所需要保持打开状态的文件句柄数也就越多，最终可能会突破你的ulimit -n的限制。

• 降低高可用性

  • kafka通过副本(replica)机制来保证高可用。具体做法就是为每个分区保存若干个副本(replica_factor指定副本数)。每个副本保存在不同的broker上。期中的一个副本充当leader 副本，负责处理producer和consumer请求。其他副本充当follower角色，由kafka controller负责保证与leader的同步。如果leader所在的broker挂掉了，contorller会检测到然后在zookeeper的帮助下重选出新的leader——这中间会有短暂的不可用时间窗口，虽然大部分情况下可能只是几毫秒级别。但如果你有10000个分区，10个broker，也就是说平均每个broker上有1000个分区。此时这个broker挂掉了，那么zookeeper和controller需要立即对这1000个分区进行leader选举。比起很少的分区leader选举而言，这必然要花更长的时间，并且通常不是线性累加的。如果这个broker还同时是controller情况就更糟了。

但分区是否越多越好呢？显然也不是，因为每个分区都有自己的开销：

怎么确定分区数

• 视情况而定。基本上你还是需要通过一系列实验和测试来确定。当然测试的依据应该是吞吐量。虽然linkedin这篇文章做了kafka的基准测试，但它的结果其实对你意义不大，因为不同的硬件、软件、负载情况测试出来的结果必然不一样。我经常碰到的问题类似于，江南app体育官方入口官网说每秒能到10mb，为什么我的producer每秒才1mb？ —— 且不说硬件条件，最后发现他使用的消息体有1kb，而江南app体育官方入口官网的基准测试是用100b测出来的，因此根本没有可比性。不过你依然可以遵循一定的步骤来尝试确定分区数：创建一个只有1个分区的topic，然后测试这个topic的producer吞吐量和consumer吞吐量。假设它们的值分别是tp和tc，单位可以是mb/s。然后假设总的目标吞吐量是tt，那么分区数 = tt / max(tp, tc)
• tp表示producer的吞吐量。测试producer通常是很容易的，因为它的逻辑非常简单，就是直接发送消息到kafka就好了。tc表示consumer的吞吐量。测试tc通常与应用的关系更大， 因为tc的值取决于你拿到消息之后执行什么操作，因此tc的测试通常也要麻烦一些。
• 另外，kafka并不能真正地做到线性扩展(其实任何系统都不能)，所以你在规划你的分区数的时候最好多规划一下，这样未来扩展时候也更加方便。

消息-分区的分配
默认情况下，kafka根据传递消息的key来进行分区的分配，即hash(key) % numpartitions

def partition(key: any, numpartitions: int): int = {
       utils.abs(key.hashcode) % numpartitions
}

这就保证了相同key的消息一定会被路由到相同的分区。如果你没有指定key，那么kafka是如何确定这条消息去往哪个分区的呢？

if(key == null) { // 如果没有指定key
    val id = sendpartitionpertopiccache.get(topic) // 先看看kafka有没有缓存的现成的分区id
    id match {
    case some(partitionid) =>
        partitionid // 如果有的话直接使用这个分区id就好了
    case none => // 如果没有的话，
        val availablepartitions = topicpartitionlist.filter(_.leaderbrokeridopt.isdefined) //找出所有可用分区的leader所在的broker
    if (availablepartitions.isempty)
        throw new leadernotavailableexception("no leader for any partition in topic "  topic)
    val index = utils.abs(random.nextint) % availablepartitions.size // 从中随机挑一个   
    val partitionid = availablepartitions(index).partitionid
    sendpartitionpertopiccache.put(topic, partitionid) // 更新缓存以备下一次直接使用
    partitionid
   }
}

可以看出，kafka几乎就是随机找一个分区发送无key的消息，然后把这个分区号加入到缓存中以备后面直接使用——当然了，kafka本身也会清空该缓存（默认每10分钟或每次请求topic元数据时）

如何设定consumer线程数

我个人的观点，如果你的分区数是n，那么最好线程数也保持为n，这样通常能够达到最大的吞吐量。超过n的配置只是浪费系统资源，因为多出的线程不会被分配到任何分区。让我们来看看具体kafka是如何分配的。

• topic下的一个分区只能被同一个consumer group下的一个consumer线程来消费，但反之并不成立，即一个consumer线程可以消费多个分区的数据，比如kafka提供的consoleconsumer，默认就只是一个线程来消费所有分区的数据。——其实consoleconsumer可以使用通配符的功能实现同时消费多个topic数据，但这和本文无关。
• 再讨论分配策略之前，先说说kafkastream——它是consumer的关键类，提供了遍历方法用于consumer程序调用实现数据的消费。其底层维护了一个阻塞队列，所以在没有新消息到来时，consumer是处于阻塞状态的，表现出来的状态就是consumer程序一直在等待新消息的到来。——你当然可以配置成带超时的consumer，具体参看参数的用法。

下面说说kafka提供的两种分配策略： range和roundrobin，由参数partition.assignment.strategy指定，默认是range策略。本文只讨论range策略。所谓的range其实就是按照阶段平均分配。
举个例子就明白了，假设你有10个分区，p0 ~ p9，consumer线程数是3， c0 ~ c2，那么每个线程都分配哪些分区呢？

c0 消费分区 0, 1, 2, 3
c1 消费分区 4, 5, 6
c2 消费分区 7, 8, 9

val npartsperconsumer = curpartitions.size / curconsumers.size // 每个consumer至少保证消费的分区数
val nconsumerswithextrapart = curpartitions.size % curconsumers.size // 还剩下多少个分区需要单独分配给开头的线程们
...
for (consumerthreadid <- consumerthreadidset) { // 对于每一个consumer线程
val myconsumerposition = curconsumers.indexof(consumerthreadid) //算出该线程在所有线程中的位置，介于[0, n-1]
assert(myconsumerposition >= 0)
// startpart 就是这个线程要消费的起始分区数
val startpart = npartsperconsumer * myconsumerposition  myconsumerposition.min(nconsumerswithextrapart)
// nparts 就是这个线程总共要消费多少个分区
val nparts = npartsperconsumer  (if (myconsumerposition  1 > nconsumerswithextrapart) 0 else 1)
...
}

针对于这个例子，npartsperconsumer就是10/3=3，nconsumerswithextrapart为10%3=1，说明每个线程至少保证3个分区，还剩下1个分区需要单独分配给开头的若干个线程。这就是为什么c0消费4个分区，后面的2个线程每个消费3个分区

leader选举

控制器（broker）选举

所谓控制器就是一个borker，在一个kafka集群中，有多个broker节点，但是它们之间需要选举出一个leader，其他的broker充当follower角色。集群中第一个启动的broker会通过在zookeeper中创建临时节点/controller来让自己成为控制器，其他broker启动时也会在zookeeper中创建临时节点，但是发现节点已经存在，所以它们会收到一个异常，意识到控制器已经存在，那么就会在zookeeper中创建watch对象，便于它们收到控制器变更的通知。

那么如果控制器由于网络原因与zookeeper断开连接或者异常退出，那么其他broker通过watch收到控制器变更的通知，就会去尝试创建临时节点/controller，如果有一个broker创建成功，那么其他broker就会收到创建异常通知，也就意味着集群中已经有了控制器，其他broker只需创建watch对象即可。

如果集群中有一个broker发生异常退出了，那么控制器就会检查这个broker是否有分区的副本leader，如果有那么这个分区就需要一个新的leader，此时控制器就会去遍历其他副本，决定哪一个成为新的leader，同时更新分区的isr集合。

如果有一个broker加入集群中，那么控制器就会通过broker id去判断新加入的broker中是否含有现有分区的副本，如果有，就会从分区副本中去同步数据。

集群中每选举一次控制器，就会通过zookeeper创建一个controller epoch，每一个选举都会创建一个更大，包含最新信息的epoch，如果有broker收到比这个epoch旧的数据，就会忽略它们，kafka也通过这个epoch来防止集群产生“脑裂”。

分区副本选举机制

在kafka的集群中，会存在着多个主题topic，在每一个topic中，又被划分为多个partition，为了防止数据不丢失，每一个partition又有多个副本，在整个集群中，总共有三种副本角色：

首领副本（leader）：也就是leader主副本，每个分区都有一个首领副本，为了保证数据一致性，所有的生产者与消费者的请求都会经过该副本来处理。
跟随者副本（follower）：除了首领副本外的其他所有副本都是跟随者副本，跟随者副本不处理来自客户端的任何请求，只负责从首领副本同步数据，保证与首领保持一致。如果首领副本发生崩溃，就会从这其中选举出一个leader。
首选首领副本：创建分区时指定的首选首领。如果不指定，则为分区的第一个副本。
follower需要从leader中同步数据，但是由于网络或者其他原因，导致数据阻塞，出现不一致的情况，为了避免这种情况，follower会向leader发送请求信息，这些请求信息中包含了follower需要数据的偏移量offset，而且这些offset是有序的。

如果有follower向leader发送了请求1，接着发送请求2，请求3，那么再发送请求4，这时就意味着follower已经同步了前三条数据，否则不会发送请求4。leader通过跟踪 每一个follower的offset来判断它们的复制进度。

默认的，如果follower与leader之间超过10s内没有发送请求，或者说没有收到请求数据，此时该follower就会被认为“不同步副本”。而持续请求的副本就是“同步副本”，当leader发生故障时，只有“同步副本”才可以被选举为leader。其中的请求超时时间可以通过参数参数来配置。

我们希望每个分区的leader可以分布到不同的broker中，尽可能的达到负载均衡，所以会有一个首选首领，如果我们设置参数auto.leader.rebalance.enable为true，那么它会检查首选首领是否是真正的首领，如果不是，则会触发选举，让首选首领成为首领。

消费组选主

在kafka的消费端，会有一个消费者协调器以及消费组，组协调器groupcoordinator需要为消费组内的消费者选举出一个消费组的leader，那么如何选举的呢？

如果消费组内还没有leader，那么第一个加入消费组的消费者即为消费组的leader，如果某一个时刻leader消费者由于某些原因退出了消费组，那么就会重新选举leader，如何选举？

private val members = new mutable.hashmap[string, membermetadata]
leaderid = members.keys.headoption

上面代码是kafka源码中的部分代码，member是一个hashmap的数据结构，key为消费者的member_id，value是元数据信息，那么它会将leaderid选举为hashmap中的第一个键值对，它和随机基本没啥区别。

对于整个选举算法的详情需要先了解raft选举算法，kafka是基于该算法来实现leader选举的

消息发送方式

先了解kafka消息的发送方式。

• kafka消息发送分同步(sync)、异步(async)两种方式
• 默认是使用同步方式，可通过producer.type属性进行配置；
• kafka保证消息被安全生产，有三个选项分别是0,1,-1
• 通过request.required.acks属性进行配置：
  • 0代表：不进行消息接收是否成功的确认(默认值)；
  • 1代表：当leader副本接收成功后，返回接收成功确认信息；
  • -1代表：当leader和follower副本都接收成功后，返回接收成功确认信息；

六种发送场景

两个维度相交，生成六种情况，如下图：

消息丢失的场景

• 网络异常
  • acks设置为0时，不和kafka集群进行消息接受确认，当网络发生异常等情况时，存在消息丢失的可能；
• 客户端异常
  • 异步发送时，消息并没有直接发送至kafka集群，而是在client端按一定规则缓存并批量发送。在这期间，如果客户端发生死机等情况，都会导致消息的丢失；
• 缓冲区满了
  • 异步发送时，client端缓存的消息超出了缓冲池的大小，也存在消息丢失的可能；
• leader副本异常
  • acks设置为1时，leader副本接收成功，kafka集群就返回成功确认信息，而follower副本可能还在同步。这时leader副本突然出现异常，新leader副本(原follower副本)未能和其保持一致，就会出现消息丢失的情况；

以上就是消息丢失的几种情况，在日常应用中，我们需要结合自身的应用场景来选择不同的配置。

消息消费

• kafka消息消费有两个consumer接口，low-level api和high-level api：
  • low-level api：消费者自己维护offset等值，可以实现对kafka的完全控制；
  • high-level api：封装了对parition和offset的管理，使用简单；

如果使用高级接口high-level api，可能存在一个问题就是当消息消费者从集群中把消息取出来、并提交了新的消息offset值后，还没来得及消费就挂掉了，那么下次再消费时之前没消费成功的消息就“诡异”的消失了；

解决办法：

• 针对消息丢失：同步模式下，确认机制设置为-1，即让消息写入leader和follower之后再确认消息发送成功；异步模式下，为防止缓冲区满，可以在配置文件设置不限制阻塞超时时间，当缓冲区满时让生产者一直处于阻塞状态；
• 针对消息重复：将消息的唯一标识保存到外部介质中，每次消费时判断是否处理过即可。

复制备份

备份机制是kafka0.8版本的新特性，备份机制的出现大大提高了kafka集群的可靠性、稳定性。有了备份机制后，kafka允许集群中的节点挂掉后而不影响整个集群工作。

一个备份数量为n的集群允许n-1个节点失败。在所有备份节点中，有一个节点作为lead节点，这个节点保存了其它备份节点列表，并维持各个备份间的状体同步。

kafka消息备份分三个角色：分别是leader副本、follower副本、isr集合

• leader副本：负责直接响应client端的读写请求，即和生产者和消费者直接对接，生产者生产一条消息，直接进入leader副本；
• follower副本：作为特殊消费者，被动的接收leader副本中的数据。注意：follower副本不能响应client端的读写请求；
• isr集合：与leader保持同步的follower，属于isr副本集合(同步的备份集合)，反过来说，在某个时刻，还在被动接收接收，不是和leader完全一致的，不能属于isr副本集合，同步完成后才属于isr集合；

isr集合作用

在当前leader不可用时，kafka集群会从isr集合中选取一个follower升级为新leader；通过维护isr集合，一个拥有(n 1)个备份的topic可用容忍n个备份不可用

复制提供了高可用性，producer继续发布消息，consumer继续接受消息。
leader处理此分区的所有的读写请求，而follower被动的复制数据。（所有的写都发给leader, 然后leader将消息发给follower。）

一个broker既可能是一个分区的leader,也可能是另一个分区的slave。kafka实际是保证在足够多的slave写入成功的情况下就认为消息写入成功，而不是全部写入成功

kafka引入了 isr的概念。isr是in-sync replicas的简写。isr的副本保持和leader的同步，当然leader本身也在isr中。初始状态所有的副本都处于isr中，当一个消息发送给leader的时候，leader会等待isr中所有的副本告诉它已经接收了这个消息，如果一个副本失败了，那么它会被移除isr。下一条消息来的时候，leader就会将消息发送给当前的isr中节点了。

kafka使用zookeeper实现leader选举。如果leader失败，controller会从isr选出一个新的leader。

kafka引入了 isr的概念。isr是in-sync replicas的简写。isr的副本保持和leader的同步，当然leader本身也在isr中。初始状态所有的副本都处于isr中，当一个消息发送给leader的时候，leader会等待isr中所有的副本告诉它已经接收了这个消息，如果一个副本失败了，那么它会被移除isr。下一条消息来的时候，leader就会将消息发送给当前的isr中节点了。

支持丰富的功能

• 发布/订阅消息传递模型
• 财务级交易消息
• 各种跨语言客户端，例如java，c / c ，python，go
• 可插拔的传输协议，例如tcp，ssl，aio
• 内置的消息跟踪功能，还支持开放式跟踪
• 多功能的大数据和流生态系统集成
• 按时间或偏移量追溯消息
• 可靠的fifo和严格的有序消息传递在同一队列中
• 高效的推拉消费模型
• 单个队列中的百万级消息累积容量
• 多种消息传递协议，例如jms和openmessaging
• 灵活的分布式横向扩展部署架构
• 快如闪电的批量消息交换系统
• 各种消息过滤器机制，例如sql和tag
• 用于隔离测试和云隔离群集的docker映像
• 功能丰富的管理仪表板，用于配置，指标和监视
• 认证与授权

他的核心模块：

• rocketmq-broker：接受生产者发来的消息并存储（通过调用rocketmq-store），消费者从这里取得消息
• rocketmq-client：提供发送、接受消息的客户端api。
• rocketmq-namesrv：nameserver，类似于zookeeper，这里保存着消息的topicname，队列等运行时的元信息。
• rocketmq-common：通用的一些类，方法，数据结构等。
• rocketmq-remoting：基于netty4的client/server fastjson序列化 自定义二进制协议。
• rocketmq-store：消息、索引存储等。
• rocketmq-filtersrv：消息过滤器server，需要注意的是，要实现这种过滤，需要上传代码到mq！（一般而言，我们利用tag足以满足大部分的过滤需求，如果更灵活更复杂的过滤需求，可以考虑filtersrv组件）。
• rocketmq-tools：命令行工具。

部署架构

tip：我们可以看到rocketmq啥都是集群部署的，这是他吞吐量大，高可用的原因之一，集群的模式也很花哨，可以支持多master 模式、多master多slave异步复制模式、多 master多slave同步双写模式。

nameserver

主要负责对于源数据的管理，包括了对于topic和路由信息的管理。

nameserver是一个功能齐全的服务器，其角色类似dubbo中的zookeeper，但nameserver与zookeeper相比更轻量。主要是因为每个nameserver节点互相之间是独立的，没有任何信息交互。

nameserver压力不会太大，平时主要开销是在维持心跳和提供topic-broker的关系数据。

但有一点需要注意，broker向nameserver发心跳时， 会带上当前自己所负责的所有topic信息，如果topic个数太多（万级别），会导致一次心跳中，就topic的数据就几十m，网络情况差的话， 网络传输失败，心跳失败，导致nameserver误认为broker心跳失败。

nameserver 被设计成几乎无状态的，可以横向扩展，节点之间相互之间无通信，通过部署多台机器来标记自己是一个伪集群。

每个 broker 在启动的时候会到 nameserver 注册，producer 在发送消息前会根据 topic 到 nameserver 获取到 broker 的路由信息，consumer 也会定时获取 topic 的路由信息。

所以从功能上看nameserver应该是和 zookeeper 差不多，据说 rocketmq 的早期版本确实是使用的 zookeeper ，后来改为了自己实现的 nameserver 。

producer

消息生产者，负责产生消息，一般由业务系统负责产生消息。

• producer由用户进行分布式部署，消息由producer通过多种负载均衡模式发送到broker集群，发送低延时，支持快速失败。
• rocketmq 提供了三种方式发送消息：同步、异步和单向
• 同步发送：同步发送指消息发送方发出数据后会在收到接收方发回响应之后才发下一个数据包。一般用于重要通知消息，例如重要通知邮件、营销短信。
• 异步发送：异步发送指发送方发出数据后，不等接收方发回响应，接着发送下个数据包，一般用于可能链路耗时较长而对响应时间敏感的业务场景，例如用户视频上传后通知启动转码服务。
• 单向发送：单向发送是指只负责发送消息而不等待服务器回应且没有回调函数触发，适用于某些耗时非常短但对可靠性要求并不高的场景，例如日志收集。

broker

消息中转角色，负责存储消息，转发消息。

• broker是具体提供业务的服务器，单个broker节点与所有的nameserver节点保持长连接及心跳，并会定时将topic信息注册到nameserver，顺带一提底层的通信和连接都是基于netty实现的。
• broker负责消息存储，以topic为纬度支持轻量级的队列，单机可以支撑上万队列规模，支持消息推拉模型。
• 江南app体育官方入口官网上有数据显示：具有上亿级消息堆积能力，同时可严格保证消息的有序性。

consumer

消息消费者，负责消费消息，一般是后台系统负责异步消费。

• consumer也由用户部署，支持push和pull两种消费模式，支持集群消费和广播消息，提供实时的消息订阅机制。
• pull：拉取型消费者（pull consumer）主动从消息服务器拉取信息，只要批量拉取到消息，用户应用就会启动消费过程，所以 pull 称为主动消费型。
• push：推送型消费者（push consumer）封装了消息的拉取、消费进度和其他的内部维护工作，将消息到达时执行的回调接口留给用户应用程序来实现。所以 push 称为被动消费类型，但从实现上看还是从消息服务器中拉取消息，不同于 pull 的是 push 首先要注册消费监听器，当监听器处触发后才开始消费消息。

消息领域模型

message

message（消息）就是要传输的信息。
一条消息必须有一个主题（topic），主题可以看做是你的信件要邮寄的地址。

一条消息也可以拥有一个可选的标签（tag）和额处的键值对，它们可以用于设置一个业务 key 并在 broker 上查找此消息以便在开发期间查找问题。

topic

topic（主题）可以看做消息的规类，它是消息的第一级类型。比如一个电商系统可以分为：交易消息、物流消息等，一条消息必须有一个 topic 。

topic 与生产者和消费者的关系非常松散，一个 topic 可以有0个、1个、多个生产者向其发送消息，一个生产者也可以同时向不同的 topic 发送消息。

一个 topic 也可以被 0个、1个、多个消费者订阅。

tag

tag（标签）可以看作子主题，它是消息的第二级类型，用于为用户提供额外的灵活性。使用标签，同一业务模块不同目的的消息就可以用相同 topic 而不同的 tag 来标识。比如交易消息又可以分为：交易创建消息、交易完成消息等，一条消息可以没有 tag 。

标签有助于保持您的代码干净和连贯，并且还可以为 rocketmq 提供的查询系统提供帮助。

group

分组，一个组可以订阅多个topic。

分为producergroup，consumergroup，代表某一类的生产者和消费者，一般来说同一个服务可以作为group，同一个group一般来说发送和消费的消息都是一样的

queue

在kafka中叫partition，每个queue内部是有序的，在rocketmq中分为读和写两种队列，一般来说读写队列数量一致，如果不一致就会出现很多问题。

message queue

message queue（消息队列），主题被划分为一个或多个子主题，即消息队列。

一个 topic 下可以设置多个消息队列，发送消息时执行该消息的 topic ，rocketmq 会轮询该 topic 下的所有队列将消息发出去。

消息的物理管理单位。一个topic下可以有多个queue，queue的引入使得消息的存储可以分布式集群化，具有了水平扩展能力。

offset

在rocketmq 中，所有消息队列都是持久化，长度无限的数据结构，所谓长度无限是指队列中的每个存储单元都是定长，访问其中的存储单元使用offset 来访问，offset 为 java long 类型，64 位，理论上在 100年内不会溢出，所以认为是长度无限。

也可以认为 message queue 是一个长度无限的数组，offset 就是下标。

消息消费模式

消息消费模式有两种：clustering（集群消费）和broadcasting（广播消费）。

默认情况下就是集群消费，该模式下一个消费者集群共同消费一个主题的多个队列，一个队列只会被一个消费者消费，如果某个消费者挂掉，分组内其它消费者会接替挂掉的消费者继续消费。

而广播消费消息会发给消费者组中的每一个消费者进行消费。

message order

message order（消息顺序）有两种：orderly（顺序消费）和concurrently（并行消费）。

顺序消费表示消息消费的顺序同生产者为每个消息队列发送的顺序一致，所以如果正在处理全局顺序是强制性的场景，需要确保使用的主题只有一个消息队列。

并行消费不再保证消息顺序，消费的最大并行数量受每个消费者客户端指定的线程池限制。

一次完整的通信流程是怎样的？

producer 与 nameserver集群中的其中一个节点（随机选择）建立长连接，定期从 nameserver 获取 topic 路由信息，并向提供 topic 服务的 broker master 建立长连接，且定时向 broker 发送心跳。

producer 只能将消息发送到 broker master，但是 consumer 则不一样，它同时和提供 topic 服务的 master 和 slave建立长连接，既可以从 broker master 订阅消息，也可以从 broker slave 订阅消息。

具体如下图：

nameservice启动流程

在org.apache.rocketmq.namesrv目录下的namesrvstartup这个启动类基本上描述了他的启动过程我们可以看一下代码：

• 第一步是初始化配置
• 创建namesrvcontroller实例，并开启两个定时任务：
• 每隔10s扫描一次broker，移除处于不激活的broker；
• 每隔10s打印一次kv配置。
• 第三步注册钩子函数，启动服务器并监听broker。

nameservice还有很多东西的哈我这里就介绍他的启动流程，大家还可以去看看代码，还是很有意思的，比如路由注册会发送心跳包，还有心跳包的处理流程，路由删除，路由发现等等。

tip：本来我想贴很多源码的，后面跟歪歪（java3y）讨论了很久做出了不贴的决定，大家理解过程为主！我主要是做只是扫盲还有一些痛点分析嘛，深究还是得大家花时间，我要啥都介绍篇幅就不够了。

producer

链路很长涉及的细节也多，看一下链路图。

producer是消息发送方，那他怎么发送的呢？

通过轮训，producer轮训某个topic下面的所有队列实现发送方的负载均衡

broker

broker在rocketmq中是进行处理producer发送消息请求，consumer消费消息的请求，并且进行消息的持久化，以及ha策略和服务端过滤，就是集群中很重的工作都是交给了broker进行处理。

broker模块是通过brokerstartup进行启动的，会实例化brokercontroller，并且调用其初始化方法

大家去看broker的源码的话会发现，他的初始化流程很冗长，会根据配置创建很多线程池主要用来发送消息、拉取消息、查询消息、客户端管理和消费者管理，也有很多定时任务，同时也注册了很多请求处理器，用来发送拉取消息查询消息的。

consumer

consumer是消息接受，那他怎么接收消息的呢？

消费端会通过rebalanceservice线程，10秒钟做一次基于topic下的所有队列负载。

总结一下

rocketmq优点：

• 单机吞吐量：十万级
• 可用性：非常高，分布式架构
• 消息可靠性：经过参数优化配置，消息可以做到0丢失
• 功能支持：mq功能较为完善，还是分布式的，扩展性好
• 支持10亿级别的消息堆积，不会因为堆积导致性能下降
• 源码是java，我们可以自己阅读源码，定制自己公司的mq，可以掌控
• 天生为金融互联网领域而生，对于可靠性要求很高的场景，尤其是电商里面的订单扣款，以及业务削峰，在大量交易涌入时，后端可能无法及时处理的情况
• roketmq在稳定性上可能更值得信赖，这些业务场景在阿里双11已经经历了多次考验，如果你的业务有上述并发场景，建议可以选择rocketmq

rocketmq缺点：

• 支持的客户端语言不多，目前是java及c ，其中c 不成熟
• 社区活跃度不是特别活跃那种
• 没有在 mq 核心中去实现jms等接口，有些系统要迁移需要修改大量代码

消息去重

去重原则：使用业务端逻辑保持幂等性

幂等性：就是用户对于同一操作发起的一次请求或者多次请求的结果是一致的，不会因为多次点击而产生了副作用，数据库的结果都是唯一的，不可变的。

只要保持幂等性，不管来多少条重复消息，最后处理的结果都一样，需要业务端来实现。

去重策略：保证每条消息都有唯一编号(比如唯一流水号)，且保证消息处理成功与去重表的日志同时出现。

建立一个消息表，拿到这个消息做数据库的insert操作。给这个消息做一个唯一主键（primary key）或者唯一约束，那么就算出现重复消费的情况，就会导致主键冲突，那么就不再处理这条消息。

消息重复

消息领域有一个对消息投递的qos定义，分为：

• 最多一次（at most once）
• 至少一次（at least once）
• 仅一次（ exactly once）

qos：quality of service，服务质量

几乎所有的mq产品都声称自己做到了at least once。

既然是至少一次，那避免不了消息重复，尤其是在分布式网络环境下。

比如：网络原因闪断，ack返回失败等等故障，确认信息没有传送到消息队列，导致消息队列不知道自己已经消费过该消息了，再次将该消息分发给其他的消费者。

不同的消息队列发送的确认信息形式不同，例如rabbitmq是发送一个ack确认消息，rocketmq是返回一个consume_success成功标志，kafka实际上有个offset的概念。

rocketmq没有内置消息去重的江南app体育官方入口的解决方案，最新版本是否支持还需确认。

消息的可用性

当我们选择好了集群模式之后，那么我们需要关心的就是怎么去存储和复制这个数据，rocketmq对消息的刷盘提供了同步和异步的策略来满足我们的，当我们选择同步刷盘之后，如果刷盘超时会给返回flush_disk_timeout，如果是异步刷盘不会返回刷盘相关信息，选择同步刷盘可以尽最大程度满足我们的消息不会丢失。

除了存储有选择之后，我们的主从同步提供了同步和异步两种模式来进行复制，当然选择同步可以提升可用性，但是消息的发送rt时间会下降10%左右。

rocketmq采用的是混合型的存储结构，即为broker单个实例下所有的队列共用一个日志数据文件（即为commitlog）来存储。

而kafka采用的是独立型的存储结构，每个队列一个文件。

这里帅丙认为，rocketmq采用混合型存储结构的缺点在于，会存在较多的随机读操作，因此读的效率偏低。同时消费消息需要依赖consumequeue，构建该逻辑消费队列需要一定开销。

rocketmq 刷盘实现

broker 在消息的存取时直接操作的是内存（内存映射文件），这可以提供系统的吞吐量，但是无法避免机器掉电时数据丢失，所以需要持久化到磁盘中。

刷盘的最终实现都是使用nio中的 mappedbytebuffer.force() 将映射区的数据写入到磁盘，如果是同步刷盘的话，在broker把消息写到commitlog映射区后，就会等待写入完成。

异步而言，只是唤醒对应的线程，不保证执行的时机，流程如图所示。

顺序消息

我简单的说一下我们使用的rocketmq里面的一个简单实现吧。

tip：为啥用rocketmq举例呢，这玩意是阿里开源的，我问了下身边的朋友很多公司都有使用，所以读者大概率是这个的话我就用这个举例吧，具体的细节我后面会在rocketmq和kafka各自章节说到。

生产者消费者一般需要保证顺序消息的话，可能就是一个业务场景下的，比如订单的创建、支付、发货、收货。

那这些东西是不是一个订单号呢？一个订单的肯定是一个订单号的说，那简单了呀。

一个topic下有多个队列，为了保证发送有序，rocketmq提供了messagequeueselector队列选择机制，他有三种实现:

我们可使用hash取模法，让同一个订单发送到同一个队列中，再使用同步发送，只有同个订单的创建消息发送成功，再发送支付消息。这样，我们保证了发送有序。

rocketmq的topic内的队列机制,可以保证存储满足fifo（first input first output 简单说就是指先进先出）,剩下的只需要消费者顺序消费即可。

rocketmq仅保证顺序发送，顺序消费由消费者业务保证!!!

这里很好理解，一个订单你发送的时候放到一个队列里面去，你同一个的订单号hash一下是不是还是一样的结果，那肯定是一个消费者消费，那顺序是不是就保证了？

真正的顺序消费不同的中间件都有自己的不同实现我这里就举个例子

分布式事务

half message(半消息)

是指暂不能被consumer消费的消息。producer 已经把消息成功发送到了 broker 端，但此消息被标记为暂不能投递状态，处于该种状态下的消息称为半消息。需要 producer

对消息的二次确认后，consumer才能去消费它。

消息回查

由于网络闪段，生产者应用重启等原因。导致 producer 端一直没有对 half message(半消息) 进行 二次确认。这是brock服务器会定时扫描长期处于半消息的消息，会

主动询问 producer端 该消息的最终状态(commit或者rollback),该消息即为 消息回查。

1. a服务先发送个half message给brock端，消息中携带 b服务 即将要 100元的信息。
2. 当a服务知道half message发送成功后，那么开始第3步执行本地事务。
3. 执行本地事务(会有三种情况1、执行成功。2、执行失败。3、网络等原因导致没有响应)
4. 如果本地事务成功，那么product像brock服务器发送commit,这样b服务就可以消费该message。
5. 如果本地事务失败，那么product像brock服务器发送rollback,那么就会直接删除上面这条半消息。
6. 如果因为网络等原因迟迟没有返回失败还是成功，那么会执行rocketmq的回调接口,来进行事务的回查。

消息过滤

• broker端消息过滤　　
  在broker中，按照consumer的要求做过滤，优点是减少了对于consumer无用消息的网络传输。缺点是增加了broker的负担，实现相对复杂。
• consumer端消息过滤
  这种过滤方式可由应用完全自定义实现，但是缺点是很多无用的消息要传输到consumer端。

broker的buffer问题

broker的buffer通常指的是broker中一个队列的内存buffer大小，这类buffer通常大小有限。

另外，rocketmq没有内存buffer概念，rocketmq的队列都是持久化磁盘，数据定期清除。

rocketmq同其他mq有非常显著的区别，rocketmq的内存buffer抽象成一个无限长度的队列，不管有多少数据进来都能装得下，这个无限是有前提的，broker会定期删除过期的数据。

例如broker只保存3天的消息，那么这个buffer虽然长度无限，但是3天前的数据会被从队尾删除。

回溯消费

回溯消费是指consumer已经消费成功的消息，由于业务上的需求需要重新消费，要支持此功能，broker在向consumer投递成功消息后，消息仍然需要保留。并且重新消费一般是按照时间维度。

例如由于consumer系统故障，恢复后需要重新消费1小时前的数据，那么broker要提供一种机制，可以按照时间维度来回退消费进度。

rocketmq支持按照时间回溯消费，时间维度精确到毫秒，可以向前回溯，也可以向后回溯。

消息堆积

消息中间件的主要功能是异步解耦，还有个重要功能是挡住前端的数据洪峰，保证后端系统的稳定性，这就要求消息中间件具有一定的消息堆积能力，消息堆积分以下两种情况：

• 消息堆积在内存buffer，一旦超过内存buffer，可以根据一定的丢弃策略来丢弃消息，如corba notification规范中描述。适合能容忍丢弃消息的业务，这种情况消息的堆积能力主要在于内存buffer大小，而且消息堆积后，性能下降不会太大，因为内存中数据多少对于对外提供的访问能力影响有限。
• 消息堆积到持久化存储系统中，例如db，kv存储，文件记录形式。当消息不能在内存cache命中时，要不可避免的访问磁盘，会产生大量读io，读io的吞吐量直接决定了消息堆积后的访问能力。
• 评估消息堆积能力主要有以下四点：
• 消息能堆积多少条，多少字节？即消息的堆积容量。
• 消息堆积后，发消息的吞吐量大小，是否会受堆积影响？
• 消息堆积后，正常消费的consumer是否会受影响？
• 消息堆积后，访问堆积在磁盘的消息时，吞吐量有多大？

定时消息

定时消息是指消息发到broker后，不能立刻被consumer消费，要到特定的时间点或者等待特定的时间后才能被消费。

如果要支持任意的时间精度，在broker层面，必须要做消息排序，如果再涉及到持久化，那么消息排序要不可避免的产生巨大性能开销。

rocketmq支持定时消息，但是不支持任意时间精度，支持特定的level，例如定时5s，10s，1m等。

出处:北京英浦教育

本作品采用《cc 协议》，转载必须注明作者和本文链接