关于kafka:Kafka4kafka生产环境规划部署

摘要

咱们上一节解说了 Kafka 架构 - 底层原理: 从根底的 partition 外面的 offset 引申到 LEO 和 HW; 以及对于 Leader partition 跟 follower partition 而言, 他们的 LEO 跟 HW 是如何更新的的？以及高水位 HW 在 leader 选举切换时候，存在的数据失落跟数据不统一问题，以及通过 leader epoch 版本号的概念引入解决下面问题，因为 HW 的变更还跟 leadder partition 外面的 ISR 列表有关联，又解说了:ISR 列表设计的 0.9.x 之前版本形式以及 0.9.x 之后的版本通过 follower 音讯数比 leader 音讯数目少形式更新 LSR 列表，以及依据工夫来更新 ISR 列表。而后这些都解说结束之后，咱们解说了 Kafka 的日志是如何存储存储到对应目录列表的，而后怎么基于存储日志疾速找到对应 offset 的数据的二分查找法；这些都解说结束之后，解说了 Kafka 底层集群外面节点如何通信？以及集群节点如果进行管控，Leader 选举，线上解体以及数据清理。

这些理解之后，咱们便可进行 Kafka 理论线上生产规划以及搭建。次要阐明以下：
1、基于理论业务进行 kafka 集群如何布局？须要思考哪些问题？
2、Kafka 集群生产参数配置：包含 Kafka 本人内核参数、操作系统参数、JVM 和 GC 参数。
3、Kafka 单点搭建用于测试、Kafka 集群搭建
4、具备环境之后，咱们须要进行一些运维工作:Kafka 的测试。

思维导图

内容

1、基于理论业务进行 Kafka 集群生产规划(电商零碎每日 10 亿数据量)

基于理论生产业务的时候, 咱们布局 kafka 集群的时候, 思考方向是: 基于每天数据量计算出零碎的大略 QPS, 依据 QPS 能够计算出须要多少台物理机器；依据每天数据量预估整体数据量大小，在联合 Kafka 本身的原理，保留工夫长短，以及多冗余正本机制避免数据失落来预估整体数据量大小来布局机器测硬盘大小。这些都做完之后，就须要具体到一台物理机或者虚拟机了, 次要思考:cpu(每台机器的线程数)、内存(Kafka 本身底层原理)、磁盘(kafka 底层原理剖析)、网络(依据数据量)

1、QPS

基于 10 亿数据量:10 亿数据量 依据 2 - 8 准则,80% 的数据是在非凌晨 0 - 8 点的 16 小时产生，并且其中 80% 的申请都是在 20% 的工夫内，所以:100.80.8= 6 亿数据在:1620%= 3 小时；QPS=600000000/360*60 近似为 6 万。

2、存储量

总存储量 = 工夫 * 每天存储量。
每天 10 亿数据量，咱们假如每条数据大小大略是 1kb, 其实很多时候均匀是达不到 1kb 的数据。咱们粗略预计下: 假如一个英文字母是 1 个字节 byte,1 个中文汉子是 2 个字节 byte, 那么 1kb 能够包容 1024 个英文字符 /512 个汉子；10 亿数据量, 是 10 亿 kb 大小数据，kafka 假如咱们依照 2 个正本，数据默认保留 7 天，则 整体数据量 =1027=140 亿 kb 近似依照 12TB 数据量算。

3、物理机

1、物理机数量

物理机数量个别依据 qps 计算，因为 qps 是 6 万左右，个别一台机器大略抗并发是 3 万左右，因为在机器资源短缺状况下，个别为了应答秒杀这种场景, 计算的 qps= 总体 qps*30%; 所以咱们的最大 qps 为 20 万 qps 左右，所以咱们的机器大略在 6 - 7 台机器，这里咱们抉择 6 台机器。

2、物理机 cpu

经验值
是状况而定，咱们为了减少 kafka 的吞吐量的话, 咱们将会把解决申请的线程数量调整为默认的 2～3 倍

