1.选择操作系统

• Kafka 是用 Java 开发的应用程序，可运行在多种操作系统中，如 Windows、macOS、Linux 等等，但在一般情况下，推荐 Linux。

2.配置 Java 环境

• ZooKeeper 和 Kafka 是用 Java 语言开发的，所以安装之前需要先准备 Java 环境。
• Linux 环境下安装 Java。

3.安装 ZooKeeper

• ZooKeeper 是一种集中式服务，用于维护配置信息、命名、提供分布式同步和组服务。Kafka 用 ZooKeeper 来保存集群元数据和消费者信息。

• 安装 ZooKeeper。

4.安装 broker

（1）安装 broker

# 解压
tar -zxf kafka_2.13-2.7.0.tgz
# 移动到安装目录下
mv kafka_2.13-2.7.0 /usr/local/kafka
# 创建存放日志片段的目录，由配置文件中 log.dirs 指定
mkdir /tmp/kafka-logs
# 指定配置文件，启动 Kafka，并以守护线程的方式在后台运行（否则一旦退出前台，服务就中断了）
sh kafka-server-start.sh -daemon /usr/local/kafka/config/server.properties

（2）验证是否安装正确

• 创建一个测试主题：

# 创建 topic：复制系数为 1，分区为 1
sh kafka-topics.sh --bootstrap-server localhost:9092 --create --replication-factor 1 
--partitions 1 --topic test
# 查看分区信息
/usr/local/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092
--describe --topic test

• 向主题中写入消息：

# 写入消息，以 CTRL-C 停止发送消息
sh kafka-console-producer.sh --bootstrap-server
localhost:9092 --topic test
> Test Message 1
> Test Message 2
> ^C

• 从主题中读取消息：

/usr/local/kafka/bin/kafka-console-consumer.sh --bootstrap-server
localhost:9092 --topic test --from-beginning

5.配置 broker

• Kafka 自带的配置示例可用来安装单机服务，主要用于概念验证，并不能满足大型集群的配置需求。Kafka 有很多配置参数，涉及安装和调优的方方面面，其中大多数参数可使用默认值，除非对调优有特别的需求。

（1）常规配置

• broker.id：每个 broker 都需要有一个整数标识符，该标识符是使用 broker.id 指定的。它的默认值是0，但可以被设置成其他任意整数，这个值在整个 Kafka 集群中必须是唯一的。
• listeners：listeners 配置参数是一个用逗号分隔的 URI 列表，也就是要监听的地址和端口。
  • 监听器的格式为：<protocol>://<hostname>:<port>。
  • 例如，PLAINTEXT://0.0.0.0:9092（监听所有的网卡，9092端口，无认证），SSL://0.0.0.0:9094（监听所有的网卡，9094端口，需要走认证加密访问）。
  • 在公有云场景下部署 Kafka 集群，公网 IP 不是在本节点网卡上的，所以无法通过 listeners 进行绑定，所以只能通过 0.0.0.0 进行绑定。但是在集群外部时，Kafka 客户端进行连接，它是需要有能力访问 Kafka 的每一个 broker 节点的，所以需要在 advertised.listeners 中配置公网 IP，并存储在ZooKeeper 中，这样 Kafka 客户端就能拿到所有 broker 节点的公网 IP 并进行访问。
  • Kafka 中 listener 和 advertised.listeners 的作用。
• advertised.listeners：该配置指定 Kafka broker 对外公开的网络 IP 和端口，用于告知客户端如何连接到 Kafka broker。公开的方式是通过存储在 ZooKeeper 中进行共享数据的。不填就默认用 listeners 的地址。
• zookeeper.connect：用于保存 broker 元数据的 ZooKeeper 地址是通过 zookeeper.connect 来指定的。这个参数的值是用逗号分隔的一组 hostname:port/path。
  • /path 是可选的 ZooKeeper 路径，以作为 Kafka 集群的 chroot。如果不指定，则默认使用根路径（数据被输出到根路径下）。如果指定的 chroot 路径不存在，那么 broker 会在启动时创建它。chroot 只需要写一次，而且是加到最后的。
• log.dir：存放日志片段的单个目录。
• log.dirs：存放日志片段的多个目录。是一组用逗号分隔的本地文件系统路径。
  • 如果指定了多条路径，那么 broker 会根据“最少使用”原则，把同一个分区的日志片段保存到同一条路径下。需注意的是，broker 会向分区数量最少的目录新增分区，而不是向可用磁盘空间最小的目录新增分区，所以并不能保证数据会被均匀地分布在多个目录中。
