Kafka 运维

一、Kafka 集群规划

1、操作系统

部署生产环境的 Kafka，强烈建议操作系统选用 Linux，根据官网的测试报告，XFS 的性能要强于 ext4，因此生产环境建议使用 XFS 文件系统。。

在 Linux 部署 Kafka 能够享受到零拷贝技术所带来的快速数据传输特性。

Windows 平台上部署 Kafka 只适合于个人测试或用于功能验证，千万不要应用于生产环境。

1、磁盘

Kafka 集群部署选择普通的机械磁盘还是固态硬盘？前者成本低且容量大，但易损坏；后者性能优势大，不过单价高。

结论是：使用普通机械硬盘即可。

SSD 固态硬盘比机械硬盘快主要体现在随机读写方面，比如 MySQL 中经常需要对硬盘进行随机读写，就要用到 SSD 固态硬盘。而 Kafka 在写磁盘的时候是 append-only 顺序追加写入的，而机械硬盘顺序读写的性能和内存是差不多的，所以对于 Kafka 集群来说使用机械硬盘就可以了。

就 Kafka 而言，一方面 Kafka 自己实现了冗余机制来提供高可靠性；另一方面通过分区的概念，Kafka 也能在软件层面自行实现负载均衡。因此可以不搭建 RAID，使用普通磁盘组成的存储空间即可。

2、带宽

大部分公司使用普通的以太网络，千兆网络（1Gbps）应该是网络的标准配置。

通常情况下你只能假设 Kafka 会用到 70% 的带宽资源，因为总要为其他应用或进程留一些资源。此外，通常要再额外预留出 2/3 的资源，因为不能让带宽资源总是保持在峰值。

基于以上原因，一个 Kafka 集群数量的大致推算公式如下：

Kafka 机器数 = 单位时间需要处理的总数据量 / 单机所占用带宽

3、内存

Kafka 自身的 JVM 是用不了过多堆内存的，因为 Kafka 设计就是规避掉用 JVM 对象来保存数据，避免频繁 Full GC 导致的问题。建议设置 Kafka 的 JVM 堆内存为 6G，这是业界比较公认的一个合理的值。Kafka 会大量用到 Page Cache 来提升读写效率，将剩下的系统内存都作为 Page Cache 空间。这里建议最少选择 32G 内存的服务器，当然如果是 64G 内存那就更好了，这样可以放更多数据到 Page Cache 中。

4、CPU

通常情况下 Kafka 不太占用 CPU，CPU 不是性能的瓶颈。Kafka 的服务器一般是建议 12 核，当然如果可以给到 16 核那就最好不过了。

5、系统参数设置

1、文件描述符

Kafka 会读写大量的文件并且和客户端建立大量的 Socket 连接，在 Linux 系统中一切皆文件，这些操作都需要使用大量的文件描述符。默认 Linux 系统只允许每个线程使用 1024 个文件描述符，这对 Kafka 来说显然是不够的，因此需要增加线程可以使用的文件描述符到 100000。

# 编辑配置文件 /etc/security/limits.conf  (永久生效)
* - nofile 100000
# 命令行执行（当前会话生效）
ulimit -n 100000

2、线程数

Kafka 中主要有以下几类线程：

• Kafka 在网络层采用的是 Reactor 模式，是一种基于事件驱动的模式。其中有 3 种线程：

• Acceptor 线程：1 个接收线程，负责监听新的连接请求，同时注册 OP_ACCEPT 事件，将新的连接按照轮询的方式交给对应的 Processor 线程处理。

• Processor 线程：N 个 处理器线程，其中每个 Processor 都有自己的 selector，它会向 Acceptor 分配的 SocketChannel 注册相应的 OP_READ 事件，N 的大小由num.networker.threads 参数决定。

• KafkaRequestHandler 线程：M 个 请求处理线程，职责是从 requestChannel 中的 requestQueue 取出 Request，处理以后再将 Response 添加到 requestChannel 中的 ResponseQueue 中。M 的大小由num.io.threads 参数决定；

• 另外 Kafka 在后台还会有一些其他线程：

• 定期清理数据的线程。

• Controller 负责感知和管控整个集群的线程。

• 副本同步拉取数据的线程。

我们可以通过以下方式修改最大可以使用的线程数。

#编辑配置文件 /etc/security/limits.conf  (永久生效)
* - nproc 4096
#命令行执行（当前会话生效）
ulimit -u 4096