Kafka Broker 过程运行起来线程数预估
依据 Kafka 多路复用的 Reeactor 模型解决申请时候，是须要一个 Accepptor 线程，Processor 线程用来将申请写入到申请队列，默认是 3 个；而后具体进行申请解决的是一个 Handler 线程池，默认线程数量是 8 个；以及后盾数据清理线程几 controller 选举等线程。依照 2～3 倍准则, 咱们的线程数:24-36；而后加上日志清理线程，或者 Controller 感知线程等, 咱们将预估线程忙碌的时候，线程数量为一两百。

cpu 核跟线程数的经验值
1、CPU 4 核，一般来说几十个线程，在高峰期 CPU 简直都快打满了，
2、CPU 8 核，也就可能比拟拮据的撑持几十个线程忙碌的工作，
3、个别是倡议 16 核，靠谱，基本上能够达到一两百线程的工作。

broker 个别都会启动几十个甚至上百个线程，大家看过 broker 端的原理了，各种解决申请的线程，后盾线程，几十个线程频繁工作，个别都倡议是 16 核 CPU，甚至 32 核 CPU

3、物理机内存

因为 Kafka 写数据是: 写入 os_cache+ 磁盘程序写，kafka 读取数据: 零拷贝 + 页缓冲技术 (os_cache)
所以咱们发现 Kafka 是大量依附 os_cache，而不是 JVM 外面的内存；所以咱们布局机器内存的时候，次要是思考 os_cache 内存大小，辅助思考下机器内存大小。

kafka jvm 堆内存，你感觉须要很大吗？
理解了 kafka 原理之后，kafka 并没有在本人的 jvm 堆内存里放入过多的数据，他跟 rabbitmq 是不一样的，rabbitmq 是数据过去优先写入 jvm 堆内存里去缓冲一下，肯定工夫之后，rabbitmq 再一次性把 jvm 堆内存里的缓冲一批数据给刷入磁盘中。rabbitmq 那种机制跟原理的话会导致 jvm 堆内存中寄存大量的数据，须要给 jvm 堆内存开拓比拟大的空间了。

然而 kafka 正好相同的，他接管到了这个数据之后不是间接写 jvm 堆内存的，而是采纳本人的二进制紧凑的数据格式，给写入到磁盘文件里去，然而是先写入 os cache（操作系统治理的一块内存缓冲空间）kafka 并没有应用过多的 jvm 堆内存空间，所以的话你不须要给 kafka jvm 堆内存调配过大的空间，基本上来说几个 G 就够了。

充分利用 os_cache 进步性能：
咱们思考下 os_cache 调配多大；
准则: broker 治理的每个 parition 的最新正在写的日志段文件的数据都能够驻留在 os cache 中，这样能够保障每个 parition 正在写的数据，最有可能被生产的数据，就能够间接从 os cache 里来读了。

剖析:
假如整个 Kafka 的集群外面有 100 个 topic，假如每个 topic 是 6 个 partition，每个 partiton 是 2 个正本，然而其实不会从 partition 的正本读取数据，你只须要思考的是 leader partirion, 所以一共 100 * 6 =600 个 leader partition，均匀到 6 台机器下来，假如每台机器是放 100 个 partition，每个 partition 的最新正在写的日志段文件的大小是默认的 1GB.

所以说单台机器上，最新的正在写的日志段文件的大小 100 个 partition 1GB 的日志段文件 = 600GB。如果最佳的状况，单台机器能够有 600GB 的内存的给 os cache 的话，就能够每个 partiton 最新的数据都在 os cache 里，比如说一个日志段文件有 1GB，可能都对应几百万条数据了，这些数据咱们也不会变频繁生产，最多也就是 1%-10% 数据生产，所以 600(1%-10%)也就是 6G 以上所以咱们大略 64G 的机器内存即可。

JVM 的堆内存
kafka 本身的 jvm 是用不了过多堆内存的，因为 kafka 设计就是躲避掉用 jvm 对象来保留数据，防止频繁 fullgc 导致的问题，所以个别 kafka 本身的 jvm 堆内存，调配个 6G 左右就够了，剩下的内存全副留给 os cache

4、物理机网络带宽

