09 生产者消息分区机制原理剖析
我们在使用Apache Kafka生产和消费消息的时候，肯定是希望能够将数据均匀地分配到所有服务器上。比如很多公司使用Kafka收集应用服务器的日志数据，这种数据都是很多的，特别是对于那种大批量机器组成的集群环境，每分钟产生的日志量都能以GB数，因此如何将这么大的数据量均匀地分配到Kafka的各个Broker上，就成为一个非常重要的问题。
今天我就来和你说说Kafka生产者如何实现这个需求，我会以Java API为例进行分析，但实际上其他语言的实现逻辑也是类似的。
为什么分区？
如果你对Kafka分区（Partition）的概念还不熟悉，可以先返回专栏第2期回顾一下。专栏前面我说过Kafka有主题（Topic）的概念，它是承载真实数据的逻辑容器，而在主题之下还分为若干个分区，也就是说Kafka的消息组织方式实际上是三级结构：主题-分区-消息。主题下的每条消息只会保存在某一个分区中，而不会在多个分区中被保存多份。官网上的这张图非常清晰地展示了Kafka的三级结构，如下所示：
现在我抛出一个问题你可以先思考一下：你觉得为什么Kafka要做这样的设计？为什么使用分区的概念而不是直接使用多个主题呢？
其实分区的作用就是提供负载均衡的能力，或者说对数据进行分区的主要原因，就是为了实现系统的高伸缩性（Scalability）。不同的分区能够被放置到不同节点的机器上，而数据的读写操作也都是针对分区这个粒度而进行的，这样每个节点的机器都能独立地执行各自分区的读写请求处理。并且，我们还可以通过添加新的节点机器来增加整体系统的吞吐量。
实际上分区的概念以及分区数据库早在1980年就已经有大牛们在做了，比如那时候有个叫Teradata的数据库就引入了分区的概念。
值得注意的是，不同的分布式系统对分区的叫法也不尽相同。比如在Kafka中叫分区，在MongoDB和Elasticsearch中就叫分片Shard，而在HBase中则叫Region，在Cassandra中又被称作vnode。从表面看起来它们实现原理可能不尽相同，但对底层分区（Partitioning）的整体思想却从未改变。
除了提供负载均衡这种最核心的功能之外，利用分区也可以实现其他一些业务级别的需求，比如实现业务级别的消息顺序的问题，这一点我今天也会分享一个具体的案例来说明。
都有哪些分区策略？
下面我们说说Kafka生产者的分区策略。所谓分区策略是决定生产者将消息发送到哪个分区的算法。Kafka为我们提供了默认的分区策略，同时它也支持你自定义分区策略。
如果要自定义分区策略，你需要显式地配置生产者端的参数partitioner.class。这个参数该怎么设定呢？方法很简单，在编写生产者程序时，你可以编写一个具体的类实现org.apache.kafka.clients.producer.Partitioner接口。这个接口也很简单，只定义了两个方法：partition()和close()，通常你只需要实现最重要的partition方法。我们来看看这个方法的方法签名：
int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster);
这里的topic、key、keyBytes、value和valueBytes都属于消息数据，cluster则是集群信息（比如当前Kafka集群共有多少主题、多少Broker等）。Kafka给你这么多信息，就是希望让你能够充分地利用这些信息对消息进行分区，计算出它要被发送到哪个分区中。只要你自己的实现类定义好了partition方法，同时设置partitioner.class参数为你自己实现类的Full Qualified Name，那么生产者程序就会按照你的代码逻辑对消息进行分区。虽说可以有无数种分区的可能，但比较常见的分区策略也就那么几种，下面我来详细介绍一下。
轮询策略
也称Round-robin策略，即顺序分配。比如一个主题下有3个分区，那么第一条消息被发送到分区0，第二条被发送到分区1，第三条被发送到分区2，以此类推。当生产第4条消息时又会重新开始，即将其分配到分区0，就像下面这张图展示的那样。
这就是所谓的轮询策略。轮询策略是Kafka Java生产者API默认提供的分区策略。如果你未指定partitioner.class参数，那么你的生产者程序会按照轮询的方式在主题的所有分区间均匀地“码放”消息。
轮询策略有非常优秀的负载均衡表现，它总是能保证消息最大限度地被平均分配到所有分区上，故默认情况下它是最合理的分区策略，也是我们最常用的分区策略之一。
随机策略
也称Randomness策略。所谓随机就是我们随意地将消息放置到任意一个分区上，如下面这张图所示。
如果要实现随机策略版的partition方法，很简单，只需要两行代码即可：
List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
return ThreadLocalRandom.current().nextInt(partitions.size());
先计算出该主题总的分区数，然后随机地返回一个小于它的正整数。
本质上看随机策略也是力求将数据均匀地打散到各个分区，但从实际表现来看，它要逊于轮询策略，所以如果追求数据的均匀分布，还是使用轮询策略比较好。事实上，随机策略是老版本生产者使用的分区策略，在新版本中已经改为轮询了。
按消息键保序策略
也称Key-ordering策略。有点尴尬的是，这个名词是我自己编的，Kafka官网上并无这样的提法。
Kafka允许为每条消息定义消息键，简称为Key。这个Key的作用非常大，它可以是一个有着明确业务含义的字符串，比如客户代码、部门编号或是业务ID等；也可以用来表征消息元数据。特别是在Kafka不支持时间戳的年代，在一些场景中，工程师们都是直接将消息创建时间封装进Key里面的。一旦消息被定义了Key，那么你就可以保证同一个Key的所有消息都进入到相同的分区里面，由于每个分区下的消息处理都是有顺序的，故这个策略被称为按消息键保序策略，如下图所示。
实现这个策略的partition方法同样简单，只需要下面两行代码即可：
List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
return Math.abs(key.hashCode()) % partitions.size();
前面提到的Kafka默认分区策略实际上同时实现了两种策略：如果指定了Key，那么默认实现按消息键保序策略；如果没有指定Key，则使用轮询策略。
在你了解了Kafka默认的分区策略之后，我来给你讲一个真实的案例，希望能加强你对分区策略重要性的理解。
我曾经给一个国企进行过Kafka培训，当时碰到的一个问题就是如何实现消息的顺序问题。这家企业发送的Kafka的消息是有因果关系的，故处理因果关系也必须要保证有序性，否则先处理了“果”后处理“因”必然造成业务上的混乱。
当时那家企业的做法是给Kafka主题设置单分区，也就是1个分区。这样所有的消息都只在这一个分区内读写，因此保证了全局的顺序性。这样做虽然实现了因果关系的顺序性，但也丧失了Kafka多分区带来的高吞吐量和负载均衡的优势。
后来经过了解和调研，我发现这种具有因果关系的消息都有一定的特点，比如在消息体中都封装了固定的标志位，后来我就建议他们对此标志位设定专门的分区策略，保证同一标志位的所有消息都发送到同一分区，这样既可以保证分区内的消息顺序，也可以享受到多分区带来的性能红利。
这种基于个别字段的分区策略本质上就是按消息键保序的思想，其实更加合适的做法是把标志位数据提取出来统一放到Key中，这样更加符合Kafka的设计思想。经过改造之后，这个企业的消息处理吞吐量一下提升了40多倍，从这个案例你也可以看到自定制分区策略的效果可见一斑。
其他分区策略
上面这几种分区策略都是比较基础的策略，除此之外你还能想到哪些有实际用途的分区策略？其实还有一种比较常见的，即所谓的基于地理位置的分区策略。当然这种策略一般只针对那些大规模的Kafka集群，特别是跨城市、跨国家甚至是跨大洲的集群。
我就拿“极客时间”举个例子吧，假设极客时间的所有服务都部署在北京的一个机房（这里我假设它是自建机房，不考虑公有云方案。其实即使是公有云，实现逻辑也差不多），现在极客时间考虑在南方找个城市（比如广州）再创建一个机房；另外从两个机房中选取一部分机器共同组成一个大的Kafka集群。显然，这个集群中必然有一部分机器在北京，另外一部分机器在广州。
假设极客时间计划为每个新注册用户提供一份注册礼品，比如南方的用户注册极客时间可以免费得到一碗“甜豆腐脑”，而北方的新注册用户可以得到一碗“咸豆腐脑”。如果用Kafka来实现则很简单，只需要创建一个双分区的主题，然后再创建两个消费者程序分别处理南北方注册用户逻辑即可。
但问题是你需要把南北方注册用户的注册消息正确地发送到位于南北方的不同机房中，因为处理这些消息的消费者程序只可能在某一个机房中启动着。换句话说，送甜豆腐脑的消费者程序只在广州机房启动着，而送咸豆腐脑的程序只在北京的机房中，如果你向广州机房中的Broker发送北方注册用户的消息，那么这个用户将无法得到礼品！
此时我们就可以根据Broker所在的IP地址实现定制化的分区策略。比如下面这段代码：
List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
return partitions.stream().filter(p -> isSouth(p.leader().host())).map(PartitionInfo::partition).findAny().get();
我们可以从所有分区中找出那些Leader副本在南方的所有分区，然后随机挑选一个进行消息发送。
小结
今天我们讨论了Kafka生产者消息分区的机制以及常见的几种分区策略。切记分区是实现负载均衡以及高吞吐量的关键，故在生产者这一端就要仔细盘算合适的分区策略，避免造成消息数据的“倾斜”，使得某些分区成为性能瓶颈，这样极易引发下游数据消费的性能下降。
开放讨论
在你的生产环境中使用最多的是哪种消息分区策略？实际在使用过程中遇到过哪些“坑”？
欢迎写下你的思考和答案，我们一起讨论。如果你觉得有所收获，也欢迎把文章分享给你的朋友。
- QQ怪 👍（50） 💬（5）
我们公司一直使用单个分区保持消息顺序性，看了老师分享的东西收益很多啊，准备回去好好分析改造下
2019-06-22 - Adol 👍（46） 💬（4）
老师好，在消息重试的时候，分区策略会重新再计算一次吗？比如一开始选择到5号分区，但是5号分区有问题导致重试，重试的时候可以重试发送到别的分区上吗？
2019-06-23 - jc9090kkk 👍（39） 💬（3）
感谢老师的分享，对于按消息键保序策略有一个疑问，假如我现在的业务数据定义了三个key，但是这三个key对应的消息生产速率不一致，按照老师上面的示意图展示的是，特定的key只会存储在特定的一个分区中，那岂不是牺牲了拓展性么，如果其中一个key的生产速率非常大，而另外2个key没那么大，却会一直占用分区，不会造成分区的空间浪费吗？还是我理解的有问题吗？希望老师解答一下，谢谢
2019-09-20 - hgf 👍（34） 💬（1）
老师您好，跨地区的kafka集群，创建的两个partition都在一个地方怎么办呢？创建topic时可以选择在哪些节点上创建partition吗？默认是随机选择节点创建partition吗？
2019-07-10 - 风轻扬 👍（27） 💬（7）
老师，我见到有网友提问，说是消费者出现reblance的情况时。key-ordering策略可能会导致消费了“因“，reblance之后，无法消费 “果“。您给出的建议是，显示设置consumer端参数partition.assignment.strategy。这个设置。是不是只要使用了key保序策略，就一定要设置上呢？消费过程中出现reblance是很正常的啊
2019-07-28 - WL 👍（20） 💬（2）
老师能不能有空能不能讲讲kafka和rocketMQ的对比, 我用下来感觉整体挺像的但是具体使用场景和性能优劣方面还是有点不知道该使用选择, 谢谢.
2019-06-22 - 嘉嘉☕ 👍（18） 💬（1）
老师好, 关于生产消息, 我有个问题请教一下老师. 生产者生产消息, 采用轮询策略, 假如轮询到分区A了, 分区A的leader所在的broker有些异常(比如不能及时给出响应), 此时, kafka的重试机制是怎样的 ? 谢谢
2019-11-28 - 我已经设置了昵称 👍（16） 💬（1）
广州机房怎么消费广州partition的数据，consumer如何指定消费的partition。这个能讲下吗
2019-06-22 - 海贼王 👍（15） 💬（8）
感觉这篇文章没有回答怎么保证分区里数据和生产者消息顺序是一致这个问题。 这里有两个例子：1是，一个生产者，发两次消息，但是网络原因，消息到达的顺序和消息发送的顺序不一致，2是两个生产者，这时候消息如何确定消息的顺序呢。 场景1在做CDC，同步数据库数据到异构数据系统很常见 场景2对某些业务很重要 对于第一种，我的一个思路是保证只有一个生产者，且设置生产者的ack为all级别 对于场景2就不知道kafka能怎么做了。 希望老师能解惑
2019-07-07 - Geek_b809ff 👍（14） 💬（2）
老师，想请教几个问题： 1、key是不是必须得完全一样，才能保证会发送到同一个分区？ 2、如果kafka搭了集群，有三个broker，分别是broker1、broker2、broker3。这时候我对名称为test的topic发送消息，key设置为A，消息会随机发送到三个broker上去吗？那这样的话顺序不就乱了吗？如果我想保证所有的消息都顺序，是不是需要指定发送到其中一个broker？
2019-08-29 - 随心而至 👍（11） 💬（1）
看了《数据库系统概念》，感觉基本上逃不开里面讲的，只是不同框架有自己的叫法罢了
2020-03-25 - hgf 👍（11） 💬（2）
如果消费过程中出现rebalance，那么可能造成因果关系之消费了因后rebalance，然后不处理之前的partition了，后面的消费者也无法处理该partition的“果”，请问，您对这种情况怎么处理的呢？
2019-07-10 - 余生尽是归途 👍（7） 💬（2）
胡老师，我们这边的场景是，解析上游推送的数据文件（xml格式文件达成的GZ数据压缩包，每个30M-、100M不等），我们采用的方式是，将数据包的路径作为消息发送至kafka（50个分区），然后sparkstreaming并行消费（手动提交的方式消费，数据包解析完成才视为消息消费成功）50个分区的消息（spark通过路径信息去解压解析数据包）。由于数据包大小差异较大，如果按照轮询的方式，在运气不好的情况下，某个分许存放了大量较大数据包，那么这个分区就会成为这一批次数据的瓶颈，后续的处理流程智能等待该分区消费完后才能处理。所以我采用了每次将生产者发送的消息发送至lag最小的分区，在一定程度上避免这个问题。但是这样每次投递消息之前我需要首先查询所有分区的lag值，然后选择lag值较小的分区进行投递，还好，我们的数据包个数不算使特别大（一个小时就几万个，并且消费逻辑重，主要时间消耗为数据消费时间）。这种情况下，分区的策略有没有什么更好的选择呢
2019-10-11 - 风中花 👍（7） 💬（3）
打卡继续。老师我有个小白问题 按消息键保序策略 实现方法 return Math.abs(key.hashCode()) % partitions.size(); 如果key 不变，增加分区数(partitions.size();)。那么这个算法，是不是就变成原来key1的消息在1区，增加分区后会不会变成ke1的消息放到其他区呢？ 我的理解是不是不对啊？
2019-06-24 - 狼暴暴 👍（6） 💬（1）
老师，分区里消息有序是指可以根据生产者生成消息数据的时间保证有序还是根据发送到分区的时间保证有序？
2020-02-08