3、进程可以使用的最大内存映射区域数

Kafka 之所以吞吐量高，其中的一个重要原因就是因为 Kafka 在 Consumer 读取日志文件时使用了 mmap 的方式。mmap 将磁盘文件映射到内存，支持读和写，对内存的操作会反映在磁盘文件上。当客户端读取 Kafka 日志文件时，在进行 log 文件读取的时候直接将 log 文件读入用户态进行缓存，绕过了内核态的 Page Cache，避免了内核态和用户态的切换。

我们可以通过以下方式修改进程可以使用的最大内存映射区域数。

#编辑配置文件 /etc/sysctl.conf  (永久生效)
vm.max_map_count=262144
编辑完文件后命令行执行 sysctl -p  立即永久生效
#命令行执行
sysctl -w vm.max_map_count=262144 (当前会话生效)

4、关闭 swap

Kafka 重度使用 Page Cache，如果内存页 swap 到磁盘中会严重影响到 Kafka 的性能。 使用以下命令永久关闭 swap。

swapoff -a && sed -i '/ swap / s/^\(.*\)$/#\1/g' /etc/fstab

5、JVM 参数

虽然 Kafka 的服务器端代码是使用 Scala 编写的，但是最终还是编译成 Class 文件在 JVM 上允许，因此运行 Kafka 之前需要准备好 Java 运行环境。Kafka 自 2.0.0 版本开始，已经正式摒弃对 Java 7 的支持了，因此至少使用 Java 8。Kafka 自 3.0.0 版本开始，已经正式摒弃对 Java 8 的支持了，因此至少使用 Java 11，在此不做讨论。

进入 Oracle 官网下载页面 下载 JDK 8 压缩包。

cd /opt/software/
tar -zxvf jdk-8u321-linux-x64.tar.gz
mv jdk1.8.0_321 /usr/local/jdk1.8.0

编辑 /etc/profile 文件添加以下内容，设置 Java 环境变量，路径根据实际安装的位置修改。

vi /etc/profile
export JAVA_HOME=/usr/local/jdk1.8.0
export JRE_HOME=${JAVA_HOME}/jre
export CLASSPATH=.:${JAVA_HOME}/lib:${JRE_HOME}/lib
export PATH=${JAVA_HOME}/bin:$PATH

编辑 /etc/profile 文件添加以下内容，设置 JVM 环境变量，在 Confluent 官网推荐了以下 GC 调优参数，该参数在 LinkedIn 的大型生产环境中得到过验证。

#推荐的 GC 调优参数和 JVM 堆大小设置
export KAFKA_HEAP_OPTS="-Xms6g -Xmx6g -XX:MetaspaceSize=96m -XX:+UseG1GC -XX:MaxGCPauseMillis=20 -XX:InitiatingHeapOccupancyPercent=35 -XX:G1HeapRegionSize=16M -XX:MinMetaspaceFreeRatio=50 -XX:MaxMetaspaceFreeRatio=80"

#后面会用到的环境变量，先提前设置了 #Kafka 环境变量
export KAFKA_HOME=/usr/local/kafka
export PATH=$KAFKA_HOME/bin:$PATH
#JMX 端口，Kafka Eagle 监控会用到
export JMX_PORT="9999"

使环境变量生效。

source /etc/profile

以下是 LinkedIn 中的 Kafka 集群之一在高峰期的统计数据：

• 60 Brokers
• 50k Partitions (replication factor 2)
• 800k Messages/sec in
• 300 MBps inbound, 1 GBps + outbound

在该集群中所有的 Broker 中 90% 的 GC 暂停时间约为 21 毫秒，并且它们每秒执行的 Young GC 不到 1 次。

二、Kafka 单点部署

1、部署zookeeper

1、下载安装包

cd /opt/software/
wget https://dlcdn.apache.org/zookeeper/zookeeper-3.7.0/apache-zookeeper-3.7.0-bin.tar.gz --no-check-certificate

2、解压并重命名服务

tar -zxvf apache-zookeeper-3.7.0-bin.tar.gz
mv apache-zookeeper-3.7.0-bin /usr/local/zookeeper-3.7.0

3、创建zookeeper服务的data目录

mkdir /usr/local/zookeeper-3.7.0/data

4、创建myid文件（zookeeper用于惟一标识自己的id）

echo 1 > /usr/local/zookeeper-3.7.0/data/myid

5、生成zoo.cfg配置文件