生产环境的机器布局，无论是 MySQL、Redis、RocketMQ、Elasticsearch、Kafka、Hadoop、Flink，Yarn，其实布局的思路都是相似的，思考的其实是从技术实质和底层的原理登程来思考：
1、申请量有多大、
2、数据量有多大、
3、内存应该调配多大（须要理解底层工作机制，怎么应用内存的？）、
4、线程数量有多少、
5、网络数据传输量有多大（个别决定了你的网卡大小抉择）

当初个别就是千兆网卡（1GB / s），还有万兆网卡（10GB / s），网卡决定了什么呢？比方你要计算一下，kafka 集群之间，broker 和 broker 之间是会做数据同步的，因为 leader 要同步数据到 follower 下来，他们是在不同的 broker 机器上的，broker 机器之间会进行频繁的数据同步，传输大量的数据。

所以在这种状况下，须要计算下每秒两台 broker 机器之间大略会传输多大的数据量？高峰期每秒大略会涌入 6 万条数据，假如每一条数据大小是 1kb，大略来说每秒就是 60mb/ s 的数据量，每秒 60mb/ s 的湖，他会在 broker 之间来传输，你就算多估算一些，每秒大略也就传输个几百 mb/ s 的数据，就足够了。

实际上每台机器能用的网卡的带宽还达不到极限，因为 kafka 只能用其中一部分的带宽资源，比方 700mb/s，然而个别不能允许 kafka 占用这么多带宽，因为防止说占用这么多带宽，万一再多一点，就容易把网卡打满

所以说，个别限度 kafka 每秒带宽资源就是 300mb / s，如果你给物理机应用的千兆网卡，那么其实他每秒最多传输数据是几百 mb，是够了，可能也就是几十 mb，或者一两百 mb，基本上都够了，能够足够传输。

5、物理机磁盘

磁盘类型
SSD 磁盘的性能相比于机械硬盘的话次要在磁盘随即写下面，在磁盘程序写的状况下, 磁盘程序写跟随即写差不多，所以因为 Kafka 是基于 os cahce+ 磁盘程序写机制, 因为磁盘程序写的性能跟写内存差不多，并且 SSD 的在磁盘随机写上会比机械硬盘成果好, 在磁盘程序写上 SSD 跟机械硬盘差不多, 所以咱们应用机械硬盘即可

磁盘大小
总共 13T 数据, 因为很多数据是没有 1kb 大小的，所以咱们预估是 12TB 数据，6 台机器，则: 每台机器加上 2 快 1TB 的磁盘, 总共 2TB。

2、Kafka 集群生产参数设置

1、内核参数

内存参数的话次要是 Kafka Broker 利用的本身配置文件, 上面是一些惯例参数：

broker.id：Kafka 利用的惟一标示；这个是每个 broker 都必须本人设置的一个惟一 id。

log.dirs：这个极为重要，kafka 的所有数据就是写入这个目录下的磁盘文件中的，如果说机器上有多块物理硬盘，那么能够把多个目录挂载到不同的物理硬盘上，而后这里能够设置多个目录，这样 kafka 能够把数据写的压力扩散到多块物理硬盘，多个硬盘的磁头能够并行写，这样能够晋升吞吐量。

zookeeper.connect： 因为你要晓得的是，Kafka 他肯定要去应用 zookeeper 的。所以须要配置底层 ** 这个是连贯 kafka 底层的 zookeeper 集群的地址。

Listeners：这是 broker 监听客户端发动申请的端口号，在 Kafka 新版本外面，通过 listeners 去配置 broker。通过哪个端口去监听客户端发送过去申请的。

unclean.leader.election.enable：1.0 版本之前都是 true：意思是容许非 ISR 列表的 follower 选举为新的 leader。这个是什么意思呢？比如说你的一个 follower，他可能落后于这个 leader。然而呢？leader 此时挂了。之前老版本也是容许非 ISR 列表外面的 follower 选举为 leader。这个时候可能会失落一些数据这样子。1.0 之后。默认设置为 false: 意思就是只能选举 ISR 列表里的 follower 成为新的 leader。unclean 指代不洁净的 leader 选举，也就是指代不在 ISR 列表外面的 follower 选举。

