一个专注于运维领域的个人技术博客

消费组应该算是kafka中一个比较有特色的设计模式了，而他的重平衡机制也是我们在实际生产使用中，无法避免的一个问题。

消费组

Consumer Group为kafka提供了可扩展、高容错特性的消费者机制。简单介绍下，大致有以下特点：

• 一个Consumer Group内可以有多个Consumer实例，该实例可以是一个进程，也可以是进程下的多线程
• 每个Consumer Group有一个唯一标识的Group ID
• 不同Consumer Group之间相互独立，互不影响
• Consumer Group内实例，与订阅的topic分区关系是一对一，或一对多的关系，Consumer Group会通过Coordinator尽量保持公平分配

理想情况下，我们应该设置Consumer实例的数量等于该Group订阅topic的分区总数，可以最大发挥消费性能。若设置的Consumer实例数少于订阅的分区数，则会为每个Consumer实例分配多个分区，消费性能会有所下降。若设置的Consumer实例数大于订阅的分区数，则会为每个Consumer实例分配 1 个分区进行消费，多余的Consumer实例则会闲置，只会浪费资源。

重平衡

重平衡（Rebalance）就是让一个Consumer Group下所有的Consumer实例,合理分配消费订阅topic的所有分区的过程。有 3 种情况会触发Consumer Group的Rebalance：

1. Group下实例数发生变化。有新的Consumer实例加入或者离开组。
2. 订阅的topic数发生变化。Consumer Group可以使用正则的方式订阅topic，比如 consumer.subscribe(Pattern.compile(“public.*log”))，该Group订阅所有以 public 开头，log 结尾的topic。这期间，新建了一个满足这样条件的topic，那么该Group也会发生Rebalance。
3. topic分区数发生变化。比如topic扩分区的时候，也会触发Rebalance。

单看上面任一触发条件，都没啥毛病。问题在于Rebalance过程中会出现以下问题：

• Rebalance过程的表现有些类似JVM FGC的情况，期间整个应用都会夯住，所有Consumer实例都会停止消费，等待Rebalance完成。
• Rebalance过程中，所有Consumer实例都会参与重新分配。即便Consumer Group中部分Consumer实例分配合理，也需要打散重新分配，会导致TCP重新建立连接，是一个比较重的操作，较为浪费资源。
• Rebalance的耗时取决于Consumer Group下的实例数量，一旦实例数过多，耗时极长，会造成大量消费延迟。

避免重平衡

对于上述Rebalance带来的一些弊端，从目前的社区版来看，暂时还没有很好的解决办法，我们只能尽量避免Rebalance的发生。
在生产业务场景中，很多Rebalance都是预期外或者不必要的。我们应用的TPS大多是被这类Rebalance拖慢的。

从上述的 3 个Rebalance触发条件抓手，后两条topic数量及分区数变化，一般都是主动运维的相关操作，这种操作带来的Rebalance一般是必然发生，难以避免的，我们组要来讨论下Consumer Group组成员变化引发的Rebalance。

Consumer Group实例增加的情况比较单一，当新启动一个Consumer的group.id已经存在，Coordinator会接管这个新实例，将其加入group.id相同的组，并重分配分区。这种操作场景，一般都还是预期内的，可能是通过扩容来提高TPS的操作。
Consumer Group实例数减少的情况就比较复杂了。除了正常停止下线某些Consumer实例，还会出现Coordinator误判实例为已停止状态，从而主动踢出Group。导致Rebalance发生。每个Consumer会定期向Coordinator发心跳包，保持keepalive。如果因为某些特殊原因，如网络抖动时，某个Consumer实例没有及时发送心跳请求，Coordinator会将其判定为离线，并从Group中移除，并开启新一轮Rebalance。针对这个问题，可以通过设置Consumer端一下几个参数来进行优化调整：

• session.timeout.ms
  即Consumer Group内实例的心跳超时时间，默认值是 10s
• heartbeat.interval.ms
  即心跳请求频率，频繁发送心跳请求会额外消耗带宽资源，但是能够更及时的触发Rebalance，默认值为 3s
• max.poll.interval.ms
  调用poll方法的时间间隔，默认值为 5min。期间没消费完poll回的消息，Coordinator会开启新一轮Rebalance