• num.recovery.threads.per.data.dir：使用线程池处理日志片段（服务器正常启动时，用于打开每个分区的日志片段。服务器发生崩溃并重启时，用于检查和截短每个分区的日志片段。服务器正常关闭时，用于关闭日志片段）。
  • 默认情况下，每个日志目录只使用一个线程。因为这些线程只在服务器启动和关闭时使用，所以可以多设置一些线程来实现并行操作。特别是对包含大量分区的服务器来说，一旦发生崩溃，在从错误中恢复时可以通过并行操作省下数小时的时间。需要注意的是，这个参数对应的是 log.dirs 中的一个目录，也就是说，如果 num.recovery.threads.per.data.dir 被设为 8，并且 log.dirs 指定了 3 条路径，那么总共需要 24 个线程。
• auto.create.topics.enable：是否能自动创建主题。
• auto.leader.rebalance.enable：确保主题的所有权不会集中在一台 broker 上，可以将这个参数设置为 true，尽可能地在集群中保持均衡。如果启用了这个功能，那么就会有一个后台线程定期检查分区的分布情况（这个时间间隔可以通过 leader.imbalance.check.interval.seconds来配置）。如果不均衡的所有权超出了leader.imbalance.per.broker.percentage 指定的百分比，则会启动一次分区首领再均衡。
• delete.topic.enable：主题能否被删除。

broker.id=1
listeners=PLAINTEXT://10.3.212.55:9092
advertised.listeners=PLANTEXT://10.3.212.55:9092
zookeeper.connect=202.141.195.10:2181,202.141.195.11:2181,202.141.195.12:2181/kafka
log.dirs=/tmp/kafka-logs
num.recovery.threads.per.data.dir=10
auto.create.topics.enable=true
auto.leader.rebalance.enable=false
delete.topic.enable=true

（2）主题的默认配置

可以通过管理工具为每个主题单独配置一些参数，比如分区数和数据保留策略。还可以将服务器提供的默认配置作为基准，应用于集群内的大部分主题。

• num.partitions：指定了新创建的主题将包含多少个分区，特别是如果启用了主题自动创建功能（默认是启用的），那么主题的分区数就是这个参数指定的值，默认是 1 个分区。
  • 可以增加主题的分区数，但不能减少。如果要让一个主题的分区数小于 num.partitions 指定的值，则需要手动创建主题。
  • Kafka 集群通过分区来实现主题的横向伸缩，当不断有新 broker 加入集群时，通过分区数来实现集群的负载均衡就变得十分重要。很多用户将主题的分区数设置为集群 broker 的数量，或者是它的倍数，这样可以让分区均衡地分布到 broker 上，进而均衡地分布消息负载。例如，在一个包含 10 台主机的 Kafka 集群中，有一个主题包含了 10个分区，这 10 个分区的所有权均衡地分布在这 10 台主机上，这样可以获得最佳的吞吐性能。但并不是必须这样做，因为还可以通过其他方式来实现消息负载均衡，比如使用多个主题。
  • 如果要向主题写入和从主题读取 1 GBps 的数据，并且每个消费者可以处理 50 MBps 的数据，那么至少需要 20 个分区。这样就可以让 20 个消费者同时读取这些分区，从而达到 1 GBps的吞吐量。