delete.topic.enable**：默认 true，容许删除 topic。

log.retention.hours**：设置一下日志要保留数据多少个小时，这个就是底层的磁盘文件，默认保留 7 天的数据。

log.retention.bytes: 这个设置如果分区的数据量超过了这个限度，就会主动清理数据，默认 - 1 不依照这个策略来清理，这个个别不罕用

min.insync.replicas： 这个跟 acks=- 1 配合起来应用，意思就是说必须要求 ISR 列表里有几个 follower，而后 acks=- 1 就是写入数据的时候，必须写入这个数量指定的 follower 才能够，个别是能够思考如果一个 leader 两个 follower，三个正本，那么这个设置为 2，能够容忍一台机器宕机，保障高可用，然而写入数据的时候必须 ISR 里有 2 个正本，而且必须正本都写入才算胜利。

如果你就一个 leader 和 follower，也就是说你只有双正本，而后 min.insync.replicas = 2。如果有一台机器宕机，导致 follower 没了，此时 ISR 列表里就一个 leader 的话，你就不能写入了, 因为你设置的参数:min.insync.replicas = 2 也就是要求 ISR 列表外面至多有哦 2 个正本。然而咱们将参数：min.insync.replicas = 2 设置为 2 也是有情理的，如果你只有一个正本了，此时还能够写入数据的话，就会导致写入到 leader 之后，万一 leader 也宕机了，此时数据必然失落。

一般来说为了防止数据占用磁盘空间过大，个别都是 kafka 设置双正本，一个 leader 和一个 follower 就够了，如果你设置的三正本的话, 数据在集群里要乘以 3 倍的空间来寄存，十分消耗磁盘空间，而且对你的整体集群的性能耗费也会更大。（因为 leader 收到写入数据时候，须要传给一个 follower 还是 2 个 follower，你要传给 2 个 follower 的花，你须要占用更多的网络带宽，而且你传给两个 follower 的话，必然会导致你须要生产更多的 cpu 执行）。会减轻更多数据资源的耗费，所以说：咱们 Kafak 集群部署的时候，应用双正本就能够了。

双正本的状况下，咱们就思考下是否容许数据失落？如果容许数据失落的话，将参数：min.insync.replicas=1，acks=-1（一条数据必须写入 ISR 里所有正本才算胜利），你写一条数据只有写入 leader 就算胜利了，不须要期待同步到 follower 才算写胜利。然而此时如果一个 follower 宕机了，你写一条数据到 leader 之后，如果 leader 也宕机，会导致数据的失落。

所以说，一般来说 在双正本的场景下，咱们为了防止数据失落，min.insync.replicas=2，acks=-1，每次写入必须保障 ISR 里 2 个正本都写胜利才能够，如果其中一个正本没了，会导致你就没法写入，必须阻塞期待 kafka 复原。这样就能够保证数据的不失落。然而对吞吐量有肯定影响。

业务场景思考：

你是要计算为客户筹备的财务数据报表，十分严格的数据必须精准的话，精准到每一分钱，你还是得设置成 2；

如果做日志剖析的话，你将其设置为 1 即可。只须要写入到一个正本之后，就认为写入胜利。

num.network.threads： 这个是负责转发申请给理论工作线程的网络申请解决线程的数量（也就是 processor 线程数），这个默认是 3，高负载场景下能够设置大一些。

num.io.threads：这个是管制理论解决申请的 Handler 线程池外面 线程数量，默认是 8，高负载场景下能够设置大一些。

要本人做一些压测，在不同的线程数量的条件下，对整体的生产和生产的吞吐量别离是多少，还得看一下线程越多的时候，对 CPU 的简单有多大，整体看一下，如果减少一些线程，吞吐量晋升很多，而且 CPU 负载还能承受。

message.max.bytes：（这个肯定要设置，不然的话，写入数如果比拟大的话，写入数据将会写失败）这个是 broker 默认能承受的音讯的最大大小，默认是 977kb，太小了，能够设置的大一些，个别倡议设置大一些，很多大音讯可能大到几 mb，这个设置个 10mb，能够思考。