根据平时的实践经验，建议:
session.timeout.ms=6s
heartbeat.interval.ms=2s
原则上最好是满足session.timeout.ms >= 3 * heartbeat.interval.ms公式。

max.poll.interval.ms则需要根据下游实际消费能力进行调整，尽量设置的大一点，需要大于下游的最大消息处理时间。

如果进行完上述的各种调整后，还是频发触发Rebalance，最好再去排查下Consumer端的 GC 情况，实际生产环境中我经常碰到因为 GC 设置问题导致的Consumer程序频发 FGC 的问题，从而导致非预期内的Rebalance。

之前和大家聊过kafka是如何保证消息不丢失的，今天再讲讲在不丢消息的同时，如何实现精确一次处理的语义实现。

消息组件对消息的可靠性保障，常见的模式有3种：

• 最多一次(at most once)：消息可能会丢失，但不会重复
• 至少一次(at least once)：消息不会丢失，但有可能重复
• 精确一次(exactly once)：消息不会丢失，且不会重复，精准一次发送

kafka默认情况下，提供的是至少一次的可靠性保障。即broker保障已提交的消息的发送，但是遇上某些意外情况，如：网络抖动，超时等问题，导致Producer没有收到broker返回的数据ack，则Producer会继续重试发送消息，从而导致消息重复发送。
相应的，如果我们禁止Producer的失败重试发送功能，消息要么写入成功，要么写入失败，但绝不会重复发送。这样就是最多一次的消息保障模式。
但对于消息组件，排除特殊业务场景，我们追求的一定是精确一次的消息保障模式。kafka通过幂等性（Idempotence）和事务（Transaction）的机制，提供了这种精确的消息保障。

幂等

这里就不多说幂等的含义了，不清楚的自己查下资料。Producer默认不是幂等性的，向分区发送数据时，可能会出现同一条消息被发送多次导致消息重复的情况。但只需增加一些参数，即可开启幂等性。

1
2
3

props.put(“enable.idempotence”, ture)
或者
props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG， true)

开启enable.idempotence后，kafka就会自动帮你做好消息去重的一系列工作。底层具体实现原理很简单，就是用空间换时间的优化思路，即在broker端多存一些字段来标识数据的唯一性。当Producer发送了具有相同字段值的消息后，broker会进行匹配去重，丢弃重复的数据。实际的代码没这么简单，但大致是这么个处理逻辑。
官方的这个幂等实现看似简单高效，但也存在他的局限性。他只能保证单分区上的幂等性，即一个幂等性Producer只能够保证某个topic的一个分区上不出现重复消息，无法实现多分区的幂等。此外，如果Producer重启，也会导致幂等重置。

事务

对于多分区保证幂等的场景，则需要事务特性来处理了。kafka的事务跟我们常见数据库事务概念差不多，也是提供经典的ACID，即原子性（Atomicity）、一致性 (Consistency)、隔离性 (Isolation) 和持久性 (Durability)。
事务Producer保证消息写入分区的原子性，即这批消息要么全部写入成功，要么全失败。此外，Producer重启回来后，kafka依然保证它们发送消息的精确一次处理。
事务特性的配置也很简单：

• 和幂等Producer一样，开启enable.idempotence = true
• 设置Producer端参数transctional.id

事务Producer的代码稍微也有点不一样，需要调一些事务处理的API。数据的发送需要放在beginTransaction和commitTransaction之间。Consumer端的代码也需要加上isolation.level参数，用以处理事务提交的数据。示例代码:

1
2
3
4
5
6
7
8
9

producer.initTransactions();
try {
            producer.beginTransaction();
            producer.send(record1);
            producer.send(record2);
            producer.commitTransaction();
} catch (KafkaException e) {
            producer.abortTransaction();
}

事务Producer虽然在多分区的数据处理上保证了幂等，但是处理性能上相应的是会有一些下降的。

依赖redis实现幂等

这里为什么还要额外讲通过依赖redis来实现幂等呢？因为笔者在早期维护kafka相关应用时，那会0.8系列版本的kafka还没有这些自带的幂等事务特性，只能依靠开发者自己来实现。
常见的方式就是通过数据的业务属性来生成个uniqueId来维护到redis中，利用redis的高并发，高吞吐，分布式锁特性，让写入kafka多分区的数据前，先去redis中校验一下uniqueId等方式，来实现幂等。得益于redis的高性能，在保证幂等同时，还能不让消息数据吞吐性能下降太多。当然，因为redis的依赖引入，也增加了架构的复杂度，从运维上来说也增加了整体的故障点，其中取舍需要自己来全局判断。