• default.replication.factor：如果启用了自动创建主题功能，那么这个参数的值就是新创建主题的复制系数。建议将复制系数设置为至少比 min.insync.replicas 大 1 的数。如果你有足够大的集群和足够多的硬件资源，则可以将复制系数设置为比 min.insync.replicas 大 2 的数，因为它允许集群内同时发生一次计划内停机和一次计划外停机。对于典型的集群，这意味着每个分区至少要有 3 个副本。如果在滚动部署或升级 Kafka 或底层操作系统期间出现网络交换机中断、磁盘故障或其他计划外的问题，你可以保证仍然有 1 个副本可用。
• min.insync.replicas：为了提升集群的数据持久性，可以将 min.insync.replicas 设置为 2，确保至少有两个副本跟生产者保持“同步”。生产者需要配合将 ack 设置为 all，这样就可以确保至少有两个副本（首领和另一个副本）确认写入成功，从而防止在以下情况下丢失数据：首领确认写入，然后发生停机，所有权被转移到一个副本，但这个副本没有写入成功。如果没有这些配置，则生产者会认为已经写入成功，但实际上消息丢失了。不过，这样做是有副作用的，因为需要额外的开销，所以效率会有所降低。因此，对于能够容忍偶尔消息丢失的高吞吐量集群，不建议修改这个参数的默认值。
• log.retention.ms：Kafka 通常根据配置的时间长短来决定数据可以被保留多久。可以使用log.retention.hours、log.retention.minutes 和 log.retention.ms 来配置，默认是 1 周。推荐使用 log.retention.ms，因为如果指定了不止一个参数，那么 Kafka 会优先使用具有最小单位值的那个。根据时间保留数据是通过检查日志片段文件的最后修改时间来实现的。一般来说，最后修改时间就是日志片段的关闭时间，也就是文件中最后一条消息的时间戳。
• log.retention.bytes：另一种数据保留策略是通过计算已保留的消息的字节总数来判断旧消息是否过期，对应的是每一个分区。如果一个主题包含 8 个分区，并且 log.retention.bytes 被设置为 1 GB，那么这个主题最多可以保留 8 GB 的数据。需要注意的是，所有保留都是针对单个分区而不是主题执行的。所以，如果配置了这个参数，那么当主题增加了新分区，整个主题可以保留的数据也会随之增加。如果这个值被设置为 –1，那么分区就可以无限期地保留数据。
  • 如果同时指定了 log.retention.bytes 和 log.retention.ms（或另一个按时间保留的参数），那么只要任意一个条件得到满足，消息就会被删除。假设log.retention.ms 被设置为 86 400 000（也就是 1 天），log.retention.bytes被设置为 1 000 000 000（也就是 1 GB），如果消息字节总数不到一天就超过了 1 GB，那么旧数据就会被删除。相反，如果消息字节总数小于 1 GB，那么一天之后这些消息也会被删除，尽管分区的数据总量小于 1 GB。为简单起见，建议只选择其中的一种保留策略，要么基于数据大小，要么基于时间，或者两种都不选择，以防发生意外的数据丢失。不过，对于复杂的场景，可以两种都使用。
• log.segment.bytes：当消息到达 broker 时，它们会被追加到分区的当前日志片段上。当日志片段大小达到 log.segment.bytes 指定的上限（默认是 1 GB）时，当前日志片段会被关闭，一个新的日志片段会被打开。一旦日志片段被关闭，就可以开始进入过期倒计时。这个参数的值越小，关闭和分配新文件就会越频繁，从而降低整体的磁盘写入效率。如果主题的消息量不是很大，那么如何设置这个参数就变得尤为重要。如果一个主题每天只接收 100 MB 的消息，并且 log.segment.bytes 使用了默认设置，那么填满一个日志片段将需要 10 天。因为在日志片段被关闭之前消息是不会过期的，所以如果 log.retention.ms被设为 604 800 000（也就是 1 周），那么日志片段最多需要 17 天才会过期。这是因为关闭日志片段需要 10 天，而根据配置的过期时间，还需要再保留数据 7 天（要等到日志片段的最后一条消息过期才能将其删除）。
  • 日志片段的大小也会影响使用时间戳获取偏移量的行为。当使用时间戳获取日志偏移量时，Kafka 会查找在指定时间戳写入的日志片段文件，也就是创建时间在指定时间戳之前且最后修改时间在指定时间戳之后的文件。然后，Kafka 会返回这个日志片段开头的偏移量（也就是文件名）
• log.roll.ms：另一个可用于控制日志片段关闭时间的参数是 log.roll.ms，它指定了多长时间之后日志片段可以被关闭。就像 log.retention.bytes 和 log.retention.ms 一样，log.segment.bytes和 log.roll.ms 并不互斥。日志片段会在大小或时间达到上限时被关闭，就看哪个条件先得到满足。在默认情况下，log.roll.ms 没有设定值，所以使用 log.roll.hours 设定的默认值——168 小时，也就是 7 天。
  • 在使用基于时间的日志片段时，需要考虑并行关闭多个日志片段对磁盘性能的影响。如果多个分区的日志片段一直未达到大小的上限，就会出现这种情况。这是因为 broker 在启动时会计算日志片段的过期时间，一旦满足条件，就会并行关闭它们，尽管它们的数据量可能很少。
