# Kafka
## 0. 目录
- Kafka Kafka基本概念及配置
Kafka 客户端实践及原理剖析
Kafka 内核及源码
Kafka 管理和监控
Kafka 流处理
## 1. Kafka基本概念及配置
- 官方QuickStart: https://kafka.apache.org/quickstart
- 官方文档: https://kafka.apache.org/documentation/
- 最新版本 3.2.0 【截止2022-7-20】
- 学习版本 kafka_2.11-2.4.0【前面是scala后面是kafka】 zookeeper-3.5.7【截止2022-7-20】
### 1.1 概念
- 概述
- Kafka 是分布式的流处理平台
- Kafka 是基于Zookeeper的分布式消息系统
- Kafka具有高吞吐率、高性能、实时、高可靠等特点
- Kafka 的组成概念
- Topic: 一个虚拟的概念, 由1到多个Partitions组成
- Partition: 实际消息存储单位
- Producer: 消息生产者
- Consumer: 消息消费者
-
### 1.2 安装Zookeeper(快捷)
```shell
# 修改配置文件
cp zoo_sample.cfg zoo.cfg
# 需要修改参数
dataDir=指定一个存储配置文件的目录
# 启动
./zkServer.sh start
# 连接测试
./zkCli.sh
```
### 1.3 安装Kafka
```shell
# 修改基本的配置文件
vim server.properties
# 需要修改的配置项
listeners=PLAINTEXT://[你的ip地址]:9092
advertised.listeners=PLAINTEXT://[你的ip地址]:9092
log.dirs=你的kafka的日志目录
zookeeper.connect=[你的ip地址]:2181
```
### 1.4 Kafka 的线上方案
- 操作系统
- 在 Linux 部署 Kafka 能够享受到零拷贝技术所带来的 快速数据传输特性。
- Windows 平台上部署 Kafka 只适合于个人测试或用于功能验证， 千万不要应用于生产环境。
- 磁盘
- **普通的机械磁盘** 即可, Kafka 大量使用磁盘不假，可它使用的方式多是顺序读写操作，一定程度上规避了机械磁盘 最大的劣势，即随机读写操作慢。
- 追求性价比的公司可以**不搭建 RAID**，使用**普通磁盘组成存储空间即可**。 [Kafka 1.1开始正式支持JBOD了。再说Kafka本身在软件层面也提供了冗余的机制来对 抗磁盘损坏。]
- 磁盘容量
- 在规划磁盘容量时你需要考虑下面这几个元素:
- 新增消息数
- 消息留存时间
- 平均消息大小
- 备份数
- 是否启用压缩
- 例子:
- 假设你所在公司有个业务每天需要向 Kafka 集群发送 **1 亿条消息** ，每条消息**保存两份**以防止数据丢失，另外消息默认**保存两周** 时间。现在假设消息的**平均大小是 1KB**?
- 每天 1 亿条 1KB 大小的消息，保存两份且留存两周的时间，那么总的空 间大小就等于 1 亿 * 1KB * 2 / 1000 / 1000 = 200GB。一般情况下 Kafka 集群除了消息 数据还有其他类型的数据，
- 比如索引数据等，故我们再为这些数据预留出 10% 的磁盘空 间，因此总的存储容量就是 220GB。既然要保存两周， ，
假设压缩比是 0.75，那么最后你需要规划的存储 空间就是 0.75 * 3 = 2.25TB。
- 带宽
- 对于 Kafka 这种通过网络大量进行数据传输的框架而言，**带宽特别容易成为瓶颈**。事实 上，在我接触的真实案例当中，带宽资源不足导致 Kafka 出现性能问题的比例至少占 60% 以上。
如果你的环境中还涉及跨机房传输，那么情况可能就更糟了。
- 带宽也主要有 两种:1Gbps 的千兆网络和 10Gbps 的万兆网络，特别是千兆网络应该是一般公司网络的标准配置了。
- 与其说是带宽资源的规划，其实真正要规划的是所需的 Kafka 服务器的数量。
- 假设你公司 的机房环境是千兆网络，即 1Gbps， ，
- 让我们来计算一下，由于带宽是 1Gbps，
通常情况下你只能假设 Kafka 会用到 70% 的带宽资源，因为总要为其 他应用或进程留一些资源。根据实际使用经验，超过 70% 的阈值就有网络丢包的可能性了，故 70% 的设定是一个比 较合理的值，
也就是说单台 Kafka 服务器最多也就能使用大约 700Mb 的带宽资源。稍等，这只是它能使用的最大带宽资源，你不能让 Kafka 服务器常规性使用这么多资源， 故通常要再额外预留出 2/3 的资源，
即单台服务器使用带宽 700Mb / 3 ≈ 240Mbps。需 要提示的是，这里的 2/3 其实是相当保守的，你可以结合你自己机器的使用情况酌情减少 此值。 好了，有了 240Mbps，
我们就可以计算 1 小时内处理 1TB 数据所需的服务器数量了。根 据这个目标，我们**每秒需要处理 2336Mb [1024*1024/3600*8] 的数据，除以 240，
如果 消**息还需要额外复制两份，那么总的服务器台数还要乘以 3，
- Kafka 性能测试脚本
- kafka-producer-perf-test 脚本还不错，
- 混部
- 一个机器上 部署一个zookeeper和一个kafka, 如果配置好**ZooKeeper事务日志**(比如设置好autopurge.purgeInterval及定期删除 snapshot等)， ，
---
- 最最最重要的集群参数配置
- Kafka 服务器端的配置，其中既有 Broker 端参数，也有主题(Topic)级别的参数、JVM 端参数和操作系统级别的参数
- Broker 端参数
- 首先 Broker 是需要配置存储信息的，即 Broker 使用哪些 磁盘。那么针对存储信息的重要参数有以下这么几个:
- **log.dirs**: 这是非常重要的参数，指定了 Broker 需要 使用的若干个文件目录路径。要知道这个参数是没有默认 值的，这说明什么?这说明它**必须由你亲自指定**。
- log.dir: 注意这是 dir， ，
- 这两个参数应该怎么设置呢?很简单，你只要设置 log.dirs， ， ，
具体格式是一个 CSV 格式，也就是用逗号分隔 的多个路径，比 如/home/kafka1,/home/kafka2,/home/kafka3这 样。如果有条件的话你最好保证这些目录挂载到不同的物理磁盘上。
- 这样做有两个好处:
- 提升读写性能:比起单块磁盘，多块物理磁盘同时读写数据有更高的吞吐量。
- 能够实现故障转移:即 Failover。这是 Kafka 1.1 版本新 引入的强大功能。要知道在以前，只要 Kafka Broker 使 用的任何一块磁盘挂掉了，整个 Broker 进程都会关闭。 但是自 1.1 开始，这种情况被修正了，
坏掉的磁盘上的数 据会自动地转移到其他正常的磁盘上，而且 Broker 还能 正常工作。这个改进正是我们舍弃 RAID 方案的基 础:没有这种 Failover 的话，我们只能依靠 RAID 来提 供保障。
- 与 ZooKeeper 相关的设置
- 首先 ZooKeeper 是 做什么的呢?它是一个分布式协调框架，负责协调管理并保 存 Kafka 集群的所有元数据信息， ，
- Kafka 与 ZooKeeper 相关的最重要的参数当属 **zookeeper.connect**
- 这也是一个 CSV 格式的参数，比 如我可以指定它的值为 **zk1:2181,zk2:2181,zk3:2181** 。2181 是 ZooKeeper 的默认端口。
- 如果我让**多个 Kafka 集群使用同一套 ZooKeeper 集群**，那么这个参数应该怎么设置呢?这时候 chroot 就派上用场了。这个 chroot 是 ZooKeeper 的概 念，类似于别名。
- 如果你有两套 Kafka 集群，假设分别叫它们 kafka1 和 kafka2，
- **切记 chroot 只需要写一次，而且是加到最后的**。我经常碰到有人这样指 定: zk1:2181/kafka1,zk2:2181/kafka2,zk3:2181/kafka3，
- 第三组参数是与 Broker 连接相关的，即客户端程序或其他 Broker 如何与该 Broker 进行通信的设置。有以下三个参 数:
- **listeners**: 学名叫监听器，其实就是告诉外部连接者要通过什么协议访问指定主机名和端口开放的 Kafka 服务。
- advertised.listeners: 和 listeners 相比多了个 advertised。Advertised 的含义表示宣称的、公布的， 就是说**这组监听器是 Broker 用于对外发布的**。
- [内网环境访问Kafka不需要配置这个参数， ， ，
- ~~host.name/port~~: 列出这两个参数就是想说你把它们忘掉吧，压根不要为它们指定值，毕竟都是过期的参数了。
- 监听器的概念，从构成上来说，它是若干个逗 号分隔的三元组，**每个三元组的格式为< 协议名称，主机 名，端口号>**。这里的协议名称可能是标准的名字，比如 PLAINTEXT 表示明文传输、SSL 表示使用 SSL 或 TLS 加密 传输等;
也可能是你自己定义的协议名字，比如 CONTROLLER: //localhost:9092。
- 一旦你自己定义了协议名称，你必须还要指定 listener.security.protocol.map参数告诉这个协议， ，
- 经常有人会问主机 名这个设置中我到底使用 IP 地址还是主机名。这里我给出 统一的建议:**最好全部使用主机名**，即 **Broker 端和 Client 端应用配置中全部填写主机名** 。 Broker 源代码中也使用的 是主机名，如果你在某些地方使用了 IP 地址进行连接，可能会发生无法连接的问题。
- 第四组参数是关于 Topic 管理的。我来讲讲下面这三个参数:
- auto.create.topics.enable:是否允许自动创建 Topic。
- [建议最好设置成 **false** ，即不允许自动创建 Topic。每个部门被分配的 Topic 应该由运维严格把控，决不能允许自行创建任何 Topic。]
- unclean.leader.election.enable:是否允许 Unclean Leader 选举。
- [每个分区都有多个副本来提供高可用。 在这些副本中只能有一个副本对外提供服务，即所谓的 Leader 副本。建议你还是显式地把它设置成 **false** 吧]
- auto.leader.rebalance.enable:是否允许定期进 行 Leader 选举。
- [对生产环境影响非常大，设置它的值为 true 表示允许 Kafka 定期地对一些 Topic 分区进行 Leader 重选举，当然这个重选举不是无脑进行的，它要满 足一定的条件才会发生。严格来说它与上一个参数中Leader 选举的最大不同在于，
它不是选 Leader， ， **false** 。]
- 最后一组参数是数据留存方面的，即:
- log.retention.{hour|minutes|ms}:这是个“三 兄弟”，都是控制一条消息数据被保存多长时间。从优先 级上来说 ms 设置最高、minutes 次之、hour 最低。
- [虽然 ms 设置有最高的优先级，但是 通常情况下我们还是设置 hour 级别的多一些，比如 log.retention.hour=168表示默认保存 7 天的数据，自动删除 7 天前的数据。很多公司把 Kafka 当做存储来使 用，那么这个值就要相应地调大。]
- log.retention.bytes:这是指定 Broker 为消息保存的总磁盘容量大小。
- [这个值默认是 -1，
- message.max.bytes:控制 Broker 能够接收的最大消息大小。
- [不能使用默认值的参数,默认的 1000012 太少了，还不到 1MB。实际场景中突 破 1MB 的消息都是屡见不鲜的，因此在线上环境中设置一 个比较大的值还是比较保险的做法。毕竟它只是一个标尺而 已，仅仅衡量 Broker 能够处理的最大消息大小，即使设置 大一点也不会耗费什么磁盘空间的。]
- 补充参数
- gg.handler.kafkahandler.Mode = tx gg.handler.kafkahandler.Mode = op 这两个的差别。遇到时 dml 数据会丢失的情况。用的是 op 。
- 当设置成op单个数据库表的变更(插入、更新、删除) 会被当成一条Kafka消息发送;当设置成tx时， ，
- Topic 级别参数
- Topic 级别参数会**覆盖全局 Broker 参数的值**，而每个 Topic 都能设置自己的参数值， 这就是所谓的 Topic 级别参数。
- **retention.ms**: 规定了该 Topic 消息被保存的时长。 默认是 7 天，即该 Topic 只保存最近 7 天的消息。一旦 设置了这个值，它会覆盖掉 Broker 端的全局参数值。
- **retention.bytes**:规定了要为该 Topic 预留多大的磁 盘空间。和全局参数作用相似，这个值通常在多租户的 Kafka 集群中会有用武之地。当前默认值是 -1，
- Topic 级别 参数的设置就是这种情况，我们有两种方式可以设置:
- 创建 Topic 时进行设置
- 修改 Topic 时设置
- 如何在创建 Topic 时设置这些参数
- 你的部门需要将交易数据发送到 Kafka 进行处理，需要保 存最近半年的交易数据，同时这些数据很大，通常都有几 MB，
- bin/kafka-topics.sh--bootstrap-serverlocalhost:9092--create--topictransaction--partitions1--replication-factor1--configretention.ms=15552000000--configmax.message.bytes=5242880
- 请注意结尾处的--config设置，
- 使用另一个自带的命令kafka-configs来修改 Topic 级别参数。假设我们现在要发送最大值是 10MB 的消息，该如何修改呢？命令如下：
- bin/kafka-configs.sh--zookeeperlocalhost:2181--entity-typetopics--entity-nametransaction--alter--add-configmax.message.bytes=10485760
- [你最好始终坚持使用第二种方式来设置， ，
- JVM 参数
- Kafka 服务器端代码是用 Scala 语言编写的，但终归还是编译成 Class 文件在 JVM 上运行，因此 JVM 参数设置对于 Kafka 集群的重要性不言而喻。
- 有条件的话至少使用 Java 8 吧, [Kafka 2.0已经不支持Java 7了， ， ，
- 堆大小这个参数至关重要。无脑给出一个通用的建议：将你的 JVM 堆大小设置成 **6GB** 吧，这是目前业界比较公认的一个合理值。我见过很多人就是使用默认的 Heap Size 来跑 Kafka，
毕竟 Kafka Broker 在与客户端进行交互时会在 JVM 堆上创建大量的 ByteBuffer 实例，
- [没有通用的标准， ： ， ，
- JVM 端配置的另一个重要参数就是垃圾回收器的设置，也就是平时常说的 GC 设置。如果你依然在使用 Java 7， ：
- 如果 Broker 所在机器的 CPU 资源非常充裕，建议使用 CMS 收集器。启用方法是指定-XX:+UseCurrentMarkSweepGC。
- 否则，使用吞吐量收集器。开启方法是指定-XX:+UseParallelGC。
- 当然了，如果你已经在使用 Java 8 了[G1是jdk9中默认的， ， ， ，
- Java8默认的新生代垃圾回收器是： ， ，
- 确定好了要设置的 JVM 参数，我们该如何为 Kafka 进行设置呢？
- 只需要设置下面这两个环境变量即可：
- KAFKA_HEAP_OPTS：
- KAFKA_JVM_PERFORMANCE_OPTS：
- 比如你可以这样启动 Kafka Broker，
- $> **export KAFKA_HEAP_OPTS=--Xms6g --Xmx6g**
- $> **export KAFKA_JVM_PERFORMANCE_OPTS= -server -XX:+UseG1GC -XX:MaxGCPauseMillis=20 -XX:InitiatingHeapOccupancyPercent=35 -XX:+ExplicitGCInvokesConcurrent -Djava.awt.headless=true**
- $> **bin/kafka-server-start.sh config/server.properties**
- 操作系统参数
- Kafka 并不需要设置太多的 OS 参数，但有些因素最好还是关注一下，比如下面这几个：
- 文件描述符限制
- 文件系统类型
- Swappiness
- 提交时间
- 首先是 **ulimit -n**
- 觉得任何一个 Java 项目最好都调整下这个值。实际上，文件描述符系统资源并不像我们想象的那样昂贵，你不用太担心调大此值会有什么不利的影响。通常情况下将它设置成一个超大的值是合理的做法，比如 **ulimit -n 1000000** 。
- 但不设置的话后果很严重，
- 其次是文件系统类型的选择。
- 这里所说的文件系统指的是如 ext3、ext4 或 XFS 这样的日志型文件系统。根据官网的测试报告，**XFS** 的性能要强于 ext4， ， **ZFS** 的数据报告，貌似性能更加强劲，有条件的话不妨一试。
- 第三是 swap 的调优
- 网上很多文章都提到设置其为 0， ， ，
它会随机挑选一个进程然后 kill 掉，即根本不给用户任何的预警。但如果设置成一个比较小的值，当开始使用 swap 空间时，你至少能够观测到 Broker 性能开始出现急剧下降，从而给你进一步调优和诊断问题的时间。基于这个考虑，
我**个人建议将 swappniess 配置成一个接近 0 但不为 0 的值，比如 1**。
### 1.5 kafka的基本操作
```shell
# 1、启动Kafka - todo 这个命令会让控制台有日志生成不太方便
bin/kafka-server-start.sh config/server.properties &
# 2、停止Kafka
bin/kafka-server-stop.sh
# 3、创建Topic
bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic jiangzh-topic
# 4、查看已经创建的Topic信息
bin/kafka-topics.sh --list --zookeeper localhost:2181
# 5、发送消息
bin/kafka-console-producer.sh --broker-list 192.168.220.128:9092 --topic jiangzh-topic
# 6、接收消息
bin/kafka-console-consumer.sh --bootstrap-server 192.168.220.128:9092 --topic jiangzh-topic --from-beginning
# {"orderId":"002","price":"80"}
```
```shell
# 【旧版】
kafka-topics.sh --zookeeper 172.16.26.183:2181 --list
# 【新版】
kafka-topics.sh --bootstrap-server 172.16.26.183:9092 --list
# （ –
# 创建topic主题
kafka-topics.sh --bootstrap-server 172.16.26.183:9092 --create --topic topic1 --partitions 1 --replication-factor 3
# --create 命令后 --topic 为创建topic 并指定 topic name
# --partitions 为指定分区数量
# --replication-factor 为指定副本集数量
# 向kafka集群发送数据
# 【无key型消息】
kafka-console-producer.sh --bootstrap-server 172.16.26.183:9092 --topic topic1
# 【有key型消息】
kafka-console-producer.sh --bootstrap-server 172.16.26.183:9092 --topic topic1 --property parse.key=true
# （ ， ）
# kafka命令接受数据
kafka-console-consumer.sh --bootstrap-server 172.16.26.183:9092 --topic topic1 --from-beginning
# kafka查看消费进度： （ ， ）
kafka-consumer-groups.sh --bootstrap-server 172.16.26.183:9092 --describe --group group1
```
## 2. 客户端实践及原理剖析
### 2.1 生产者消息分区机制原理剖析
- 为什么分区？
- Kafka 的消息组织方式实际上是三级结构：主题 - 分区 - 消息
- 主题下的每条消息只会保存在某一个分区中，而不会在多个分区中被保存多份。
- 为什么使用分区的概念而不是直接使用多个主题呢？
- 分区的作用就是提供负载均衡的能力，或者说对数据进行分区的主要原因，就是为了**实现系统的高伸缩性（ ）
- 不同的分区能够被放置到不同节点的机器上，而数据的读写操作也都是针对分区这个粒度而进行的，这样每个节点的机器都能**独立地执行各自分区的读写请求处理**。
- 并且，我们还可以通过**添加新的节点机器来增加整体系统的吞吐量**。
- 实际上分区的概念以及分区数据库早在 1980 年就已经有大牛们在做了，比如那时候有个叫 Teradata 的数据库就引入了**分区的概念**。
- Kafka 中叫分区，在 MongoDB 和 Elasticsearch 中就叫分片 Shard， ，
- 除了提供负载均衡这种最核心的功能之外，利用分区也可以实现其他一些业务级别的需求，比如**实现业务级别的消息顺序的问题**
- 都有哪些分区策略？
- 所谓分区策略是决定生产者将消息发送到哪个分区的算法。
- 如果要自定义分区策略，
- 在编写生产者程序时，
- 这个接口也很简单， ：
- 方法签名：
- 这里的topic、key、keyBytes、value和valueBytes都属于消息数据， （
- Kafka 给你这么多信息，就是希望让你能够充分地利用这些信息对消息进行分区，计算出它要被发送到哪个分区中。
- 只要你自己的**实现类定义好了 partition 方法，
- 比较常见的分区策略
- **轮询策略**
- 也称 Round-robin 策略，即顺序分配。比如一个主题下有 3 个分区，那么第一条消息被发送到分区 0， ， ， ，
- 
- 这就是所谓的轮询策略。轮询策略是 Kafka Java 生产者 API ** 默认提供的分区策略**。如果你未指定partitioner.class参数，
- [轮询策略有非常优秀的负载均衡表现，它总是能保证消息最大限度地被平均分配到所有分区上，故默认情况下它是最合理的分区策略，也是我们最常用的分区策略之一。]
- **随机策略**
- 也称 Randomness 策略。所谓随机就是我们随意地将消息放置到任意一个分区上，如下面这张图所示。
- 
- 如果要实现随机策略版的 partition 方法，很简单，只需要两行代码即可：
- List< PartitionInfo > partitions = cluster.partitionsForTopic(topic);
- return ThreadLocalRandom.current().nextInt(partitions.size());
- 先计算出该主题总的分区数，然后随机地返回一个小于它的正整数。
- 本质上看随机策略也是力求将数据均匀地打散到各个分区，但从实际表现来看，它要逊于轮询策略，所以**如果追求数据的均匀分布，还是使用轮询策略比较好**。事实上，随机策略是老版本生产者使用的分区策略，在新版本中已经改为轮询了。
- **按消息键保序策略**
- 也称 Key-ordering 策略。这个名词是我自己编的，
- Kafka 允许为每条消息定义消息键，简称为 Key。这个 Key 的作用非常大，它可以是一个有着明确业务含义的字符串，比如客户代码、部门编号或是业务 ID 等；也可以用来表征消息元数据。
特别是在 Kafka 不支持时间戳的年代，在一些场景中，工程师们都是直接将消息创建时间封装进 Key 里面的。一旦消息被定义了 Key，
** 由于每个分区下的消息处理都是有顺序的，故这个策略被称为按消息键保序策略**，如下图所示。
- 
- 实现这个策略的 partition 方法同样简单，只需要下面两行代码即可：
- List< PartitionInfo > partitions = cluster.partitionsForTopic(topic);
- return Math.abs(key.hashCode()) % partitions.size();
- 前面提到的 Kafka 默认分区策略实际上同时实现了两种策略：**如果指定了 Key， ； ，
- 如何实现消息的顺序问题?
- 企业发送的 Kafka 的消息是有因果关系的，故处理因果关系也必须要保证有序性，否则先处理了“果”后处理“因”必然造成业务上的混乱。
- 当时企业的做法是给 Kafka 主题设置单分区，也就是 1 个分区。这样所有的消息都只在这一个分区内读写，因此保证了全局的顺序性。
这样做虽然实现了因果关系的顺序性，但也丧失了 Kafka 多分区带来的高吞吐量和负载均衡的优势。
- 后来经过了解和调研，我发现这种具有因果关系的消息都有一定的特点，比如在**消息体中都封装了固定的标志位**，改进建议他们**对此标志位设定专门的分区策略，保证同一标志位的所有消息都发送到同一分区**，
这样既可以保证分区内的消息顺序，也可以享受到多分区带来的性能红利。
- 这种基于个别字段的分区策略本质上就是按消息键保序的思想，其实更加合适的做法是把**标志位数据提取出来统一放到 Key 中，这样更加符合 Kafka 的设计思想**。
- 其他分区策略
- 其实还有一种比较常见的，即所谓的**基于地理位置的分区策略**。当然这种策略一般只针对那些大规模的 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-client生产者默认分区源码时， ， ， ？
- 分区没有leader的时候就是unavailable了。某些操作会导致瞬间没有leader，
- RocketMQ与Kafka的主要区别
- Kafka吞吐量大， ，
- RocketMQ毕竟是阿里出品， ；
- Kafka现在主要发力Streaming， ，
- 其他方面这两者确实都差不多
- 广州机房怎么消费广州partition的数据，
- 使用这个方法：
### 2.2 生产者压缩算法面面观
- 说起压缩（ ） ，
- 怎么压缩？
- Kafka 是如何压缩消息的呢？要弄清楚这个问题，就要从 Kafka 的消息格式说起了。目前 **Kafka 共有两大类消息格式，社区分别称之为 V1 版本和 V2 版本** 。V2 版本是 Kafka 0.11.0.0 中正式引入的。
- 不论是哪个版本， （ ） （ ） （ ） ，
Kafka 底层的消息日志由一系列消息集合日志项组成。Kafka ** 通常不会直接操作具体的一条条消息，它总是在消息集合这个层面上进行写入操作**。
- V2 版本主要是针对 V1 版本的一些弊端做了修正
- 一个，就是把消息的公共部分抽取出来放到外层消息集合里面，这样就不用每条消息都保存这些信息了。
- 在 V2 版本中，消息的 CRC 校验工作就被移到了消息集合这一层。
- V2 版本还有一个和压缩息息相关的改进，就是保存压缩消息的方法发生了变化。
- 之前 V1 版本中保存压缩消息的方法是把多条消息进行压缩然后保存到外层消息的消息体字段中；而 V2 版本的做法是对整个消息集合进行压缩。显然后者应该比前者有更好的压缩效果。
- V2 版本都比 V1 版本节省磁盘空间。
- 何时压缩？
- 在 Kafka 中，压缩可能发生在两个地方：生产者端和 Broker 端。
- 生产者程序中配置 compression.type 参数即表示启用指定类型的压缩算法。
- 下面这段程序代码展示了如何构建一个开启 GZIP 的 Producer 对象：
```java
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("acks", "all");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
// 开启 GZIP 压缩
props.put("compression.type", "gzip");
Producer< String , String > producer = new KafkaProducer< >(props);
```
- 这里比较关键的代码行是 props.put(“compression.type”, “gzip”)，它表明该 Producer 的压缩算法使用的是 GZIP。这样 Producer 启动后生产的每个消息集合都是经 GZIP 压缩过的，故而能很好地节省网络传输带宽以及 Kafka Broker 端的磁盘占用。
- 在生产者端启用压缩是很自然的想法，那为什么我说在 Broker 端也可能进行压缩呢？其实大部分情况下 **Broker 从 Producer 端接收到消息后仅仅是原封不动地保存而不会对其进行任何修改** ，但这里的“大部分情况”也是要满足一定条件的。
- 有两种例外情况就可能让 Broker 重新压缩消息。
- 情况一：
- Producer 说：“我要使用 GZIP 进行压缩。”
- Broker 说：“不好意思，我这边接收的消息必须使用 Snappy 算法进行压缩。”
- Kafka 官网，你会发现 Broker 端也有一个参数叫 compression.type，
可一旦你在 Broker 端设置了不同的 compression.type 值，就一定要小心了，因为可能会发生预料之外的压缩 / 解压缩操作，**通常表现为 Broker 端 CPU 使用率飙升**。
- 情况二：
- 在一个生产环境中， ，
它还让 Kafka 丧失了引以为豪的 Zero Copy 特性。
- 所以尽量保证消息格式的统一吧，这样不仅可以避免不必要的解压缩 / 重新压缩，对提升其他方面的性能也大有裨益。如果有兴趣你可以深入地了解下 Zero Copy 的原理。
- 何时解压缩？
- 通常来说解压缩发生在消费者程序中，也就是说 Producer 发送压缩消息到 Broker 后， ，
- Consumer 怎么知道这些消息是用何种压缩算法压缩的呢？其实答案就在消息中。**Kafka 会将启用了哪种压缩算法封装进消息集合中**，这样当 Consumer 读取到消息集合时，它自然就知道了这些消息使用的是哪种压缩算法。
- 住这句话：**Producer 端压缩、Broker 端保持、Consumer 端解压缩。**
- 除了在 Consumer 端解压缩，
我们必须承认**这种解压缩对 Broker 端性能是有一定影响的，特别是对 CPU 的使用率而言**。
- 国内京东的小伙伴们刚刚向社区提出了一个 bugfix， ， ，
- 目前社区并未采纳这个建议，原因就是这种消息校验是非常重要的，不可盲目去之。毕竟先把事情做对是最重要的，**在做对的基础上，再考虑把事情做好做快**。
- 你也可以思考一下，是否有一个两全其美的方案，既能避免消息解压缩也能对消息执行校验。
- 各种压缩算法对比
- 在 Kafka 2.1.0 版本之前， ， **Zstandard** 算法（简写为 zstd） ， （ ）
- 对了，看一个压缩算法的优劣，有两个重要的指标：
- 一个指标是**压缩比**，原先占 100 份空间的东西经压缩之后变成了占 20 份空间，那么压缩比就是 5， ；
- 另一个指标就是**压缩 / 解压缩吞吐量**，比如每秒能压缩或解压缩多少 MB 的数据。同样地，吞吐量也是越高越好。
- Facebook Zstandard 官网提供的一份压缩算法 benchmark 比较结果：
- 
- zstd 算法有着最高的压缩比，而在吞吐量上的表现只能说中规中矩。
- 反观 LZ4 算法，它在吞吐量方面则是毫无疑问的执牛耳者。
- 在实际使用中，
- 但对于 Kafka 而言，它们的性能测试结果却出奇得一致，
- 即**在吞吐量方面： ；
- 而**在压缩比方面，
- 具体到**物理资源**，使用 **Snappy 算法占用的网络带宽最多， ，这是合理的，毕竟 zstd 就是要提供超高的压缩比；
- 在 **CPU 使用率** 方面，各个算法表现得差不多，**只是在压缩时 Snappy 算法使用的 CPU 较多一些**
- 而**在解压缩时 GZIP 算法则可能使用更多的 CPU**。
- 最佳实践
- 首先来说压缩。何时启用压缩是比较合适的时机呢？
- 你现在已经知道 Producer 端完成的压缩，那么启用压缩的一个条件就是 **Producer 程序运行机器上的 CPU 资源要很充足** 。如果 Producer 运行机器本身 CPU 已经消耗殆尽了，那么启用消息压缩无疑是雪上加霜，只会适得其反。
- 除了 CPU 资源充足这一条件，如果你的环境中带宽资源有限，那么我也建议你开启压缩。
- 事实上我见过的很多 Kafka 生产环境都遭遇过带宽被打满的情况。这年头，**带宽可是比 CPU 和内存还要珍贵的稀缺资源**，毕竟万兆网络还不是普通公司的标配，因此千兆网络中 Kafka 集群带宽资源耗尽这件事情就特别容易出现。
- **如果你的客户端机器 CPU 资源有很多富余，我强烈建议你开启 zstd 压缩，这样能极大地节省网络资源消耗**。
- 其次说说解压缩。其实也没什么可说的。
- 一旦启用压缩，解压缩是不可避免的事情。
- 这里只想强调一点：我们对不可抗拒的解压缩无能为力，但至少能规避掉那些意料之外的解压缩。就像我前面说的，因为要兼容老版本而引入的解压缩操作就属于这类。有条件的话尽量保证不要出现消息格式转换的情况。
- 补充问题
- 正常情况下broker端会原样保存起来， ， ？
- 它只是解压缩读取而已，不会将解压缩之后的数据回写到磁盘。另外就像我置顶的留言那样，目前社区已经接纳了京东小伙伴的修改，貌似可以绕过这部分解压缩了.
- 规避了broker端为执行校验而做的解压缩操作，
- https://issues.apache.org/jira/browse/KAFKA-8106
- 消息层次、消息集合、消息、日志项这些概念与它们之间的关系
- 消息批次RecordBatch里面包含若干条消息（
Producer以recordbatch为单位发送消息， ， ； ，
### 2.3 无消息丢失配置怎么实现？
- 一句话概括， （ ）
- 第一个核心要素是 ** “已提交的消息”**。什么是已提交的消息？当 Kafka 的若干个 Broker 成功地接收到一条消息并写入到日志文件后，它们会告诉生产者程序这条消息已成功提交。此时，这条消息在 Kafka 看来就正式变为“已提交”消息了。
- 那为什么是若干个 Broker 呢？这取决于你对“已提交”的定义。你可以选择只要有一个 Broker 成功保存该消息就算是已提交，也可以是令所有 Broker 都成功保存该消息才算是已提交。不论哪种情况，
- 第二个核心要素就是 ** “有限度的持久化保证”**
- 假如你的消息保存在 N 个 Kafka Broker 上，那么这个前提条件就是这 N 个 Broker 中**至少有 1 个存活**。只要这个条件成立，
- “消息丢失”案例
- 案例 1：
- 目前 Kafka Producer 是异步发送消息的，也就是说如果你调用的是 producer.send(msg) 这个 API， ，
- 这种发送方式有个有趣的名字， ， ，
- 如果用这个方式，可能会有哪些因素导致消息没有发送成功呢？
- 其实原因有很多，例如**网络抖动，导致消息压根就没有发送到 Broker 端**；或者**消息本身不合格导致 Broker 拒绝接收**（比如消息太大了，超过了 Broker 的承受能力）
- 就像前面说过的，
- 解决此问题的方法非常简单：
- Producer ** 永远要使用带有回调通知的发送 API**，也就是说不要使用 producer.send(msg)，而要使用 producer.send(msg, callback)。
- （回调），它能准确地告诉你消息是否真的提交成功了。一旦出现消息提交失败的情况，你就可以有针对性地进行处理。
- 如果是因为那些瞬时错误，那么仅仅让 Producer 重试就可以了；如果是消息不合格造成的，那么可以调整消息格式后再次发送。
- 总之，**处理发送失败的责任在 Producer 端而非 Broker 端。**
- 案例 2：
- Consumer 端丢失数据主要体现在 Consumer 端要消费的消息不见了。Consumer 程序有个“位移”的概念，表示的是这个 Consumer 当前消费到的 Topic 分区的位置。
- 
- 比如对于 Consumer A 而言，它当前的位移值就是 9；
- 这里的“位移”类似于我们看书时使用的书签，它会标记我们当前阅读了多少页，下次翻书的时候我们能直接跳到书签页继续阅读。
- 正确使用书签有两个步骤：第一步是读书，第二步是更新书签页。**不能颠倒**
- 要对抗这种消息丢失，办法很简单：**维持先消费消息（阅读），再更新位移（书签）的顺序即可**。这样就能最大限度地保证消息不丢失。
- 当然，**这种处理方式可能带来的问题是消息的重复处理**，类似于同一页书被读了很多遍，但这不属于消息丢失的情形。
- 还存在一种比较隐蔽的消息丢失场景。
- 对于 Kafka 而言， Consumer 程序从 Kafka 获取到消息后开启了多个线程异步处理消息，而 Consumer 程序自动地向前更新位移。
假如其中某个线程运行失败了，它负责的消息没有被成功处理，但位移已经被更新了，因此这条消息对于 Consumer 而言实际上是丢失了。
- 这里的**关键在于 Consumer 自动提交位移**
- 这个问题的解决方案也很简单：
- 如果是多线程异步处理消费消息，
- 单个 Consumer 程序使用多线程来消费消息说起来容易，写成代码却异常困难，因为你很难正确地处理位移的更新，也就是说避免无消费消息丢失很简单，但极易出现消息被消费了多次的情况。
- 最佳实践
- 不要使用 producer.send(msg)，而要使用 producer.send(msg, callback)。记住，**一定要使用带有回调通知的 send 方法**。
- **设置 acks = all**。acks 是 Producer 的一个参数，代表了你对“已提交”消息的定义。如果设置成 all，
- **设置 retries 为一个较大的值**。这里的 retries 同样是 Producer 的参数，对应前面提到的 Producer 自动重试。当出现网络的瞬时抖动时，消息发送可能会失败，此时配置了 retries > 0 的 Producer 能够自动重试消息发送，避免消息丢失。
- **设置 unclean.leader.election.enable = false**。这是 Broker 端的参数，它控制的是哪些 Broker 有资格竞选分区的 Leader。如果一个 Broker 落后原先的 Leader 太多，那么它一旦成为新的 Leader，
故一般都要将该参数设置成 false，
- **设置 replication.factor >= 3**。这也是 Broker 端的参数。其实这里想表述的是，最好将消息多保存几份，毕竟目前防止消息丢失的主要机制就是冗余。
- **设置 min.insync.replicas > 1**。这依然是 Broker 端参数，控制的是消息至少要被写入到多少个副本才算是“已提交”。设置成大于 1 可以提升消息持久性。在实际环境中千万不要使用默认值 1。
- 确保 replication.factor > min.insync.replicas。如果两者相等， ， ， ，
** 推荐设置成 replication.factor = min.insync.replicas + 1**。
- 确保消息消费完成再提交。Consumer 端有个参数 **enable.auto.commit， ，并采用手动提交位移的方式。就像前面说的，这对于单 Consumer 多线程处理的场景而言是至关重要的。
- 补充
- 设置 acks = all。表明所有副本 Broker 都要接收到消息， ， ？
- 碰到的实际场景， ， ，
### 2.4 客户端都有哪些不常见但是很高级的功能？
- 什么是拦截器？
- 基本思想就是允许应用程序在**不修改逻辑的情况下**，
- Kafka 拦截器
- Kafka ** 拦截器分为生产者拦截器和消费者拦截器**。
- 生产者拦截器允许你在发送消息前以及消息提交成功后植入你的拦截器逻辑；而消费者拦截器支持在消费消息前以及提交位移后编写特定逻辑。
值得一提的是，这两种拦截器都支持链的方式，即你可以将一组拦截器串连成一个大的拦截器，**Kafka 会按照添加顺序依次执行拦截器逻辑**。
- Kafka 拦截器的设置方法是通过参数配置完成的。生产者和消费者两端有一个相同的参数，名字叫 **interceptor.classes** ，它指定的是一组类的列表，每个类就是特定逻辑的拦截器实现类。
- 假设第一个拦截器的完整类路径是 com.yourcompany.kafkaproject.interceptors.AddTimeStampInterceptor， ，
那么你需要按照以下方法在 Producer 端指定拦截器：
```java
Properties props = new Properties();
List< String > interceptors = new ArrayList< >();
interceptors.add("com.yourcompany.kafkaproject.interceptors.AddTimestampInterceptor"); // 拦截器 1
interceptors.add("com.yourcompany.kafkaproject.interceptors.UpdateCounterInterceptor"); // 拦截器 2
props.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, interceptors);
```
- 我们应该怎么编写 AddTimeStampInterceptor 和 UpdateCounterInterceptor 类呢？
- 这两个类以及你自己编写的所有 Producer 端拦截器实现类都**要继承 org.apache.kafka.clients.producer.ProducerInterceptor 接口**。该接口是 Kafka 提供的，里面有两个核心的方法。
- onSend：
- onAcknowledgement：
值得注意的是，**这个方法和 onSend 不是在同一个线程中被调用的**，因此如果你在这两个方法中调用了某个共享可变对象，**一定要保证线程安全哦**。
还有一点很重要，这个方法处在 Producer 发送的主路径中，所以**最好别放一些太重的逻辑进去**，否则你会发现你的 Producer TPS 直线下降。
- 同理，指定消费者拦截器也是同样的方法，只是具体的实现类要**实现 org.apache.kafka.clients.consumer.ConsumerInterceptor 接口**，这里面也有两个核心方法。
- onConsume：
- onCommit：
- 一定要注意的是，指定拦截器类时要指定它们的**全限定名**，即 full qualified name。通俗点说就是要把完整包名也加上， ，
- 典型使用场景
- 其实，跟很多拦截器的用法相同，**Kafka 拦截器可以应用于包括客户端监控、端到端系统性能检测、消息审计**等多种功能在内的场景。
- Kafka 默认提供的监控指标都是针对单个客户端或 Broker 的，你很难从具体的消息维度去**追踪集群间消息的流转路径**。同时，如何监控一条消息从生产到最后消费的端到端延时也是很多 Kafka 用户迫切需要解决的问题。
- 通过实现拦截器的逻辑以及可插拔的机制，我们能够快速地观测、验证以及监控集群间的客户端性能指标，特别是能够**从具体的消息层面上去收集这些数据**。这就是 Kafka 拦截器的一个非常典型的使用场景。
- 我们再来看看消息审计（ ）
- 作为私有云的 PaaS 提供方，你肯定要能够随时查看每条消息是哪个业务方在什么时间发布的，之后又被哪些业务方在什么时刻消费。
一个可行的做法就是你编写一个拦截器类，实现相应的消息审计逻辑，然后强行规定所有接入你的 Kafka 服务的客户端程序必须设置该拦截器。
- 案例分享
- 通过编写拦截器类来统计消息端到端处理的延时，非常实用，我建议你可以直接移植到你自己的生产环境中。
- 某个业务只有一个 Producer 和一个 Consumer， ，
- 既然是要计算总延时，**那么一定要有个公共的地方来保存它**，并且这个公共的地方还是要让生产者和消费者程序都能访问的。在这个例子中，我们假设数据被保存在 Redis 中。
- 实现生产者拦截器
```java
public class AvgLatencyProducerInterceptor implements ProducerInterceptor< String , String > {
private Jedis jedis; // 省略 Jedis 初始化
@Override
public ProducerRecord< String , String > onSend(ProducerRecord< String , String > record) {
jedis.incr("totalSentMessage");
return record;
}
@Override
public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
}
@Override
public void close() {
}
@Override
public void configure(Map< java.lang.String , ? > configs) {
}
```
- 上面的代码比较关键的是在发送消息前更新总的已发送消息数。为了节省时间，我没有考虑发送失败的情况，因为发送失败可能导致总发送数不准确。不过好在处理思路是相同的，你可以有针对性地调整下代码逻辑。
- 消费者端的拦截器实现，代码如下：
```java
public class AvgLatencyConsumerInterceptor implements ConsumerInterceptor< String , String > {
private Jedis jedis; // 省略 Jedis 初始化
@Override
public ConsumerRecords< String , String > onConsume(ConsumerRecords< String , String > records) {
long lantency = 0L;
for (ConsumerRecord< String , String > record : records) {
lantency += (System.currentTimeMillis() - record.timestamp());
}
jedis.incrBy("totalLatency", lantency);
long totalLatency = Long.parseLong(jedis.get("totalLatency"));
long totalSentMsgs = Long.parseLong(jedis.get("totalSentMessage"));
jedis.set("avgLatency", String.valueOf(totalLatency / totalSentMsgs));
return records;
}
@Override
public void onCommit(Map< TopicPartition , OffsetAndMetadata > offsets) {
}
@Override
public void close() {
}
@Override
public void configure(Map< String , ? > configs) {
```
- 在上面的消费者拦截器中，我们在真正消费一批消息前首先更新了它们的总延时，方法就是用**当前的时钟时间减去封装在消息中的创建时间，然后累计得到这批消息总的端到端处理延时并更新到 Redis 中**。
之后的逻辑就很简单了，我们分别从 Redis 中读取更新过的总延时和总消息数，两者相除即得到端到端消息的平均处理延时。
- 创建好生产者和消费者拦截器后，我们按照上面指定的方法分别将它们配置到各自的 Producer 和 Consumer 程序中，这样就能计算消息从 Producer 端到 Consumer 端平均的处理延时了。
这种端到端的指标监控能够从全局角度俯察和审视业务运行情况，及时查看业务是否满足端到端的 SLA 目标。
### 2.5 Java生产者是如何管理TCP连接的？
- 为何采用 TCP？
- 从社区的角度来看，在开发客户端时，人们能够利用 TCP 本身提供的一些高级功能，比如**多路复用请求以及同时轮询多个连接**的能力。
- 所谓的多路复用请求，即 multiplexing request，
其实严格来说，**TCP 并不能多路复用，它只是提供可靠的消息交付语义保证**，比如自动重传丢失的报文。
- 作为一个基于报文的协议， ）
- 除了 TCP 提供的这些高级功能有可能被 Kafka 客户端的开发人员使用之外，社区还发现，**目前已知的 HTTP 库在很多编程语言中都略显简陋**。
- 基于这两个原因，
- Kafka 生产者程序概览
- Kafka 的 Java 生产者 API 主要的对象就是 KafkaProducer。通常我们开发一个生产者的步骤有 4 步。
- 第 1 步：构造生产者对象所需的参数对象。
- 第 2 步：利用第 1 步的参数对象，创建 KafkaProducer 对象实例。
- 第 3 步：使用 KafkaProducer 的 send 方法发送消息。
- 第 4 步：调用 KafkaProducer 的 close 方法关闭生产者并释放各种系统资源。
- 代码示例:
```java
Properties props = new Properties ();
props.put("参数 1", "参数 1 的值")；
props.put("参数 2", "参数 2 的值")；
//
try (Producer< String , String > producer = new KafkaProducer< >(props)) {
producer.send(new ProducerRecord< String , String > (……), callback);
//
}
```
- 这段代码使用了 Java 7 提供的 try-with-resource 特性，所以并没有显式调用 producer.close() 方法。无论是否显式调用 close 方法，所有生产者程序大致都是这个路数。
- 当我们开发一个 Producer 应用时，生产者会向 Kafka 集群中指定的主题（ ） ， ，
- 何时创建 TCP 连接？
- 生产者代码是什么时候创建 TCP 连接的。就上面的那段代码而言，可能创建 TCP 连接的地方有两处：
- **在创建 KafkaProducer 实例时，生产者应用会在后台创建并启动一个名为 Sender 的线程，该 Sender 线程开始运行时首先会创建与 Broker 的连接。**
- 测试环境中的日志来说明这一点：
```txt
[2018-12-09 09:35:45,620] DEBUG [Producer clientId=producer-1] Initialize connection to node localhost:9093 (id: -2 rack: null) for sending metadata request (org.apache.kafka.clients.NetworkClient:1084)
[2018-12-09 09:35:45,622] DEBUG [Producer clientId=producer-1] Initiating connection to node localhost:9093 (id: -2 rack: null) using address localhost/127.0.0.1 (org.apache.kafka.clients.NetworkClient:914)
[2018-12-09 09:35:45,814] DEBUG [Producer clientId=producer-1] Initialize connection to node localhost:9092 (id: -1 rack: null) for sending metadata request (org.apache.kafka.clients.NetworkClient:1084)
[2018-12-09 09:35:45,815] DEBUG [Producer clientId=producer-1] Initiating connection to node localhost:9092 (id: -1 rack: null) using address localhost/127.0.0.1 (org.apache.kafka.clients.NetworkClient:914)
[2018-12-09 09:35:45,828] DEBUG [Producer clientId=producer-1] Sending metadata request (type=MetadataRequest, topics=) to node localhost:9093 (id: -2 rack: null) (org.apache.kafka.clients.NetworkClient:1068)
```
- 如果不调用 send 方法，这个 Producer 都不知道给哪个主题发消息，它又怎么能知道连接哪个 Broker 呢？难不成它会连接 bootstrap.servers 参数指定的所有 Broker 吗？
- 解释一下 bootstrap.servers 参数。它是 Producer 的核心参数之一，指定了这个 Producer 启动时要连接的 Broker 地址。
- 这里的“启动时”，代表的是 Producer 启动时会发起与这些 Broker 的连接。因此，如果你为这个参数指定了 1000 个 Broker 连接信息，那么很遗憾，你的 **Producer 启动时会首先创建与这 1000 个 Broker 的 TCP 连接** 。
- 在实际使用过程中，我并**不建议把集群中所有的 Broker 信息都配置到 bootstrap.servers 中**，通常你指定 **3～ 了。因为 **Producer 一旦连接到集群中的任一台 Broker， 。
- 在 KafkaProducer 实例被创建后以及消息被发送前，
- 你不需要了解 RecordAccumulator 类是做什么的，你只要知道它主要的数据结构是一个 ConcurrentMap< TopicPartition , Deque > 。TopicPartition 是 Kafka 用来表示主题分区的 Java 对象，本身是不可变对象。而 RecordAccumulator 代码中用到 Deque 的地方都有锁的保护，所以基本上可以认定 RecordAccumulator 类是线程安全的。
- **TCP 连接是在创建 KafkaProducer 实例时建立的。**
- 它只会在这个时候被创建吗？
- 当然不是！
- 当 Producer 更新了集群的元数据信息之后，如果发现与某些 Broker 当前没有连接，那么它就会创建一个 TCP 连接。同样地， ，
- Producer 更新集群元数据信息的两个场景
- 场景一：当 Producer 尝试给一个不存在的主题发送消息时，
- 场景二： ， ， ，
- 何时关闭 TCP 连接？
- Producer 端关闭 TCP 连接的方式有两种：**一种是用户主动关闭；一种是 Kafka 自动关闭**。
- 第一种。这里的主动关闭实际上是广义的主动关闭，甚至包括用户调用 kill -9 主动“杀掉”Producer 应用。当然最推荐的方式还是调用 producer.close() 方法来关闭。
- 第二种是 Kafka 帮你关闭，这与 Producer 端参数 connections.max.idle.ms 的值有关。默认情况下该参数值是 9 分钟，即如果在 9 分钟内没有任何请求“流过”某个 TCP 连接，那么 Kafka 会主动帮你把该 TCP 连接关闭。用户可以在 Producer 端设置 connections.max.idle.ms=-1 禁掉这种机制。一旦被设置成 -1，
当然这只是软件层面的“长连接”机制，由于 Kafka 创建的这些 Socket 连接都开启了 keepalive，
- 在第二种方式中，
### 2.6 幂等生产者和事务生产者是一回事吗？
- Kafka 消息交付可靠性保障以及精确处理一次语义的实现。
- 所谓的消息交付可靠性保障，**是指 Kafka 对 Producer 和 Consumer 要处理的消息提供什么样的承诺**。常见的承诺有以下三种：
- 最多一次（ ） ： ，
- 至少一次（ ） ： ，
- 精确一次（ ） ： ，
- 目前， ** 默认提供的交付可靠性保障是第二种**，即至少一次。
- Kafka 也可以提供最多一次交付保障，**只需要让 Producer 禁止重试即可**。这样一来，消息要么写入成功，要么写入失败，但绝不会重复发送。我们通常不会希望出现消息丢失的情况，但一些场景里偶发的消息丢失其实是被允许的，相反，消息重复是绝对要避免的。此时，使用最多一次交付保障就是最恰当的。
- 那么问题来了， （ ） （ ）
- 什么是幂等性（ ） ？
- 幂等性有很多好处，其最大的优势在于**我们可以安全地重试任何幂等性操作，反正它们也不会破坏我们的系统状态**。如果是非幂等性操作，我们还需要担心某些操作执行多次对状态的影响，但对于幂等性操作而言，我们根本无需担心此事。
- 幂等性 Producer
- 在 Kafka 中， ** 默认不是幂等性的**，但我们可以创建幂等性 Producer。它其实是 **0.11.0.0 版本引入的新功能** 。
- 在此之前，
- **props.put(“enable.idempotence”, ture)**，或 **props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG， 。
- enable.idempotence 被设置成 true 后， ** 自动升级成幂等性 Producer**，其他所有的代码逻辑都不需要改变。
- **kafka 自动帮你做消息的重复去重**。底层具体的原理很简单，就是经典的用空间去换时间的优化思路，即在 Broker 端多保存一些字段。当 Producer 发送了具有相同字段值的消息后，
- 必须要了解幂等性 Producer 的作用范围
- 首先，它只能保证**单分区上的幂等性**，即一个幂等性 Producer 能够保证某个主题的一个分区上不出现重复消息，它无法实现多个分区的幂等性。
- 其次，它只能实现**单会话上的幂等性**，不能实现跨会话的幂等性。这里的会话，你可以理解为 Producer 进程的一次运行。当你重启了 Producer 进程之后，这种幂等性保证就丧失了。
- 如果我想实现多分区以及多会话上的消息无重复，应该怎么做呢？答案就是**事务（ ） ！
- 事务
- 经典的 ACID， （ ）
- 隔离性表明并发执行的事务彼此相互隔离，互不影响。
- Kafka 自 **0.11 版本开始也提供了对事务的支持** ，目前主要是在 read committed 隔离级别上做事情。它能保证多条消息原子性地写入到目标分区，同时也能保证 Consumer 只能看到事务成功提交的消息。
- 事务型 Producer
- 事务型 Producer 能够保证将消息原子性地写入到多个分区中。这批消息要么全部写入成功，要么全部失败。另外，事务型 Producer 也不惧进程的重启。Producer 重启回来后，
- 设置事务型 Producer 的方法也很简单，满足两个要求即可：
- 和幂等性 Producer 一样，开启 **enable.idempotence = true** 。
- **设置 Producer 端参数 transctional. id**。最好为其设置一个有意义的名字。
- 此外，你还需要在 Producer 代码中做一些调整，如这段代码所示：
```java
producer.initTransactions();
try {
producer.beginTransaction();
producer.send(record1);
producer.send(record2);
producer.commitTransaction();
} catch (KafkaException e) {
producer.abortTransaction();
}
```
- 和普通 Producer 代码相比，事务型 Producer 的显著特点是调用了一些事务 API， ，
- 这段代码能够保证 Record1 和 Record2 被当作一个事务统一提交到 Kafka， ， ，
因此在 Consumer 端，读取事务型 Producer 发送的消息也是需要一些变更的。
- 修改起来也很简单，设置 isolation.level 参数的值即可。当前这个参数有两个取值：
- read_uncommitted： ， ，
- read_committed：
- 比起幂等性 Producer，
- 补充
- 事务的使用场景
- 事务更多用在Kafka Streams中。如果要实现流处理中的精确一次语义，
- 事务型 Producer 可以实现一组消息要么全部写入成功，要么全部失败，但是事务型 Producer 是具体怎么实现多分区以及多会话上的消息无重复的呢？
- 主要的机制是两阶段提交（ ）
### 2.7 消费者组到底是什么？
- 用一句话概括就是：
- 既然是一个组， （ ） ， ， （ ） （ ） ，
- Consumer Group 记住下面这三个特性
- Consumer Group 下可以有一个或多个 Consumer 实例。这里的实例可以是一个单独的进程，也可以是同一进程下的线程。在实际场景中，**使用进程更为常见一些**。
- Group ID 是一个字符串，在一个 Kafka 集群中，它标识唯一的一个 Consumer Group。
- Consumer Group 下所有实例订阅的主题的单个分区，只能分配给组内的某个 Consumer 实例消费。这个分区当然也可以被其他的 Group 消费。
- 两种消息引擎模型
- 点对点模型和发布 / 订阅模型，前者也称为消费队列。
- Kafka 的 Consumer Group 就是这样的机制。当 Consumer Group 订阅了多个主题后，**组内的每个实例不要求一定要订阅主题的所有分区，它只会消费部分分区中的消息**。
- Consumer Group 之间彼此独立，互不影响，它们能够订阅相同的一组主题而互不干涉。再加上 Broker 端的消息留存机制，
- 可以这么说，
- 如果所有实例都属于同一个 Group， ； ，
- 在实际使用场景中，我怎么知道一个 Group 下该有多少个 Consumer 实例呢？
- **理想情况下，
- 假设一个 Consumer Group 订阅了 3 个主题，分别是 A、B、C， ， ，
- **在实际使用过程中一般不推荐设置大于总分区数的 Consumer 实例**。设置多余的实例只会浪费资源，而没有任何好处。
- 针对 Consumer Group， ， ： （ ）
- 看上去该 Offset 就是一个数值而已，其实对于 Consumer Group 而言，它是一组 KV 对， ，
- 如果用 Java 来表示的话，你大致可以认为是这样的数据结构，即 Map< TopicPartition , Long > ，其中 TopicPartition 表示一个分区，而 Long 表示位移的类型。
- Kafka 源码中并不是这样简单的数据结构，而是要比这个复杂得多
- Kafka 有新旧客户端 API 之分，那自然也就有新旧 Consumer 之分。老版本的 Consumer 也有消费者组的概念，它和我们目前讨论的 Consumer Group 在使用感上并没有太多的不同，只是**它管理位移的方式和新版本是不一样的**。
- 老版本的 Consumer Group 把位移保存在 ZooKeeper 中。Apache ZooKeeper 是一个分布式的协调服务框架，
- **现在比较流行的提法是将服务器节点做成无状态的，这样可以自由地扩缩容，实现超强的伸缩性。**
- Kafka 最开始也是基于这样的考虑，才将 Consumer Group 位移保存在独立于 Kafka 集群之外的框架中。
- 发现了一个问题，即 ZooKeeper 这类元框架其实并不适合进行频繁的写更新，而 Consumer Group 的位移更新却是一个非常频繁的操作。这种大吞吐量的写操作会极大地拖慢 ZooKeeper 集群的性能，因此 Kafka 社区渐渐有了这样的共识：将 Consumer 位移保存在 ZooKeeper 中是不合适的做法。
- 于是，在新版本的 Consumer Group 中，
- **新版本的 Consumer Group 将位移保存在 Broker 端的内部主题中**。
- 最后，我们来说说 Consumer Group 端大名鼎鼎的重平衡，也就是所谓的 Rebalance 过程。我形容其为“大名鼎鼎”，从某种程度上来说其实也是“臭名昭著”，因为有关它的 bug 真可谓是此起彼伏，从未间断
- Rebalance 本质上是一种协议，规定了一个 Consumer Group 下的所有 Consumer 如何达成一致，来分配订阅 Topic 的每个分区。
- 比如某个 Group 下有 20 个 Consumer 实例，它订阅了一个具有 100 个分区的 Topic。正常情况下，
- 那么 Consumer Group 何时进行 Rebalance 呢？
- 组成员数发生变更。比如有新的 Consumer 实例加入组或者离开组，抑或是有 Consumer 实例崩溃被“踢出”组。
- 订阅主题数发生变更。Consumer Group 可以使用正则表达式的方式订阅主题，比如 consumer.subscribe(Pattern.compile(“t.*c”)) 就表明该 Group 订阅所有以字母 t 开头、字母 c 结尾的主题。在 Consumer Group 的运行过程中，你新创建了一个满足这样条件的主题，那么该 Group 就会发生 Rebalance。
- 订阅主题的分区数发生变更。Kafka 当前只能允许增加一个主题的分区数。当分区数增加时，就会触发订阅该主题的所有 Group 开启 Rebalance。
- Rebalance 发生时，
- 当前 Kafka 默认提供了 3 种分配策略，每种策略都有一定的优势和劣势，我们今天就不展开讨论了
- 例子来说明一下 Consumer Group 发生 Rebalance 的过程。假设目前某个 Consumer Group 下有两个 Consumer， ， ， ，
- 
- 显然，
- Rebalance 过程对 Consumer Group 消费过程有极大的影响
- 在 Rebalance 过程中，所有 Consumer 实例都会停止消费，等待 Rebalance 完成。
- 其次，目前 Rebalance 的设计是所有 Consumer 实例共同参与，全部重新分配所有分区。其实更高效的做法是尽量减少分配方案的变动。
- 例如实例 A 之前负责消费分区 1、2、3， ， ，
- 最后，
- 最好的解决方案就是避免 Rebalance 的发生。
- 补充
- 消费组中的消费者个数如果超过topic的分区数， ， ？
- 如果使用assign， （ ） ， ， ，
- 减少topic的分区数可以减少服务部署时rebalance的时间呢？
- 减少consumer个数也有缩短rebalance。
- Kafka 社区重新设计了 Consumer Group 的位移管理方式”
- 0.9版本引入了新版本consumer。新版本将位移保存在__consumer_offsets中， ， ，
### 2.8 揭开神秘的“位移主题”面纱
- Kafka 中神秘的内部主题（ ）
- __consumer_offsets 在 Kafka 源码中有个更为正式的名字，叫位移主题，即 Offsets Topic。为了方便今天的讨论， ，
- 新版本 Consumer 的位移管理机制其实也很简单，就是将 Consumer 的位移数据作为一条条普通的 Kafka 消息，提交到 __consumer_offsets 中。可以这么说，**__consumer_offsets 的主要作用是保存 Kafka 消费者的位移信息**。
- 它要求这个提交过程不仅要实现高持久性， ，
- 位移主题就是普通的 Kafka 主题。你可以手动地创建它、修改它，甚至是删除它。只不过，它同时也是一个内部主题，大部分情况下，你其实并不需要“搭理”它，也不用花心思去管理它，把它丢给 Kafka 就完事了。
- 虽说位移主题是一个普通的 Kafka 主题，但它的**消息格式却是 Kafka 自己定义的**，
- 事实上，
- 首先从 Key 说起。
- 一个 Kafka 集群中的 Consumer 数量会有很多，既然这个主题保存的是 Consumer 的位移数据，那么消息格式中必须要有字段来标识这个位移数据是哪个 Consumer 的。这种数据放在哪个字段比较合适呢？显然放在 Key 中比较合适。
- Group ID ，它能够标识唯一的 Consumer Group。
- 除了 Consumer Group， ， ，
- Key 中保存了 Group ID， ？ ， ， ， ，
- 结论
- 位移主题的 Key 中应该保存 3 部分内容：< Group ID ，主题名，分区号 >
- 消息体的设计
- 也许你会觉得消息体应该很简单，保存一个位移值就可以了。实际上，社区的方案要复杂得多，比如消息体还保存了位移提交的一些其他元数据，诸如时间戳和用户自定义的数据等。
- 保存这些元数据是为了帮助 Kafka 执行各种各样后续的操作，比如删除过期位移消息等。但总体来说，我们还是可以简单地认为消息体就是保存了位移值。
- 位移主题的消息格式可不是只有这一种。事实上，它有 3 种消息格式。除了刚刚我们说的这种格式，还有 2 种格式：
- 用于保存 Consumer Group 信息的消息。
- 用于删除 Group 过期位移甚至是删除 Group 的消息。
- 第 1 种格式非常神秘，以至于你几乎无法在搜索引擎中搜到它的身影。不过，你只需要记住它是用来注册 Consumer Group 的就可以了。
- 第 2 种格式相对更加有名一些。它有个专属的名字： ，
- 何时会写入这类消息呢？
- 一旦某个 Consumer Group 下的所有 Consumer 实例都停止了， ，
- 位移主题是怎么被创建的。
- 通常来说，当 Kafka 集群中的第一个 Consumer 程序启动时，
但如果是 Kafka 自动创建的，分区数是怎么设置的呢？这就要看 Broker 端参数 offsets.topic.num.partitions 的取值了。它的默认值是 50，
如果你曾经惊讶于 Kafka 日志路径下冒出很多 __consumer_offsets-xxx 这样的目录，那么现在应该明白了吧，这就是 Kafka 自动帮你创建的位移主题啊。
- 位移主题是怎么被创建的。
- 通常来说，当 Kafka 集群中的第一个 Consumer 程序启动时，
这就要看 Broker 端参数 offsets.topic.num.partitions 的取值了。它的默认值是 50，
那么现在应该明白了吧，这就是 Kafka 自动帮你创建的位移主题啊。
- **如果位移主题是 Kafka 自动创建的，那么该主题的分区数是 50，
- 你也可以选择手动创建位移主题。
- 具体方法就是，在 Kafka 集群尚未启动任何 Consumer 之前，使用 Kafka API 创建它。手动创建的好处在于，你可以创建满足你实际场景需要的位移主题。比如很多人说 50 个分区对我来讲太多了，我不想要这么多分区，那么你可以自己创建它，不用理会 offsets.topic.num.partitions 的值。
- **让 Kafka 自动创建比较好。目前 Kafka 源码中有一些地方硬编码了 50 分区数，因此如果你自行创建了一个不同于默认分区数的位移主题，可能会碰到各种各种奇怪的问题。**
- 目前 Kafka Consumer 提交位移的方式有两种：自动提交位移和手动提交位移。
- Consumer 端有个参数叫 enable.auto.commit， ，
但这一点同时也是缺点。因为它太省事了，以至于丧失了很大的灵活性和可控性，你完全没法把控 Consumer 端的位移管理。
- 很多与 Kafka 集成的大数据框架都是禁用自动提交位移的，如 Spark、Flink 等。这就引出了另一种位移提交方式：手动提交位移，即设置 enable.auto.commit = false。一旦设置了 false，
Kafka Consumer API 为你提供了位移提交的方法，如 consumer.commitSync 等。当调用这些方法时，
- 如果你选择的是自动提交位移，那么就可能存在一个问题：只要 Consumer 一直启动着，它就会无限期地向位移主题写入消息。
- Kafka 是怎么删除位移主题中的过期消息的呢？
- 答案就是 Compaction。国内很多文献都将其翻译成压缩， ， ，
- Kafka 使用Compact 策略来删除位移主题中的过期消息，避免该主题无限期膨胀。那么应该如何定义 Compact 策略中的过期呢？对于同一个 Key 的两条消息 M1 和 M2， ，
- Compact 的过程就是扫描日志的所有消息，剔除那些过期的消息，然后把剩下的消息整理在一起。
- 
- 图中位移为 0、2 和 3 的消息的 Key 都是 K1。Compact 之后，分区只需要保存位移为 3 的消息，因为它是最新发送的。
- **Kafka 提供了专门的后台线程定期地巡检待 Compact 的主题，看看是否存在满足条件的可删除数据**。
- 这个后台线程叫 Log Cleaner。很多实际生产环境中都出现过位移主题无限膨胀占用过多磁盘空间的问题， ，
- 补充
- consumer端 日常业务发版呢， ，
- 可以考虑使用standalone consumer，
- 具体是从哪个版本开始， ？ ， ？
- 1.0.9， ，
- offset topic是在coordinator对应的broker上创建且只创建一次是么？
- offset topic在整个集群上被创建出来，
- consumer 是如何从这个位移主题中拿到曾经属于自己组的offset呢
- 首先找到对应的Coordinator， ，
### 2.9 消费者组重平衡能避免吗？
- Rebalance 就是让一个 Consumer Group 下所有的 Consumer 实例就如何消费订阅主题的所有分区达成共识的过程。在 Rebalance 过程中，所有 Consumer 实例共同参与，在协调者组件的帮助下，完成订阅主题分区的分配。 但是，在整个过程中，所有实例都不能消费任何消息，因此它对 Consumer 的 TPS 影响很大。
- 所谓协调者，在 Kafka 中对应的术语是 Coordinator，
- 具体来讲，**Consumer 端应用程序在提交位移时，其实是向 Coordinator 所在的 Broker 提交位移**。同样地，当 Consumer 应用启动时，也是向 Coordinator 所在的 Broker 发送各种请求，然后由 Coordinator 负责执行消费者组的注册、成员管理记录等元数据管理操作。
- 所有 Broker 在启动时，都会创建和开启相应的 Coordinator 组件。也就是说，**所有 Broker 都有各自的 Coordinator 组件**。
- Consumer Group 如何确定为它服务的 Coordinator 在哪台 Broker 上呢？ 答案Kafka 内部位移主题 __consumer_offsets 身上。
- 目前，
- 第 1 步：确定由位移主题的哪个分区来保存该 Group 数据：
- 首先， ，
- 其次， ，
- 有了分区号，算法的第 2 步就变得很简单了，我们只需要找出位移主题分区 12 的 Leader 副本在哪个 Broker 上就可以了。
- 第 2 步：找出该分区 Leader 副本所在的 Broker， ，
- 在实际使用过程中， ， ， ，
- 当 Consumer Group 出现问题，需要快速排查 Broker 端日志时，我们能够根据这个算法准确定位 Coordinator 对应的 Broker，
- Rebalance 的弊端是什么呢？总结起来有以下 3 点：
- **Rebalance 影响 Consumer 端 TPS**。这个之前也反复提到了，这里就不再具体讲了。总之就是，在 Rebalance 期间，
- **Rebalance 很慢**。如果你的 Group 下成员很多，就一定会有这样的痛点。还记得我曾经举过的那个国外用户的例子吧？他的 Group 下有几百个 Consumer 实例， ，
- **Rebalance 效率不高**。当前 Kafka 的设计机制决定了每次 Rebalance 时，
- 关于第 3 点，我们来举个简单的例子。比如一个 Group 下有 10 个成员，每个成员平均消费 5 个分区。假设现在有一个成员退出了，此时就需要开启新一轮的 Rebalance，
- 显然，比较好的做法是维持当前 9 个成员消费分区的方案不变，然后将 5 个分区随机分配给这 9 个成员，这样能最大限度地减少 Rebalance 对剩余 Consumer 成员的冲击。
- 遗憾的是，目前 Kafka 并不是这样设计的。在默认情况下，每次 Rebalance 时，之前的分配方案都不会被保留。就拿刚刚这个例子来说，当 Rebalance 开始时， （
- 基于这个原因，社区于 0.11.0.0 版本推出了 StickyAssignor， ， ，
- 特别是针对 Rebalance 慢和影响 TPS 这两个弊端，社区有解决办法吗？
- 我可以很负责任地告诉你：“无解！”特别是 Rebalance 慢这个问题，
- 就我个人经验而言，在真实的业务场景中，很多 Rebalance 都是计划外的或者说是不必要的。我们应用的 TPS 大多是被这类 Rebalance 拖慢的，因此避免这类 Rebalance 就显得很有必要了。
- 要避免 Rebalance， ，
- 组成员数量发生变化
- 订阅主题数量发生变化
- 订阅主题的分区数发生变化
- 后面两个通常都是运维的主动操作，所以它们引发的 Rebalance 大都是不可避免的。接下来，我们主要说说因为组成员数量变化而引发的 Rebalance 该如何避免。
- **如果 Consumer Group 下的 Consumer 实例数量发生变化，就一定会引发 Rebalance**。这是 Rebalance 发生的最常见的原因。我碰到的 99% 的 Rebalance，
- Consumer 实例增加的情况很好理解，**当我们启动一个配置有相同 group.id 值的 Consumer 程序时，实际上就向这个 Group 添加了一个新的 Consumer 实例**。
- 此时，**Coordinator 会接纳这个新实例，将其加入到组中，并重新分配分区**。通常来说，增加 Consumer 实例的操作都是计划内的，可能是出于增加 TPS 或提高伸缩性的需要。总之，它不属于我们要规避的那类“不必要 Rebalance”。
- 我们更在意的是 Group 下实例数减少这件事。如果你就是要停掉某些 Consumer 实例， ， ， ，
- Coordinator 会在什么情况下认为某个 Consumer 实例已挂从而要退组呢？
- 当 Consumer Group 完成 Rebalance 之后，每个 Consumer 实例都会定期地向 Coordinator 发送心跳请求，表明它还存活着。
- 如果某个 Consumer 实例不能及时地发送这些心跳请求，
- Consumer 端有个参数，叫 session.timeout.ms，
- 可以这么说，**session.timout.ms 决定了 Consumer 存活性的时间间隔**。
- 除了这个参数，**Consumer 还提供了一个允许你控制发送心跳请求频率的参数，就是 heartbeat.interval.ms。这个值设置得越小，
- 频繁地发送心跳请求会额外消耗带宽资源，但好处是能够更加快速地知晓当前是否开启 Rebalance， ，
- 除了以上两个参数，
它的默认值是 5 分钟，表示你的 Consumer 程序如果在 5 分钟之内无法消费完 poll 方法返回的消息，那么 Consumer 会主动发起“离开组”的请求，
- 明确一下到底哪些 Rebalance 是“不必要的”。
- **第一类非必要 Rebalance 是因为未能及时发送心跳，导致 Consumer 被“踢出”Group 而引发的**。因此，你需要仔细地设置**session.timeout.ms 和 heartbeat.interval.ms**的值。我在这里给出一些推荐数值，你可以“无脑”地应用在你的生产环境中。
- **设置 session.timeout.ms = 6s**。
- 将 session.timeout.ms 设置成 6s 主要是为了让 Coordinator 能够更快地定位已经挂掉的 Consumer。毕竟， ，
- **设置 heartbeat.interval.ms = 2s**。
- 要保证 Consumer 实例在被判定为“dead”之前，
- **第二类非必要 Rebalance 是 Consumer 消费时间过长导致的。**
- **max.poll.interval.ms**参数值的设置显得尤为关键。如果要避免非预期的 Rebalance， ，
- 总之， ，
- 如果你按照上面的推荐数值恰当地设置了这几个参数，却发现还是出现了 Rebalance，
- 为什么特意说 GC？ ，
- 补充
- Consumer 消费时间过长为啥会导致rebalance？ ？
- consumer主动关闭会主动向Coordinator发送LeaveGroup请求，
- 如果同一个group 的不同consumer 设置的session.timeout.ms 的不一样怎么办？
- 取最大的
- 在一个session.timeout.ms周期内， ， ？
- consumer重启了，
- 2.3修复了之前StickyAssignor的一个重大bug
- 每个 Consumer 实例都会定期地向 Coordinator 发送心跳请求， ？
- 0.10.1之前是在调用poll方法时发送的，
### 2.10 Kafka中位移提交那些事儿
- Consumer 需要向 Kafka 汇报自己的位移数据， （ ） **Consumer 需要为分配给它的每个分区提交各自的位移数据** 。
- 提交位移主要是为了表征 Consumer 的消费进度，这样当 Consumer 发生故障重启之后，就能够从 Kafka 中读取之前提交的位移值，然后从相应的位移处继续消费，从而避免整个消费过程重来一遍。
- 换句话说，位移提交是 Kafka 提供给你的一个工具或语义保障，你负责维持这个语义保障，即如果你提交了位移 X，
- 因为位移提交非常灵活，你完全可以提交任何位移值，但由此产生的后果你也要一并承担。
- 假设你的 Consumer 消费了 10 条消息，你提交的位移值却是 20， ， ～
- 相反地，如果你提交的位移值是 5， ～
- 所以， ， ，
- 你对位移提交的管理直接影响了你的 Consumer 所能提供的消息语义保障。
- KafkaConsumer API， ；
- 所谓自动提交，就是指 Kafka Consumer 在后台默默地为你提交位移， ； ， ，
- 开启自动提交位移的方法很简单。Consumer 端有个参数 enable.auto.commit， ，
- 如果启用了自动提交， ：
- 这段代码展示了设置自动提交位移的方法。
```java
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "test");
props.put("enable.auto.commit", "true");
props.put("auto.commit.interval.ms", "2000");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
KafkaConsumer< String , String > consumer = new KafkaConsumer< >(props);
consumer.subscribe(Arrays.asList("foo", "bar"));
while (true) {
ConsumerRecords< String , String > records = consumer.poll(100);
for (ConsumerRecord< String , String > record : records)
System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
}
```
- 和自动提交相反的，就是手动提交了。开启手动提交位移的方法就是设置 enable.auto.commit 为 false。但是，
- 最简单的 API 就是KafkaConsumer#commitSync()。该方法会提交 KafkaConsumer#poll() 返回的最新位移。从名字上来看，它是一个同步操作，即该方法会一直等待，直到位移被成功提交才会返回。如果提交过程中出现异常，该方法会将异常信息抛出。
```java
while (true) {
ConsumerRecords< String , String > records =
consumer.poll(Duration.ofSeconds(1));
process(records); // 处理消息
try {
consumer.commitSync();
} catch (CommitFailedException e) {
handle(e); // 处理提交失败异常
}
}
```
- 可见，调用 consumer.commitSync() 方法的时机，是在你处理完了 poll() 方法返回的所有消息之后。如果你莽撞地过早提交了位移，就可能会出现消费数据丢失的情况。
- 那么你可能会问，自动提交位移就不会出现消费数据丢失的情况了吗？它能恰到好处地把握时机进行位移提交吗？为了搞清楚这个问题，我们必须要深入地了解一下自动提交位移的顺序。
- 一旦设置了 enable.auto.commit 为 true，
- 从顺序上来说，
- 在默认情况下，
虽然你能够通过减少 auto.commit.interval.ms 的值来提高提交频率，但这么做只能缩小重复消费的时间窗口，不可能完全消除它。**这是自动提交机制的一个缺陷**。
- 反观手动提交位移，它的好处就在于更加灵活，你完全能够把控位移提交的时机和频率。但是，它也有一个缺陷，就是在调用 commitSync() 时，
当然，你可以选择拉长提交间隔，但这样做的后果是 Consumer 的提交频率下降，在下次 Consumer 重启回来后，会有更多的消息被重新消费。
- 鉴于这个问题， ：
由于它是异步的， （ ） ， ，
```java
while (true) {
ConsumerRecords< String , String > records =
consumer.poll(Duration.ofSeconds(1));
process(records); // 处理消息
consumer.commitAsync((offsets, exception) -> {
if (exception != null)
handle(exception);
});
}
```
- commitAsync 是否能够替代 commitSync 呢？
- 显然，如果是手动提交，我们需要将 commitSync 和 commitAsync 组合使用才能到达最理想的效果，原因有两个：
- 我们可以利用 commitSync 的自动重试来规避那些瞬时错误， ，
- 我们不希望程序总处于阻塞状态，影响 TPS。
- 我们来看一下下面这段代码，它展示的是如何将两个 API 方法结合使用进行手动提交。
```java
try {
while (true) {
ConsumerRecords< String , String > records = consumer.poll(Duration.ofSeconds(1));
process(records); // 处理消息
commitAysnc(); // 使用异步提交规避阻塞
}
} catch (Exception e) {
handle(e); // 处理异常
} finally {
try {
consumer.commitSync(); // 最后一次提交使用同步阻塞式提交
} finally {
consumer.close();
}
}
```
- 这段代码同时使用了 commitSync() 和 commitAsync()。对于常规性、阶段性的手动提交，我们调用 commitAsync() 避免程序阻塞，而在 Consumer 要关闭前，我们调用 commitSync() 方法执行同步阻塞式的位移提交，以确保 Consumer 关闭前能够保存正确的位移数据。将两者结合后，我们既实现了异步无阻塞式的位移管理，也确保了 Consumer 位移的正确性
- 所以，如果你需要自行编写代码开发一套 Kafka Consumer 应用，那么我推荐你使用上面的代码范例来实现手动的位移提交。
- 实际上，
** 简单来说，就是直接提交最新一条消息的位移**。但如果我想更加细粒度化地提交位移，该怎么办呢？
- 设想这样一个场景：你的 poll 方法返回的不是 500 条消息，而是 5000 条。那么，你肯定不想把这 5000 条消息都处理完之后再提交位移，因为一旦中间出现差错，之前处理的全部都要重来一遍。这类似于我们数据库中的事务处理。很多时候，我们希望将一个大事务分割成若干个小事务分别提交，这能够有效减少错误恢复的时间。
- 在 Kafka 中也是相同的道理。对于一次要处理很多消息的 Consumer 而言，它会关心社区有没有方法允许它在消费的中间进行位移提交。比如前面这个 5000 条消息的例子，你可能希望每处理完 100 条消息就提交一次位移，这样能够避免大批量的消息重新消费。
- 庆幸的是，
- commitSync(Map< TopicPartition , OffsetAndMetadata > ) 和 commitAsync(Map< TopicPartition , OffsetAndMetadata > )。它们的参数是一个 Map 对象，键就是 TopicPartition， ，
- 就拿刚刚提过的那个例子来说，如何每处理 100 条消息就提交一次位移呢？在这里，我以 commitAsync 为例， ， ，
```java
private Map< TopicPartition , OffsetAndMetadata > offsets = new HashMap< >();
int count = 0;
// ...
while (true) {
ConsumerRecords< String , String > records = consumer.poll(Duration.ofSeconds(1));
for (ConsumerRecord< String , String > record: records) {
process(record); // 处理消息
offsets.put(new TopicPartition(record.topic(), record.partition()), new OffsetAndMetadata(record.offset() + 1);
if(count % 100 == 0)
consumer.commitAsync(offsets, null); // 回调处理逻辑是 null
count++;
}
}
```
- 简单解释一下这段代码。程序先是创建了一个 Map 对象，用于保存 Consumer 消费处理过程中要提交的分区位移，之后开始逐条处理消息，并构造要提交的位移值。还记得之前我说过要提交下一条消息的位移吗？这就是这里构造 OffsetAndMetadata 对象时，使用当前消息位移加 1 的原因。代码的最后部分是做位移的提交。
我在这里设置了一个计数器，每累计 100 条消息就统一提交一次位移。与调用无参的 commitAsync 不同，这里调用了带 Map 对象参数的 commitAsync 进行细粒度的位移提交。这样，这段代码就能够实现每处理 100 条消息就提交一次位移，不用再受 poll 方法返回的消息总数的限制了。
- 补充
- 对于手动同步和异步提交结合的场景， ， ， ， ， ， ？
- 如果调用没有参数的commit，
- poll下来的数据是有序的吗？ ，
- 是的
- 老师好， ？
- 可以控制，
- 自动提交就一定能够保证不丢消息吗？
- 不能绝对保证
### 2.11 CommitFailedException异常怎么处理？
- 所谓 CommitFailedException，
- 如果异常是可恢复的瞬时错误，提交位移的 API 自己就能规避它们了，因为很多提交位移的 API 方法是支持自动错误重试的，
- 每次和 CommitFailedException 一起出现的，还有一段非常著名的注释。
- 第一，我想不出在近 50 万行的 Kafka 源代码中，还有哪个异常类能有这种待遇，可以享有这么大段的注释，来阐述其异常的含义；
- 第二，纵然有这么长的文字解释，却依然有很多人对该异常想表达的含义感到困惑。
- 现在，我们一起领略下这段文字的风采，看看社区对这个异常的最新解释：
> Commit cannot be completed since the group has already rebalanced and assigned the partitions to another member.
> This means that the time between subsequent calls to poll() was longer than the configured max.poll.interval.ms,
> which typically implies that the poll loop is spending too much time message processing.
> **You can address this either by increasing max.poll.interval.**
> **ms or by reducing the maximum size of batches returned in poll() with max.poll.records.**
- 这段话前半部分的意思是，本次提交位移失败了，原因是消费者组已经开启了 Rebalance 过程，并且将要提交位移的分区分配给了另一个消费者实例。
- 出现这个情况的原因是，你的消费者实例连续两次调用 poll 方法的时间间隔超过了期望的 max.poll.interval.ms 参数值。这通常表明，你的消费者实例花费了太长的时间进行消息处理，耽误了调用 poll 方法。
- 在后半部分，社区给出了两个相应的解决办法（即橙色字部分）：
- 增加期望的时间间隔 max.poll.interval.ms 参数值。
- 减少 poll 方法一次性返回的消息数量，即减少 max.poll.records 参数值。
- 在详细讨论这段文字之前，我还想提一句，实际上这段文字总共有 3 个版本，除了上面的这个最新版本，还有 2 个版本，它们分别是：
> Commit cannot be completed since the group has already rebalanced and assigned the partitions to another member.
> **This means that the time between subsequent calls to poll() was longer than the configured session.timeout.**
> **ms**, which typically implies that the poll loop is spending too much time message processing.
> You can address this either by **increasing the session timeout** or by reducing the maximum size of batches returned in poll() with max.poll.records.
> Commit cannot be completed since the group has already rebalanced and assigned the partitions to another member.
> **This means that the time between subsequent calls to poll() was longer than the configured max.poll.interval.**
> **ms**, which typically implies that the poll loop is spending too much time message processing.
> You can address this either by **increasing the session timeout** or by reducing the maximum size of batches returned in poll() with max.poll.records.
- 其实不论是哪段文字，它们都表征位移提交出现了异常。
- 下面我们就来讨论下该异常是什么时候被抛出的。
- 从源代码方面来说，
- 从使用场景来说，有两种典型的场景可能遭遇该异常。
- 场景一
- 我们先说说最常见的场景。当消息处理的总时间超过预设的 max.poll.interval.ms 参数值时，
这是该异常最“正宗”的登场方式。你只需要写一个 Consumer 程序，使用 KafkaConsumer.subscribe 方法随意订阅一个主题，之后设置 Consumer 端参数 max.poll.interval.ms=5 秒，
最后在循环调用 KafkaConsumer.poll 方法之间，插入 Thread.sleep(6000) 和手动提交位移，就可以成功复现这个异常了。在这里，我展示一下主要的代码逻辑。
```java
…
Properties props = new Properties();
…
props.put("max.poll.interval.ms", 5000);
consumer.subscribe(Arrays.asList("test-topic"));
while (true) {
ConsumerRecords< String , String > records =
consumer.poll(Duration.ofSeconds(1));
// 使用 Thread.sleep 模拟真实的消息处理逻辑
Thread.sleep(6000L);
consumer.commitSync();
}
```
- 如果要防止这种场景下抛出异常，你需要简化你的消息处理逻辑。具体来说有 4 种方法。
- **缩短单条消息处理的时间**。比如，之前下游系统消费一条消息的时间是 100 毫秒，优化之后成功地下降到 50 毫秒，那么此时 Consumer 端的 TPS 就提升了一倍。
- **增加 Consumer 端允许下游系统消费一批消息的最大时长**。这取决于 Consumer 端参数 max.poll.interval.ms 的值。在最新版的 Kafka 中，该参数的默认值是 5 分钟。
如果你的消费逻辑不能简化， ， ，
那么你需要增加 session.timeout.ms 参数的值。不幸的是，
这也是社区在 0.10.1.0 版本引入 max.poll.interval.ms 参数，将这部分含义从 session.timeout.ms 中剥离出来的原因之一。
- **减少下游系统一次性消费的消息总数**。这取决于 Consumer 端参数 max.poll.records 的值。当前该参数的默认值是 500 条，表明调用一次 KafkaConsumer.poll 方法，
最多返回 500 条消息。可以说，该参数规定了单次 poll 方法能够返回的消息总数的上限。如果前两种方法对你都不适用的话，降低此参数值是避免 CommitFailedException 异常最简单的手段。
- **下游系统使用多线程来加速消费**。这应该算是“最高级”同时也是最难实现的解决办法了。具体的思路就是，让下游系统手动创建多个消费线程处理 poll 方法返回的一批消息。
之前你使用 Kafka Consumer 消费数据更多是单线程的，所以当消费速度无法匹及 Kafka Consumer 消息返回的速度时，它就会抛出 CommitFailedException 异常。
如果是多线程，你就可以灵活地控制线程数量，随时调整消费承载能力，再配以目前多核的硬件条件，该方法可谓是防止 CommitFailedException 最高档的解决之道。事实上，
很多主流的大数据流处理框架使用的都是这个方法，比如 Apache Flink 在集成 Kafka 时，就是创建了多个 KafkaConsumerThread 线程，自行处理多线程间的数据消费。
不过，凡事有利就有弊，这个方法实现起来并不容易，特别是在多个线程间如何处理位移提交这个问题上，更是极容易出错。在专栏后面的内容中，我将着重和你讨论一下多线程消费的实现方案。
- 综合以上这 4 个处理方法，**我个人推荐你首先尝试采用方法 1 来预防此异常的发生**。优化下游系统的消费逻辑是百利而无一害的法子，不像方法 2、3 那样涉及到 Kafka Consumer 端 TPS 与消费延时（ ）
- 首先，你需要弄清楚你的下游系统消费每条消息的平均延时是多少。比如你的消费逻辑是从 Kafka 获取到消息后写入到下游的 MongoDB 中，假设访问 MongoDB 的平均延时不超过 2 秒，那么你可以认为消息处理需要花费 2 秒的时间。
如果按照 max.poll.records 等于 500 来计算，一批消息的总消费时长大约是 1000 秒，因此你的 Consumer 端的 max.poll.interval.ms 参数值就不能低于 1000 秒。
如果你使用默认配置，那默认值 5 分钟显然是不够的，你将有很大概率遭遇 CommitFailedException 异常。将 max.poll.interval.ms 增加到 1000 秒以上的做法就属于上面的第 2 种方法。
- 除了调整 max.poll.interval.ms 之外，你还可以选择调整 max.poll.records 值，减少每次 poll 方法返回的消息数。
还拿刚才的例子来说，你可以设置 max.poll.records 值为 150， ， （ ） ，
这种减少 max.poll.records 值的做法就属于上面提到的方法 3。
- 场景二
- 从理论上讲，关于该异常你了解到这个程度，已经足以帮助你应对应用开发过程中由该异常带来的“坑”了 。但其实，该异常还有一个不太为人所知的出现场景。了解这个冷门场景，可以帮助你拓宽 Kafka Consumer 的知识面，也能提前预防一些古怪的问题。下面我们就来说说这个场景。
- 之前我们花了很多时间学习 Kafka 的消费者，不过大都集中在消费者组上，即所谓的 Consumer Group。其实， **Standalone Consumer** 的独立消费者。它没有消费者组的概念，每个消费者实例都是独立工作的，彼此之间毫无联系。
- 不过，你需要注意的是，独立消费者的位移提交机制和消费者组是一样的，因此独立消费者的位移提交也必须遵守之前说的那些规定，比如独立消费者也要指定 group.id 参数才能提交位移。你可能会觉得奇怪，既然是独立消费者，为什么还要指定 group.id 呢？
没办法，谁让社区就是这么设计的呢？总之，消费者组和独立消费者在使用之前都要指定 group.id。
- 现在问题来了，如果你的应用中同时出现了设置相同 group.id 值的消费者组程序和独立消费者程序， ，
- 虽然说这个场景很冷门，但也并非完全不会遇到。在一个大型公司中，特别是那些将 Kafka 作为全公司级消息引擎系统的公司中，每个部门或团队都可能有自己的消费者应用，谁能保证各自的 Consumer 程序配置的 group.id 没有重复呢？一旦出现不凑巧的重复，发生了上面提到的这种场景，你使用之前提到的哪种方法都不能规避该异常。
- 令人沮丧的是，无论是刚才哪个版本的异常说明，都完全没有提及这个场景，因此，如果是这个原因引发的 CommitFailedException 异常，前面的 4 种方法全部都是无效的。
- 更为尴尬的是，无论是社区官网，还是网上的文章，都没有提到过这种使用场景。我个人认为，这应该算是 Kafka 的一个 bug。比起返回 CommitFailedException 异常只是表明提交位移失败，更好的做法应该是，在 Consumer 端应用程序的某个地方，能够以日志或其他方式友善地提示你错误的原因，这样你才能正确处理甚至是预防该异常。
- 补充
- 1、请问Standalone Consumer 的独立消费者一般什么情况会用到 2、Standalone Consumer 的独立消费者 使用跟普通消费者组有什么区别的。
- 1. 很多流处理框架的Kafka connector都没有使用consumer group， ， ， ， （ ）
- 0.10.1.0之后的session.timeout.ms还有什么作用呢？
- 用于侦测会话超时。standalone consumer和group的区分体现在API上
### 2.12 多线程开发消费者实例
- Kafka Java Consumer 设计原理
- 谈到 Java Consumer API， ，
- 所谓用户主线程，就是你启动 Consumer 应用程序 main 方法的那个线程， （ ） ， （ ） ，
- 不过，虽然有心跳线程，**但实际的消息获取逻辑依然是在用户主线程中完成的**。因此，在消费消息的这个层面上，我们依然可以**安全地认为 KafkaConsumer 是单线程的设计**。
- 其实，在社区推出 Java Consumer API 之前， ， ， ，
- 多线程方案
- 我们来具体分析一下 KafkaConsumer 这个类的使用方法，以及如何推演出对应的多线程方案。
- 我们要明确的是，**KafkaConsumer 类不是线程安全的 (thread-safe)**。所有的网络 I/O 处理都是发生在用户主线程中，因此，你在使用过程中必须要确保线程安全。简单来说，就是你不能在多个线程中共享同一个 KafkaConsumer 实例，否则程序会抛出 ConcurrentModificationException 异常。
- 当然了， ， ，
- 鉴于 KafkaConsumer 不是线程安全的事实，我们能够制定两套多线程方案。
- **消费者程序启动多个线程，每个线程维护专属的 KafkaConsumer 实例，负责完整的消息获取、消息处理流程**。如下图所示：
- 
- **消费者程序使用单或多线程获取消息，同时创建多个消费线程执行消息处理逻辑**。获取消息的线程可以是一个，也可以是多个，每个线程维护专属的 KafkaConsumer 实例，**处理消息则交由特定的线程池来做**，从而实现消息获取与消息处理的真正解耦。具体架构如下图所示：
- 
- 总体来说，这两种方案都会创建多个线程，这些线程都会参与到消息的消费过程中，但各自的思路是不一样的。
- 比如一个完整的消费者应用程序要做的事情是 1、2、3、4、5， ， ， ；
而方案 2 则更细粒度化，它会将 1、2 分割出来，用单线程（也可以是多线程）来做，对于 3、4、5，
- 这两种方案的优缺点，我们先来看看下面这张表格。
- 
- 我们先看方案 1，
- 实现起来简单，因为它比较符合目前我们使用 Consumer API 的习惯。我们在写代码的时候，使用多个线程并在每个线程中创建专属的 KafkaConsumer 实例就可以了。
- 多个线程之间彼此没有任何交互，省去了很多保障线程安全方面的开销。
- 由于每个线程使用专属的 KafkaConsumer 实例来执行消息获取和消息处理逻辑， ，
- 说完了方案 1 的优势，我们来看看这个方案的不足之处。
- 每个线程都维护自己的 KafkaConsumer 实例， ，
- 这个方案能使用的线程数受限于 Consumer 订阅主题的总分区数。我们知道，在一个消费者组中，每个订阅分区都只能被组内的一个消费者实例所消费。假设一个消费者组订阅了 100 个分区，那么方案 1 最多只能扩展到 100 个线程，多余的线程无法分配到任何分区，只会白白消耗系统资源。
当然了，这种扩展性方面的局限可以被多机架构所缓解。除了在一台机器上启用 100 个线程消费数据，我们也可以选择在 100 台机器上分别创建 1 个线程，效果是一样的。因此，如果你的机器资源很丰富，这个劣势就不足为虑了。
- 每个线程完整地执行消息获取和消息处理逻辑。一旦消息处理逻辑很重，造成消息处理速度慢，就很容易出现不必要的 Rebalance， ， ，
- 下面我们来说说方案 2。
- 与方案 1 的粗粒度不同，方案 2 将任务切分成了消息获取和消息处理两个部分，分别由不同的线程处理它们。比起方案 1，
如果你的消费获取速度慢，那么增加消费获取的线程数即可；如果是消息的处理速度慢，那么增加 Worker 线程池线程数即可。
- 它的缺陷
- 它的实现难度要比方案 1 大得多，毕竟它有两组线程，你需要分别管理它们。
- 因为该方案将消息获取和消息处理分开了，也就是说获取某条消息的线程不是处理该消息的线程，因此无法保证分区内的消费顺序。举个例子，比如在某个分区中，消息 1 在消息 2 之前被保存，那么 Consumer 获取消息的顺序必然是消息 1 在前，消息 2 在后，但是，
后面的 Worker 线程却有可能先处理消息 2， ， ，
- 方案 2 引入了多组线程，使得整个消息消费链路被拉长，最终导致正确位移提交会变得异常困难，结果就是可能会出现消息的重复消费。如果你在意这一点，那么我不推荐你使用方案 2。
- 实现代码示例
- 分享一段方案 1 的主体代码：
```java
public class KafkaConsumerRunner implements Runnable {
private final AtomicBoolean closed = new AtomicBoolean(false);
private final KafkaConsumer consumer;
public void run() {
try {
consumer.subscribe(Arrays.asList("topic"));
while (!closed.get()) {
ConsumerRecords records =
consumer.poll(Duration.ofMillis(10000));
// 执行消息处理逻辑
}
} catch (WakeupException e) {
// Ignore exception if closing
if (!closed.get()) throw e;
} finally {
consumer.close();
}
}
// Shutdown hook which can be called from a separate thread
public void shutdown() {
closed.set(true);
consumer.wakeup();
}
```
- 这段代码创建了一个 Runnable 类，表示执行消费获取和消费处理的逻辑。每个 KafkaConsumerRunner 类都会创建一个专属的 KafkaConsumer 实例。在实际应用中，你可以创建多个 KafkaConsumerRunner 实例，并依次执行启动它们，以实现方案 1 的多线程架构。
- 对于方案 2 来说，核心的代码是这样的：
```java
private final KafkaConsumer< String , String > consumer;
private ExecutorService executors;
...
private int workerNum = ...;
executors = new ThreadPoolExecutor(
workerNum, workerNum, 0L, TimeUnit.MILLISECONDS,
new ArrayBlockingQueue< >(1000),
new ThreadPoolExecutor.CallerRunsPolicy());
...
while (true) {
ConsumerRecords< String , String > records =
consumer.poll(Duration.ofSeconds(1));
for (final ConsumerRecord record : records) {
// 最重要的部分
executors.submit(new Worker(record));
}
}
..
```
- 当 Consumer 的 poll 方法返回消息后，由专门的线程池来负责处理具体的消息。调用 poll 方法的主线程不负责消息处理逻辑，这样就实现了方案 2 的多线程架构。
- 补充
- Kafka重启时间比较长， ， ， ， ？
- 有可能是要加载的日志段数据太多导致的，
- 方案2的代码consumer实例也是单线程的？
- 嗯，如果唯一用来拉取消息不执行小处理逻辑，那么单线程已然很高效了。
### 2.13 Java 消费者是如何管理TCP连接的?
## 4. Consumer
## 5. Stream
## 6. Connect
## 7. Kafka集群部署与开发
## 8. Kafka集群监控、安全与最佳实践
## 9. Kafka面试点梳理