cp /usr/local/zookeeper-3.7.0/conf/zoo_sample.cfg /usr/local/zookeeper-3.7.0/conf/zoo.cfg

6、修改zoo.cfg配置文件

vi /usr/local/zookeeper-3.7.0/conf/zoo.cfg

tickTime=2000
initLimit=10
syncLimit=5
dataDir=/usr/local/zookeeper-3.7.0/data
clientPort=2181

server.1=192.168.96.10:2888:3888

2、部署kafka

1、下载软件包并解压、重命名服务

cd /opt/software/
wget https://dlcdn.apache.org/kafka/3.1.0/kafka_2.13-3.1.0.tgz
tar -zxvf kafka_2.13-3.1.0.tgz --no-check-certificate
mv kafka_2.13-3.1.0 /usr/local/kafka-2.13

2、配置kafka

vi /usr/local/kafka-2.13/config/server.properties

kafka服务器

broker.id=0
listeners=PLAINTEXT://192.168.96.10:9092
log.dirs=/tmp/kafka-logs
zookeeper.connect=192.168.96.10:2181

2、启动服务器

由于 Kafka 依赖于 ZooKeeper，所以运行前需要先启动 ZooKeeper

/usr/local/zookeeper-3.7.0/bin/zkServer.sh start

然后，启动 Kafka

cd /usr/local/kafka-2.13 && bin/kafka-server-start.sh config/server.properties &

3、停止服务器

执行所有操作后，可以使用以下命令停止服务器

bin/kafka-server-stop.sh config/server.properties

三、Kafka 集群部署

1、修改配置

复制配置为多份（Windows 使用 copy 命令代理）：

scp -r /usr/local/zookeeper-3.7.0 root@192.168.96.11:/usr/local/
scp -r /usr/local/zookeeper-3.7.0 root@192.168.96.12:/usr/local/
scp -r /usr/local/kafka-2.13 root@192.168.96.11:/usr/local/
scp -r /usr/local/kafka-2.13 root@192.168.96.12:/usr/local/

修改配置：

zookeeper

echo 2 > /usr/local/zookeeper-3.7.0/data/myid			# 192.168.96.11 服务器
echo 3 > /usr/local/zookeeper-3.7.0/data/myid			# 192.168.96.12 服务器

kafka

broker.id=0
listeners=PLAINTEXT://192.168.96.10:9092
log.dirs=/tmp/kafka-logs
zookeeper.connect=192.168.96.10:2181,192.168.96.11:2181,192.168.96.12:2181

broker.id=1
listeners=PLAINTEXT://192.168.96.11:9092

broker.id=2
listeners=PLAINTEXT://192.168.96.12:9092

其中，broker.id 这个参数必须是唯一的。

2、启动

根据这两份配置启动三个服务器节点：

$ cd /usr/local/kafka-2.13 && bin/kafka-server-start.sh config/server.properties &
...
$ cd /usr/local/kafka-2.13 && bin/kafka-server-start.sh config/server.properties &
...
$ cd /usr/local/kafka-2.13 && bin/kafka-server-start.sh config/server.properties &
...

创建一个新的 Topic 使用 三个备份：

bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 3 --partitions 1 --topic my-replicated-topic

查看主题：

$ bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic my-replicated-topic
Topic:my-replicated-topic   PartitionCount:1    ReplicationFactor:3 Configs:
    Topic: my-replicated-topic  Partition: 0    Leader: 1   Replicas: 1,2,0 Isr: 1,2,0

• leader - 负责指定分区的所有读取和写入的节点。每个节点将成为随机选择的分区部分的领导者。
• replicas - 是复制此分区日志的节点列表，无论它们是否为领导者，或者即使它们当前处于活动状态。
• isr - 是“同步”复制品的集合。这是副本列表的子集，该列表当前处于活跃状态并且已经被领导者捕获。

四、Kafka 命令

1、主题（Topic）

1、创建 Topic

kafka-topics --create --zookeeper localhost:2181 --replication-factor 1 --partitions 3 --topic my-topic

2、查看 Topic 列表

kafka-topics --list --zookeeper localhost:2181

3、添加 Partition

kafka-topics --zookeeper localhost:2181 --alter --topic my-topic --partitions 16

4、删除 Topic

kafka-topics --zookeeper localhost:2181 --delete --topic my-topic

5、查看 Topic 详细信息

kafka-topics --zookeeper localhost:2181/kafka-cluster --describe