这次大概先介绍了下kafka的幂等各种实现方式，实际在事务，和依赖redis分布式锁来实现幂等的方式中，还要许多点值得我们深究来聊一下的，篇幅所限，后续再细讲

今天和大家聊一下，kafka对于消息的可靠性保证。作为消息引擎组件，保证消息不丢失，是非常重要的。

那么kafka是如何保证消息不丢失的呢？

前提条件

任何消息组件不丢数据都是在特定场景下一定条件的，kafka要保证消息不丢，有两个核心条件。

第一，必须是已提交的消息，即committed message。kafka对于committed message的定义是，生产者提交消息到broker，并等到多个broker确认并返回给生产者已提交的确认信息。而这多个broker是由我们自己来定义的，可以选择只要有一个broker成功保存该消息就算是已提交，也可以是令所有broker都成功保存该消息才算是已提交。不论哪种情况，kafka只对已提交的消息做持久化保证。

第二，也就是最基本的条件，虽然kafka集群是分布式的，但也必须保证有足够broker正常工作，才能对消息做持久化做保证。也就是说 kafka不丢消息是有前提条件的，假如你的消息保存在 N 个kafka broker上，那么这个前提条件就是这 N 个broker中至少有 1 个存活。只要这个条件成立，kafka就能保证你的这条消息永远不会丢失。

今天继续和大家聊一下，kafka的各种发行版。kafka历经数年的发展，从最初纯粹的消息引擎，到近几年开始在流处理平台生态圈发力，衍生出了各种不同特性的版本。

你了解几种kafka

kafka的确有好几种，这里我不是指他的版本，是指存在多个组织或公司发布不同特性的kafka。你应该听说过Linux发行版，比如我们熟知的CentOS、RedHat、Ubuntu等，它们都是Linux系统，其实就是因为它们是不同公司发布的Linux系统，即不同的发行版。kafka也同样有多个发行版。

Apache Kafka

Apache Kafka是最“正统”的kafka，也应该是你最熟悉的发行版了。自kafka开源之初，它便在Apache基金会孵化并最终毕业成为顶级项目，也被称为社区版kafka。重要的是，它是后面其他所有发行版的基础。也就是说，后面提到的其他发行版，要么是原封不动地继承了Apache Kafka，要么是在此之上扩展了新功能，总之Apache Kafka是我们学习和使用kafka的基础。

上篇文章我们了解到，如果一个topic分区越多，理论上整个集群所能达到的吞吐量就越大。那么，分区数越多就越好吗？显然不是。今天我们来聊下kafka在分区数过多的情况下，会带来哪些弊端。

内存开销

客户端producer有个参数batch.size默认为16KB。它会为每个分区缓存消息，一旦批次数满了后，将消息批量发出。一般来说，这个设计是用于提升吞吐性能的。但是由于这个参数是partition级别的，如果分区数越多，这部分缓存所需的内存占用也会越多。假如有10000个分区，按照默认配置，这部分缓存就要占用约157MB的内存。而consumer端呢？抛开拉取数据所需的内存不说，单说线程的开销。如果还是10000个分区，同时consumer线程数要匹配分区数的话(大部分情况下是最佳的消费吞吐量配置)，那么在consumer client就要创建10000个线程，也需要创建大约10000个Socket去获取分区数据，这里面的线程切换的开销本身就已经不容小觑了。
服务器端的开销也不小，如果阅读kafka源码的话就会发现，服务器端的很多组件在内存中维护了partition级别的缓存，比如controller，FetcherManager等，因此分区数越多，这种缓存的成本就越大。

分区(partition)概念

要讲kafka分区数和吞吐量的关系，首先得理解什么是分区(partition)。