• message.max.bytes：限制单条消息的大小，默认值是 1 000 000，也就是 1 MB。如果生产者尝试发送超过这个大小的消息，那么不仅消息不会被 broker 接收，还会收到 broker 返回的错误信息。与其他 broker 配置参数一样，这个参数指的是压缩后的消息大小，也就是说，消息的实际大小可以远大于 message.max.bytes，只要压缩后小于这个值即可。
  • 生产者客户端的 max.request.size 和消费者客户端的 fetch.max.bytes 需要与服务器端设置的消息大小保持一致。如果参数的值比 message.max.bytes 小，那么消费者就无法读取比较大的消息，进而造成阻塞，无法继续处理消息。 在配置 broker 的 replica.fetch.max.bytes 参数时，也遵循同样的原则，否则 broker 会接收此消息，但无法将此消息复制出去，从而造成数据丢失。

num.partitions=3
default.replication.factor=3
min.insync.replicas=1
log.retention.ms=86400000 # 一天
# log.retention.bytes=1073741824 # 1 GB
log.segment.bytes=1073741824 # 1 GB
# log.roll.ms=86400000 # 一天
message.max.bytes=1073741824 # 1 GB
replica.fetch.max.bytes=1073741824 # 1 GB

6.配置 Kafka 集群

• 使用集群的最大好处是可以跨服务器进行负载均衡，再者就是可以使用复制功能来避免因单点故障造成的数据丢失。

（1）broker 的数量

• 磁盘及单个 broker 的可用空间：如果整个集群需要保留 10 TB 数据，每个 broker 可以存储 2 TB，那么至少需要 5 个 broker。另外，如果增加了复制系数，那么至少还需要多一倍的空间，具体取决于配置的复制系数是多少。复制系数是指一个 broker 的数据需要被复制到多少个其他 broker 上。如果这个集群配置的复制系数是 2，那么将需要至少 10 个 broker。
• 集群处理请求的能力：假设你有一个包含 10 个 broker 的集群，集群中有 100 多万个副本（也就是说，有 500 000个分区，复制系数为 2），如果分区分布得均衡，那么每个 broker 大约有 100 000 个副本。这可能会导致生产者、消费者和控制器队列出现瓶颈。在过去，官方的建议是每个 broker的分区副本不超过4000个，每个集群的分区副本不超过200 000个。随着集群效率的提升，Kafka 可以被扩展到更大的规模。目前，建议每个 broker 的分区副本不超过 14 000 个，每个集群的分区副本不超过 100 万个。
• CPU：关注有多少个客户端和消费者群组，并进行相应的伸缩，以满足它们的需求。
• 网络：关注网络接口容量，以及它们是否能够在有多个消费者或流量与数据保留策略不一致（比如高峰期的流量激增）的情况下处理客户端流量。如果一个 broker 的网络接口在高峰期使用了 80% 的容量，并且有两个消费者，那么除非有两个 broker，**否则消费者将无法及时读取高峰流量。如果集群启用了复制功能，那么就又多了一个消费者。**为了解决因磁盘吞吐量不足或可用系统内存较少而引起的性能问题，你可能需要通过增加更多的 broker 来扩展集群。

（2）broker配置

• 要让一个 broker 加入集群，只需要修改两个配置参数。首先，所有 broker 都必须配置相同的 zookeeper.connect，这个参数指定了用于保存元数据的 ZooKeeper 的群组和路径。其次，每个 broker 都必须为 broker.id 指定唯一的值。如果两个 broker 使用相同的 broker.id，那么第二个 broker 将无法启动。还可以为集群配置其他的一些参数，特别是那些用于控制数据复制的参数。

7.其它优化

（1）垃圾回收器

• Java 7 为我们带来了垃圾回收器 G1GC。其在 JDK 8 和 JDK 11 中有了显著改进。建议将G1GC 作为 Kafka 的默认垃圾回收器。在应用程序的整个生命周期中，G1GC 会根据工作负载进行自我调节，并且停顿时间是恒定的。它可以轻松收集大块堆内存，把堆内存分为若干小块区域，并不是每次都回收整个堆空间。
• Kafka 使用堆内存以及处理垃圾对象的效率是比较高的，所以可以把这些参数设置得小一些。如果一台服务器有 64 GB 内存，并使用 5 GB 堆内存来运行 Kafka，那么可以将 MaxGCPauseMillis 设置为 20 毫秒，将 InitiatingHeapOccupancyPercent 设置为 35，让垃圾回收比默认的早一些启动。
• Kafka 是在 G1GC 之前发布的，因此，它默认使用的是 CMS（并发标记和清除）垃圾回收器，以确保与所有的 JVM 兼容。现在，强烈建议使用 G1GC，只要 Java 版本大于等于 1.8
  即可。