6、查看备份分区

kafka-topics --zookeeper localhost:2181/kafka-cluster --describe --under-replicated-partitions

2、生产者（Producers）

1、通过控制台输入生产消息

kafka-console-producer --broker-list localhost:9092 --topic my-topic

2、通过文件输入生产消息

kafka-console-producer --broker-list localhost:9092 --topic test < messages.txt

3、通过控制台输入 Avro 生产消息

kafka-avro-console-producer --broker-list localhost:9092 --topic my.Topic --property value.schema='{"type":"record","name":"myrecord","fields":[{"name":"f1","type":"string"}]}' --property schema.registry.url=http://localhost:8081

然后，可以选择输入部分 json key：

{ "f1": "value1" }

4、生成消息性能测试

kafka-producer-perf-test --topic position-reports --throughput 10000 --record-size 300 --num-records 20000 --producer-props bootstrap.servers="localhost:9092"

3、消费者（Consumers）

1、消费所有未消费的消息

kafka-console-consumer --bootstrap-server localhost:9092 --topic my-topic --from-beginning

2、消费一条消息

kafka-console-consumer --bootstrap-server localhost:9092 --topic my-topic  --max-messages 1

3、从指定的 offset 消费一条消息

从指定的 offset __consumer_offsets 消费一条消息：

kafka-console-consumer --bootstrap-server localhost:9092 --topic __consumer_offsets --formatter 'kafka.coordinator.GroupMetadataManager$OffsetsMessageFormatter' --max-messages 1

4、从指定 Group 消费消息

kafka-console-consumer --topic my-topic --new-consumer --bootstrap-server localhost:9092 --consumer-property group.id=my-group

5、消费 avro 消息

kafka-avro-console-consumer --topic position-reports --new-consumer --bootstrap-server localhost:9092 --from-beginning --property schema.registry.url=localhost:8081 --max-messages 10

kafka-avro-console-consumer --topic position-reports --new-consumer --bootstrap-server localhost:9092 --from-beginning --property schema.registry.url=localhost:8081

6、查看消费者 Group 列表

kafka-consumer-groups --new-consumer --list --bootstrap-server localhost:9092

7、查看消费者 Group 详细信息

kafka-consumer-groups --bootstrap-server localhost:9092 --describe --group testgroup

4、配置（Config）

1、设置 Topic 的保留时间

kafka-configs --zookeeper localhost:2181 --alter --entity-type topics --entity-name my-topic --add-config retention.ms=3600000

2、查看 Topic 的所有配置

kafka-configs --zookeeper localhost:2181 --describe --entity-type topics --entity-name my-topic

3、修改 Topic 的配置

kafka-configs --zookeeper localhost:2181 --alter --entity-type topics --entity-name my-topic --delete-config retention.ms

5、ACL

1、查看指定 Topic 的 ACL

kafka-acls --authorizer-properties zookeeper.connect=localhost:2181 --list --topic topicA

2、添加 ACL

kafka-acls --authorizer-properties zookeeper.connect=localhost:2181 --add --allow-principal User:Bob --consumer --topic topicA --group groupA

kafka-acls --authorizer-properties zookeeper.connect=localhost:2181 --add --allow-principal User:Bob --producer --topic topicA

6、ZooKeeper

zookeeper-shell localhost:2182 ls /

五、Kafka 工具

• kafka-manager
• KafkaOffsetMonitor

六、Kafka 核心配置

1、Broker 级别配置

1、存储配置

首先 Broker 是需要配置存储信息的，即 Broker 使用哪些磁盘。那么针对存储信息的重要参数有以下这么几个：

• log.dirs：指定了 Broker 需要使用的若干个文件目录路径。这个参数是没有默认值的，必须由使用者亲自指定。
• log.dir：注意这是 dir，结尾没有 s，说明它只能表示单个路径，它是补充上一个参数用的。

log.dirs 具体格式是一个 CSV 格式，也就是用逗号分隔的多个路径，比如/home/kafka1,/home/kafka2,/home/kafka3这样。如果有条件的话你最好保证这些目录挂载到不同的物理磁盘上。这样做有两个好处：

• 提升读写性能：比起单块磁盘，多块物理磁盘同时读写数据有更高的吞吐量。
• 能够实现故障转移：即 Failover。这是 Kafka 1.1 版本新引入的强大功能。要知道在以前，只要 Kafka Broker 使用的任何一块磁盘挂掉了，整个 Broker 进程都会关闭。但是自 1.1 开始，这种情况被修正了，坏掉的磁盘上的数据会自动地转移到其他正常的磁盘上，而且 Broker 还能正常工作。