下面是一些比拟惯例的参数：，除了下面的参数外，咱们还须要思考 os cache 刷新数据到磁盘的参数。

log.flush.interval.messages:指代 os cache 中打到多少数据时候，刷新数据到磁盘。
log.flush.interval.ms: 指代每多长时间刷新数据到磁盘。

线上的罕用的一个配合，高负载高吞吐的状况下，倡议设置为 1 分钟

2、操作系统参数

咱们当初须要设置操作系统参数：

文件个数预计： kafka 会频繁的创立和批改文件，Kafka 上文件的数量大概是：分区数量 （分区总大小 / 日志段大小） 3(.index .timeindex .log)；

比方 一个 broker上大略有 100 个分区，每个分区大略是 10G 的数据，日志段大小是默认的 1G，那么就是 100 10（分区里的日志段文件数量） 3 = 3000 个文件

这个阐明 Kafka 会频繁地在机器上创立文件，所以你须要设置下一下参数：把文件描述符开大一点。这样的话能够容许 Kafka 创立大量的文件。

ulimit -n 100000： 这个能够设置很大的描述符限度，容许创立大量的文件

磁盘 flush 工夫：默认是 5 秒，默认状况下，Kafka 是先把数据写入到 os cache; 而后每隔一段时间，会将 os cache 外面的数据刷入到磁盘外面。os cache 刷入磁盘，能够设置为几分钟，比方 1 分钟才刷入磁盘，这样能够大幅度晋升单机的吞吐量。

参考: https://www.cnblogs.com/zengk…。

sysctl -a | grep dirty

vm.dirty_background_bytes = 0
vm.dirty_background_ratio = 5  
vm.dirty_bytes = 0
vm.dirty_expire_centisecs = 3000
vm.dirty_ratio = 10
vm.dirty_writeback_centisecs = 500

vm.dirty_writeback_centisecs： 下面设置的是 500，意思是每隔 5 秒唤醒一次刷磁盘的线程；cat /proc/sys/vm/dirty_writeback_centisecs 查看这个值，默认个别是 500（单位是 1 /100 秒）。这个参数示意 5s 的工夫 pdflush 就会被唤起去刷新脏数据。

vm.dirty_expire_centisecs： 下面设置的是 3000 代表：os cache 里的数据在 3 秒后会被刷入磁盘；这个能够设置大一些，设置为：1 分钟~2 分钟都能够，特地大规模的企业和公司，如果你能够接管机器宕机的时候，数据适当能够失落一些，kafka 里的数据能够适当失落一些，然而为了晋升集群的吞吐量的话。能够设置为 1 分钟~2 分钟都能够，然而机器宕机的话，os cahce 数据清空，而后数据失落。
你大数据分析类的利用的话，次要是出一些数据报表，不要求数据 100% 精准的话，容许适当丢一些数据，此时的话，这里给他调大一些是没问题的。

3、JVM 跟 GC 参数

Kafka 除了本人内核参数外，还须要思考下 JVM 参数设置：

批改 bin/kafka-start-server.sh 中的 jvm 设置

export KAFKA_HEAP_OPTS=”-Xmx6g -Xms6g -XX:MetaspaceSize=96m -XX:+UseG1GC -XX:MaxGCPauseMillis=20 -XX:InitiatingHeapOccupancyPercent=35 -XX:G1HeapRegionSize=16M -XX:MinMetaspaceFreeRatio=50 -XX:MaxMetaspaceFreeRatio=80”

-Xmx6g -Xms6g: 这个是设置堆内存大小的。这个肯定要配置的。
-XX:+UseG1GC： 垃圾回收器应用 G1 垃圾回收器。
-XX:MetaspaceSize: 参考 https://www.jianshu.com/p/b44…

3、Kafka 集群搭建

1、zk 集群搭建

zookeeper 进行集群或者单点搭建的话，参考另一篇文章:https://segmentfault.com/a/11…

未完待续 ……

2、Kafka 集群搭建

4、Kafka 根本测试