export KAFKA_JVM_PERFORMANCE_OPTS="-server -Xmx6g -Xms6g
-XX:MetaspaceSize=96m -XX:+UseG1GC
-XX:MaxGCPauseMillis=20 -XX:InitiatingHeapOccupancyPercent=35
-XX:G1HeapRegionSize=16M -XX:MinMetaspaceFreeRatio=50
-XX:MaxMetaspaceFreeRatio=80 -XX:+ExplicitGCInvokesConcurrent"

sh kafka-server-start.sh -daemon /usr/local/kafka/config/server.properties

（2）数据中心布局

• Kafka 可以将新创建的分区分配给部署在不同机架上的 broker（机架感知），确保单个分区的副本不会都位于同一个机架。要做到这一点，必须正确配置每个 broker 的 broker.rack 参数。
• 建议使用集群均衡工具来保持分区的机架感知，比如 Cruise Control。恰当配置这个属性有助于确保持续的机架感知。
• 最好把集群的 broker 安装在不同的机架上，至少不要让它们共享可能出现单点故障的基础设施，比如电源和网络。也就是说，部署服务器需要至少两个电源连接（两个不同的回路）和两个网络交换器（保证可以进行无缝的故障切换）。除了使用两个电源连接，最好把 broker 安装在不同的机架上，因为随着时间的推移，机架也需要维护，这个时候机器需要离线（比如移动机器或重新连接电源）。

（3）ZooKeeper

• Kafka 使用 ZooKeeper 保存 broker、主题和分区的元数据。只有当消费者群组成员或 Kafka 集群本身发生变化时才会向 ZooKeeper 写入数据。这些流量通常很小，所以没有必要为单个 Kafka 集群使用专门的 ZooKeeper 群组。实际上，有很多 Kafka 部署环境使用单个 ZooKeeper 群组来保存多个 Kafka 集群的元数据（正如本章之前所描述的那样，每个 Kafka集群使用一个单独的 chroot 路径）。
• 旧版本 Kafka 中，除了 broker，消费者也利用 ZooKeeper 来保存消费者群组的信息和已消费的主题的信息，并定期提交分区的偏移量（为了实现消费者群组内的故障转移）。在 0.9.0.0 版本中，消费者接口发生了变化，我们可以直接在 Kafka 中完成上述的这些任务。在每一个 2.x 版本中，我们都看到 Kafka 的一些路径移除了对 ZooKeeper 的依赖。现在，Kafka 管理工具不再需要连到ZooKeeper，它们可以直接连到 Kafka 集群完成主题创建、动态配置变更等操作。同样，很多以前使用 --zookeeper 选项的命令行工具现在改用了 --bootstrap-server。–zookeeper 选项仍然可以使用，但已经处于被弃用的状态。当将来 Kafka 不再需要通过连接 ZooKeeper 进行主题的创建、管理或消费时，这个选项将被移除。
• 虽然不建议使用 ZooKeeper来保存元数据，但消费者仍然可以选择是使用 ZooKeeper 还是 Kafka 来保存偏移量，还可以选择提交的时间间隔。如果消费者使用 ZooKeeper 来保存偏移量，那么每一个消费者会在每一个提交时间间隔内执行一次 ZooKeeper 写入操作。合理的偏移量提交时间间隔是 1分钟，因为如果有消费者发生故障，那么消费者群组将在这段时间内读取到重复消息。这些提交偏移量的操作可能会给 ZooKeeper 带来很大的流量，特别是在有很多消费者的集群中。这个问题需要引起注意。如果 ZooKeeper 群组不能轻松地处理这些流量，则可能需要使用更长的提交时间间隔。不管怎样，还是建议使用新版的 Kafka 消费者，并将偏移量提交到 Kafka，消除对 ZooKeeper 的依赖。
• 如果可以的话，除了让多个 Kafka 集群共享一个 ZooKeeper 群组，不建议再把 ZooKeeper 共享给其他应用程序。Kafka 对 ZooKeeper 的延迟和超时比较敏感，与 ZooKeeper 群组之间的一个通信中断都可能导致 Kafka 出现不可预测的行为。如果有多个 broker 与ZooKeeper 断开连接，那么它们就会离线，进而导致分区离线。这也会给集群控制器造成压力，因为在发生中断一段时间后，当控制器尝试关闭 broker 时，会表现出细微的异常。其他应用程序也会给 ZooKeeper 群组造成压力，它们可能会大量使用 ZooKeeper 或者执行一些不恰当的操作。所以，最好让其他应用程序使用自己的 ZooKeeper 群组。