2、zookeeper 配置

Kafka 与 ZooKeeper 相关的最重要的参数当属 zookeeper.connect。这也是一个 CSV 格式的参数，比如我可以指定它的值为zk1:2181,zk2:2181,zk3:2181。2181 是 ZooKeeper 的默认端口。

现在问题来了，如果我让多个 Kafka 集群使用同一套 ZooKeeper 集群，那么这个参数应该怎么设置呢？这时候 chroot 就派上用场了。这个 chroot 是 ZooKeeper 的概念，类似于别名。

如果你有两套 Kafka 集群，假设分别叫它们 kafka1 和 kafka2，那么两套集群的zookeeper.connect参数可以这样指定：zk1:2181,zk2:2181,zk3:2181/kafka1和zk1:2181,zk2:2181,zk3:2181/kafka2。切记 chroot 只需要写一次，而且是加到最后的。我经常碰到有人这样指定：zk1:2181/kafka1,zk2:2181/kafka2,zk3:2181/kafka3，这样的格式是不对的。

3、Broker 连接配置

• listeners：告诉外部连接者要通过什么协议访问指定主机名和端口开放的 Kafka 服务。
• advertised.listeners：和 listeners 相比多了个 advertised。Advertised 的含义表示宣称的、公布的，就是说这组监听器是 Broker 用于对外发布的。
• host.name/port：列出这两个参数就是想说你把它们忘掉吧，压根不要为它们指定值，毕竟都是过期的参数了。

我们具体说说监听器的概念，从构成上来说，它是若干个逗号分隔的三元组，每个三元组的格式为<协议名称，主机名，端口号>。这里的协议名称可能是标准的名字，比如 PLAINTEXT 表示明文传输、SSL 表示使用 SSL 或 TLS 加密传输等；也可能是你自己定义的协议名字，比如CONTROLLER: //localhost:9092。

最好全部使用主机名，即 Broker 端和 Client 端应用配置中全部填写主机名。

4、Topic 管理

• auto.create.topics.enable：是否允许自动创建 Topic。一般设为 false，由运维把控创建 Topic。
• unclean.leader.election.enable：是否允许 Unclean Leader 选举。
• auto.leader.rebalance.enable：是否允许定期进行 Leader 选举。

第二个参数unclean.leader.election.enable是关闭 Unclean Leader 选举的。何谓 Unclean？还记得 Kafka 有多个副本这件事吗？每个分区都有多个副本来提供高可用。在这些副本中只能有一个副本对外提供服务，即所谓的 Leader 副本。

那么问题来了，这些副本都有资格竞争 Leader 吗？显然不是，只有保存数据比较多的那些副本才有资格竞选，那些落后进度太多的副本没资格做这件事。

好了，现在出现这种情况了：假设那些保存数据比较多的副本都挂了怎么办？我们还要不要进行 Leader 选举了？此时这个参数就派上用场了。

如果设置成 false，那么就坚持之前的原则，坚决不能让那些落后太多的副本竞选 Leader。这样做的后果是这个分区就不可用了，因为没有 Leader 了。反之如果是 true，那么 Kafka 允许你从那些“跑得慢”的副本中选一个出来当 Leader。这样做的后果是数据有可能就丢失了，因为这些副本保存的数据本来就不全，当了 Leader 之后它本人就变得膨胀了，认为自己的数据才是权威的。

这个参数在最新版的 Kafka 中默认就是 false，本来不需要我特意提的，但是比较搞笑的是社区对这个参数的默认值来来回回改了好几版了，鉴于我不知道你用的是哪个版本的 Kafka，所以建议你还是显式地把它设置成 false 吧。

第三个参数auto.leader.rebalance.enable的影响貌似没什么人提，但其实对生产环境影响非常大。设置它的值为 true 表示允许 Kafka 定期地对一些 Topic 分区进行 Leader 重选举，当然这个重选举不是无脑进行的，它要满足一定的条件才会发生。严格来说它与上一个参数中 Leader 选举的最大不同在于，它不是选 Leader，而是换 Leader！比如 Leader A 一直表现得很好，但若auto.leader.rebalance.enable=true，那么有可能一段时间后 Leader A 就要被强行卸任换成 Leader B。