Partition是作用于具体的Topic而已的，而不是一个独立的概念。Partition能水平扩展客户端的读写性能，是高吞吐量的保障。通俗的讲，Partition就是一块保存具体数据的空间，本质就是磁盘上存放数据的文件夹，所以Partition是不能跨Broker存在的，也不能在同一个Broker上跨磁盘。对于一个Topic，可以根据需要设定Partition的个数。数据持久化时，每条消息都是根据一定的分区规则路由到对应的Partition中，并append在log文件的尾部(这一点类似于HDFS)，如上图；在同一个Partition中消息是顺序写入的且始终保持有序性；但是不同Partition之间不能保证消息的有序性(高吞吐量的保障)。kafka就是通过使用分区的设计将topic的消息打散到多个分区分布保存在不同的broker上，实现了producer和consumer消息处理的高吞吐量。

生产环境kafka集群，在数据量大的情况下，会出现单机各个磁盘间的占用不均匀情况，经常出现“一边倒”的情形。

原因探究

这是因为kafka只保证分区数量在各个磁盘上均匀分布，但它无法统计每个分区实际占用磁盘空间。因此很有可能出现某些分区消息数量巨大导致占用大量磁盘空间的情况。在1.1版本之前，用户对此基本没有优雅的处理方法，即便手动迁移日志文件和offset信息，也需要重启生效，风险极高。因为1.1之前kafka只支持分区数据在不同broker间的重分配，而无法做到在同一个broker下的不同磁盘间做重分配。1.1版本正式支持副本在不同路径间的迁移，具体的实现细节详见kafka官方wikiKIP-113。

目录间迁移步骤

假设我在server.properties文件中配置了多个日志存储路径(代表多块磁盘)，如下所示：

1
2
3

# A comma seperated list of directories under which to store log files
log.dirs=/data1/kafka-logs,/data2/kafka-logs,/data3/kafka-logs

生产环境的kafka集群扩容，是一个比较常见的需求和操作。然而kafka在新增节点后并不会像elasticsearch那样感知到新节点加入后，自动将数据reblance到整个新集群中，因此这个过程需要我们手动分配。

分区重分配方案

扩容后的数据均衡，其本质就是对topic进行分区重分配，数据迁移的过程。
在执行分区重分配的过程中，对集群的影响主要有两点：

1. 分区重分配主要是对topic数据进行Broker间的迁移，因此会占用集群的带宽资源；
2. 分区重分配会改变分区Leader所在的Broker，因此会影响客户端。

通过之前一系列的文章叙述，想必大家都对dr.elephant有了一个较为清晰的了解。通过自己线上经验的积累，以及和一些读者的交流，我汇总了一些大家在实战中遇到的问题和解决方案。

常规问题

由于在和读者交流的过程中，发现大家技术水平参差不齐，本着科普性文章的初衷，这里先讲一些比较基础的要点，大佬们可以忽略，直接跳过。

1. 在打包时，需要对照自己的Hadoop或者Spark版本，修改compile.conf文件中的版本号。否则有可能出现采集不到集群作业信息的情况。
2. 最好将自己Hadoop集群的相关配置文件都拷贝到dr.elephant的app-conf目录下。
3. 统一自己Hadoop集群的环境变量。

数据库问题

Database ‘default’ is in an inconsistent state!

启动失败并出现这个报错，一般是play框架的evolution问题，解决方法如下：

1. 停止dr.elephant并确保进程已kill
2. 删除原来的数据库并重新建库
3. 配置app-conf/elephant.conf中jvm_props="-Devolutionplugin=enabled -DapplyEvolutions.default=true"，开启evolution，使其能够自动初始化表结构。

你可以使用Dr. Elephant来分析你的作业（只需在搜索页贴入你的作业ID），就可以知道你的作业有哪些地方需要优化。

加速你的作业流

一般对于特定的作业，最好有自己的配置。大多数情况下，作业的默认配置无法提供最佳性能。尽管作业调优比较费劲，但一些简单的调整往往也能带来不错的效果。

需要特别注意的是mapper和reducer的数量，io和内存使用的配置，以及生成的文件数量。对这几个参数进行调整，让参数更适合当前的任务，可以极大的提升任务的执行性能。

Apache的官网中Hadoop Map/Reduce Tutorial这篇文章提供很多详细且有用的调试建议，有兴趣的可以仔细看看。

常规建议

逐步调优很重要

对于Pig作业来说，如果使用默认参数来设置reducer的数量，这对作业的性能可能是致命的。一般来说，对每个Pig作业，都花一些时间来调优参数PARALLEL是非常值得做的。例如：

1

memberFeaturesGrouped = GROUP memberFeatures BY memberId PARALLEL 90;