你要知道换一次 Leader 代价很高的，原本向 A 发送请求的所有客户端都要切换成向 B 发送请求，而且这种换 Leader 本质上没有任何性能收益，因此我建议你在生产环境中把这个参数设置成 false。

5、数据留存

• log.retention.{hour|minutes|ms}：都是控制一条消息数据被保存多长时间。从优先级上来说 ms 设置最高、minutes 次之、hour 最低。通常情况下我们还是设置 hour 级别的多一些，比如log.retention.hour=168表示默认保存 7 天的数据，自动删除 7 天前的数据。很多公司把 Kafka 当做存储来使用，那么这个值就要相应地调大。
• log.retention.bytes：这是指定 Broker 为消息保存的总磁盘容量大小。这个值默认是 -1，表明你想在这台 Broker 上保存多少数据都可以，至少在容量方面 Broker 绝对为你开绿灯，不会做任何阻拦。这个参数真正发挥作用的场景其实是在云上构建多租户的 Kafka 集群：设想你要做一个云上的 Kafka 服务，每个租户只能使用 100GB 的磁盘空间，为了避免有个“恶意”租户使用过多的磁盘空间，设置这个参数就显得至关重要了。
• message.max.bytes：控制 Broker 能够接收的最大消息大小。默认的 1000012 太少了，还不到 1MB。实际场景中突破 1MB 的消息都是屡见不鲜的，因此在线上环境中设置一个比较大的值还是比较保险的做法。毕竟它只是一个标尺而已，仅仅衡量 Broker 能够处理的最大消息大小，即使设置大一点也不会耗费什么磁盘空间的。

2、Topic 级别配置

• retention.ms：规定了该 Topic 消息被保存的时长。默认是 7 天，即该 Topic 只保存最近 7 天的消息。一旦设置了这个值，它会覆盖掉 Broker 端的全局参数值。
• retention.bytes：规定了要为该 Topic 预留多大的磁盘空间。和全局参数作用相似，这个值通常在多租户的 Kafka 集群中会有用武之地。当前默认值是 -1，表示可以无限使用磁盘空间。

3、操作系统参数

• 文件描述符限制
• 文件系统类型
• Swappiness
• 提交时间

文件描述符系统资源并不像我们想象的那样昂贵，你不用太担心调大此值会有什么不利的影响。通常情况下将它设置成一个超大的值是合理的做法，比如ulimit -n 1000000。其实设置这个参数一点都不重要，但不设置的话后果很严重，比如你会经常看到“Too many open files”的错误。

其次是文件系统类型的选择。这里所说的文件系统指的是如 ext3、ext4 或 XFS 这样的日志型文件系统。根据官网的测试报告，XFS 的性能要强于 ext4，所以生产环境最好还是使用 XFS。对了，最近有个 Kafka 使用 ZFS 的数据报告，貌似性能更加强劲，有条件的话不妨一试。

第三是 swap 的调优。网上很多文章都提到设置其为 0，将 swap 完全禁掉以防止 Kafka 进程使用 swap 空间。我个人反倒觉得还是不要设置成 0 比较好，我们可以设置成一个较小的值。为什么呢？因为一旦设置成 0，当物理内存耗尽时，操作系统会触发 OOM killer 这个组件，它会随机挑选一个进程然后 kill 掉，即根本不给用户任何的预警。但如果设置成一个比较小的值，当开始使用 swap 空间时，你至少能够观测到 Broker 性能开始出现急剧下降，从而给你进一步调优和诊断问题的时间。基于这个考虑，我个人建议将 swappniess 配置成一个接近 0 但不为 0 的值，比如 1。

最后是提交时间或者说是 Flush 落盘时间。向 Kafka 发送数据并不是真要等数据被写入磁盘才会认为成功，而是只要数据被写入到操作系统的页缓存（Page Cache）上就可以了，随后操作系统根据 LRU 算法会定期将页缓存上的“脏”数据落盘到物理磁盘上。这个定期就是由提交时间来确定的，默认是 5 秒。一般情况下我们会认为这个时间太频繁了，可以适当地增加提交间隔来降低物理磁盘的写操作。当然你可能会有这样的疑问：如果在页缓存中的数据在写入到磁盘前机器宕机了，那岂不是数据就丢失了。的确，这种情况数据确实就丢失了，但鉴于 Kafka 在软件层面已经提供了多副本的冗余机制，因此这里稍微拉大提交间隔去换取性能还是一个合理的做法。