开源流处理平台Kafka教程讲解 Kafka性能好在什么地方？

Kafka基础

消息系统的作用

应该大部分小伙伴都清楚，用机油装箱举个例子。

所以消息系统就是如上图我们所说的仓库，能在中间过程作为缓存，并且实现解耦合的作用。

引入一个场景，我们知道中国移动，中国联通，中国电信的日志处理，是交给外包去做大数据分析的，假设现在它们的日志都交给了你做的系统去做用户画像分析。

按照刚刚前面提到的消息系统的作用，我们知道了消息系统其实就是一个模拟缓存，且仅仅是起到了缓存的作用而并不是真正的缓存，数据仍然是存储在磁盘上面而不是内存。

Topic主题

Kafka学习了数据库里面的设计，在里面设计了topic(主题)，这个东西类似于关系型数据库的表。

此时我需要获取中国移动的数据，那就直接监听TopicA即可。

Partition分区

kafka还有一个概念叫Partition(分区)，分区具体在服务器上面表现起初就是一个目录，一个主题下面有多个分区，这些分区会存储到不同的服务器上面，或者说，其实就是在不同的主机上建了不同的目录。这些分区主要的信息就存在了.log文件里面。跟数据库里面的分区差不多，是为了提高性能。

至于为什么提高了性能，很简单，多个分区多个线程，多个线程并行处理肯定会比单线程好得多。

Topic和partition像是HBASE里的table和region的概念，table只是一个逻辑上的概念，真正存储数据的是region，这些region会分布式地存储在各个服务器上面，对应于Kafka，也是一样，Topic也是逻辑概念，而partition就是分布式存储单元。这个设计是保证了海量数据处理的基础。我们可以对比一下，如果HDFS没有block的设计，一个100T的文件也只能单独放在一个服务器上面，那就直接占满整个服务器了，引入block后，大文件可以分散存储在不同的服务器上。

注意：

分区会有单点故障问题，所以我们会为每个分区设置副本数;

分区的编号是从0开始的。

Producer - 生产者

往消息系统里面发送数据的就是生产者。

Consumer - 消费者

从Kafka里读取数据的就是消费者。

Message - 消息

Kafka里面的我们处理的数据叫做消息。

Kafka的集群架构

创建一个TopicA的主题，3个分区分别存储在不同的服务器，也就是broker下面。Topic是一个逻辑上的概念，并不能直接在图中把Topic的相关单元画出。

需要注意：Kafka在0.8版本以前是没有副本机制的，所以在面对服务器宕机的突发情况时会丢失数据，所以尽量避免使用这个版本之前的Kafka。

Replica - 副本

Kafka中的partition为了保证数据安全，所以每个partition可以设置多个副本。

此时我们对分区0，1，2分别设置3个副本(其实设置两个副本是比较合适的)。

而且其实每个副本都是有角色之分的，它们会选取一个副本作为leader，而其余的作为follower，我们的生产者在发送数据的时候，是直接发送到leader partition里面，然后follower partition会去leader那里自行同步数据，消费者消费数据的时候，也是从leader那去消费数据的。

Consumer Group - 消费者组

我们在消费数据时会在代码里面指定一个group.id，这个id代表的是消费组的名字，而且这个group.id就算不设置，系统也会默认设置。

conf.setProperty( "group.id", "tellYourDream")

我们所熟知的一些消息系统一般来说会这样设计，就是只要有一个消费者去消费了消息系统里面的数据，那么其余所有的消费者都不能再去消费这个数据。可是Kafka并不是这样，比如现在consumerA去消费了一个topicA里面的数据。

consumerA:

group.id = a

consumerB:

group.id = a

consumerC:

group.id = b

consumerD:

group.id = b

再让consumerB也去消费TopicA的数据，它是消费不到了，但是我们在consumerC中重新指定一个另外的group.id，consumerC是可以消费到topicA的数据的。而consumerD也是消费不到的，所以在Kafka中，不同组可有唯一的一个消费者去消费同一主题的数据。

所以消费者组就是让多个消费者并行消费信息而存在的，而且它们不会消费到同一个消息，如下，consumerA，B，C是不会互相干扰的。

consumer group:a

consumerA

consumerB

consumerC

如图，因为前面提到过了消费者会直接和leader建立联系，所以它们分别消费了三个leader，所以一个分区不会让消费者组里面的多个消费者去消费，但是在消费者不饱和的情况下，一个消费者是可以去消费多个分区的数据的。

Controller

熟知一个规律：在大数据分布式文件系统里面，95%的都是主从式的架构，个别是对等式的架构，比如ElasticSearch。

Kafka也是主从式的架构，主节点就叫controller，其余的为从节点，controller是需要和ZooKeeper进行配合管理整个Kafka集群。

Kafka和ZooKeeper如何配合工作

Kafka严重依赖于ZooKeeper集群。所有的broker在启动的时候都会往ZooKeeper进行注册，目的就是选举出一个controller，这个选举过程非常简单粗暴，就是一个谁先谁当的过程，不涉及什么算法问题。

那成为controller之后要做啥呢，它会监听ZooKeeper里面的多个目录，例如有一个目录/brokers/，其他从节点往这个目录上注册(就是往这个目录上创建属于自己的子目录而已)自己，这时命名规则一般是它们的id编号，比如/brokers/0，1，2。

注册时各个节点必定会暴露自己的主机名，端口号等等的信息，此时controller就要去读取注册上来的从节点的数据(通过监听机制)，生成集群的元数据信息，之后把这些信息都分发给其他的服务器，让其他服务器能感知到集群中其它成员的存在。

此时模拟一个场景，我们创建一个主题(其实就是在ZooKeeper上/topics/topicA这样创建一个目录而已)，Kafka会把分区方案生成在这个目录中，此时controller就监听到了这一改变，它会去同步这个目录的元信息，然后同样下放给它的从节点，通过这个方法让整个集群都得知这个分区方案，此时从节点就各自创建好目录等待创建分区副本即可。这也是整个集群的管理机制。

加餐时间

Kafka性能好在什么地方？

顺序写

操作系统每次从磁盘读写数据的时候，需要先寻址，也就是先要找到数据在磁盘上的物理位置，然后再进行数据读写，如果是机械硬盘，寻址就需要较长的时间。

Kafka的设计中，数据其实是存储在磁盘上面，一般来说，会把数据存储在内存上面性能才会好。但是Kafka用的是顺序写，追加数据是追加到末尾，磁盘顺序写的性能极高，在磁盘个数一定，转数达到一定的情况下，基本和内存速度一致。

随机写的话是在文件的某个位置修改数据，性能会较低。

零拷贝

先来看看非零拷贝的情况。

可以看到数据的拷贝从内存拷贝到Kafka服务进程那块，又拷贝到socket缓存那块，整个过程耗费的时间比较高，Kafka利用了Linux的sendFile技术(NIO)，省去了进程切换和一次数据拷贝，让性能变得更好。

日志分段存储

Kafka规定了一个分区内的.log文件最大为1G，做这个限制目的是为了方便把.log加载到内存去操作。

这个9936472之类的数字，就是代表了这个日志段文件里包含的起始offset，也就说明这个分区里至少都写入了接近1000万条数据了。Kafka broker有一个参数，log.segment.bytes，限定了每个日志段文件的大小，最大就是1GB，一个日志段文件满了，就自动开一个新的日志段文件来写入，避免单个文件过大，影响文件的读写性能，这个过程叫做log rolling，正在被写入的那个日志段文件，叫做active log segment。

如果大家有看前面的两篇有关于HDFS的文章时，就会发现NameNode的edits log也会做出限制，所以这些框架都是会考虑到这些问题。

Kafka的网络设计

Kafka的网络设计和Kafka的调优有关，这也是为什么它能支持高并发的原因。

首先客户端发送请求全部会先发送给一个Acceptor，broker里面会存在3个线程(默认是3个)，这3个线程都是叫做processor，Acceptor不会对客户端的请求做任何的处理，直接封装成一个个socketChannel发送给这些processor形成一个队列，发送的方式是轮询，就是先给第一个processor发送，然后再给第二个，第三个，然后又回到第一个。消费者线程去消费这些socketChannel时，会获取一个个request请求，这些request请求中就会伴随着数据。

线程池里面默认有8个线程，这些线程是用来处理request的，解析请求，如果request是写请求，就写到磁盘里。读的话返回结果。

processor会从response中读取响应数据，然后再返回给客户端。这就是Kafka的网络三层架构。

所以如果我们需要对Kafka进行增强调优，增加processor并增加线程池里面的处理线程，就可以达到效果。request和response那一块部分其实就是起到了一个缓存的效果，是考虑到processor们生成请求太快，线程数不够不能及时处理的问题。

所以这就是一个加强版的reactor网络线程模型。

Kafka的生产集群部署

方案背景

假设每天集群需要承载10亿数据。一天24小时，晚上12点到凌晨8点几乎没多少数据。

使用二八法则估计，也就是80%的数据(8亿)会在16个小时涌入，而且8亿的80%的数据(6.4亿)会在这16个小时的20%时间(3小时)涌入。

QPS计算公式：640000000 ÷ (3x60x60) = 60000，也就是说高峰期的时候Kafka集群要扛住每秒6万的并发。

磁盘空间计算，每天10亿数据，每条50kb，也就是46T的数据。保存2个副本(在上一篇中也提到过其实两个副本会比较好，因为follower需要去leader那里同步数据，同步数据的过程需要耗费网络，而且需要磁盘空间，但是这个需要根据实际情况考虑)，46 * 2 = 92T，保留最近3天的数据。故需要 92 * 3 = 276T。

QPS方面

部署Kafka，Hadoop，MySQL……等核心分布式系统，一般建议直接采用物理机，抛弃使用一些低配置的虚拟机的想法。高并发这个东西，不可能是说，你需要支撑6万QPS，你的集群就刚好把这6万并发卡的死死的。假如某一天出一些活动让数据量疯狂上涨，那整个集群就会垮掉。

但是，假如说你只要支撑6w QPS，单台物理机本身就能扛住4~5万的并发。所以这时2台物理机绝对绝对够了。但是这里有一个问题，我们通常是建议，公司预算充足，尽量是让高峰QPS控制在集群能承载的总QPS的30%左右(也就是集群的处理能力是高峰期的3~4倍这个样子)，所以我们搭建的kafka集群能承载的总QPS为20万~30万才是安全的。所以大体上来说，需要5~7台物理机来部署，基本上就很安全了，每台物理机要求吞吐量在每秒4~5万条数据就可以了，物理机的配置和性能也不需要特别高。

磁盘方面

磁盘数量

需要5台物理机的情况，需要存储276T的数据，平均下来差不多一台56T的数据。这个具体看磁盘数和盘的大小。

SAS还是SSD

现在我们需要考虑一个问题：是需要SSD固态硬盘，还是普通机械硬盘？

SSD就是固态硬盘，比机械硬盘要快，那么到底是快在哪里呢？其实SSD的快主要是快在磁盘随机读写，就要对磁盘上的随机位置来读写的时候，SSD比机械硬盘要快。比如说MySQL这种就应该使用SSD了(MySQL需要随机读写)。比如说我们在规划和部署线上系统的MySQL集群的时候，一般来说必须用SSD，性能可以提高很多，这样MySQL可以承载的并发请求量也会高很多，而且SQL语句执行的性能也会提高很多。

因为写磁盘的时候Kafka是顺序写的。机械硬盘顺序写的性能机会跟内存读写的性能是差不多的，所以对于Kafka集群来说其实使用机械硬盘就可以了。如果是需要自己创业或者是在公司成本不足的情况下，经费是能够缩减就尽量缩减的。

内存角度

JVM非常怕出现full gc的情况。Kafka自身的JVM是用不了过多堆内存的，因为Kafka设计就是规避掉用JVM对象来保存数据，避免频繁full gc导致的问题，所以一般Kafka自身的JVM堆内存，分配个10G左右就够了，剩下的内存全部留给OS cache。

那服务器需要多少内存呢。我们估算一下，大概有100个topic，所以要保证有100个topic的leader partition的数据在操作系统的内存里。100个topic，一个topic有5个partition。那么总共会有500个partition。每个partition的大小是1G(在上一篇中的日志分段存储中规定了.log文件不能超过1个G)，我们有2个副本，也就是说要把100个topic的leader partition数据都驻留在内存里需要1000G的内存。

我们现在有5台服务器，所以平均下来每天服务器需要200G的内存，但是其实partition的数据我们没必要所有的都要驻留在内存里面，只需要25%的数据在内存就行，200G * 0.25 = 50G就可以了(因为在集群中的生产者和消费者几乎也算是实时的，基本不会出现消息积压太多的情况)。所以一共需要60G(附带上刚刚的10G Kafka服务)的内存，故我们可以挑选64G内存的服务器也行，大不了partition的数据再少一点在内存，当然如果能够提供128G内存那就更好。

CPU core

CPU规划，主要是看你的这个进程里会有多少个线程，线程主要是依托多核CPU来执行的，如果你的线程特别多，但是CPU核很少，就会导致你的CPU负载很高，会导致整体工作线程执行的效率不太高，上一篇的Kafka的网络设计中讲过Kafka的Broker的模型。acceptor线程负责去接入客户端的连接请求，但是他接入了之后其实就会把连接分配给多个processor，默认是3个，但是一般生产环境建议大家还是多加几个，整体可以提升kafka的吞吐量比如说你可以增加到6个，或者是9个。另外就是负责处理请求的线程，是一个线程池，默认是8个线程，在生产集群里，建议大家可以把这块的线程数量稍微多加个2倍~3倍，其实都正常，比如说搞个16个工作线程，24个工作线程。

后台会有很多的其他的一些线程，比如说定期清理7天前数据的线程，Controller负责感知和管控整个集群的线程，副本同步拉取数据的线程，这样算下来每个broker起码会有上百个线程。根据经验4个CPU core，一般来说几十个线程，在高峰期CPU几乎都快打满了。8个CPU core，也就能够比较宽裕的支撑几十个线程繁忙的工作。所以Kafka的服务器一般是建议16核，基本上可以hold住一两百线程的工作。当然如果可以给到32 CPU core那就最好不过了。

网卡

现在的网基本就是千兆网卡(1GB / s)，还有万兆网卡(10GB / s)。kafka集群之间，broker和broker之间是会做数据同步的，因为leader要同步数据到follower上去，他们是在不同的broker机器上的，broker机器之间会进行频繁的数据同步，传输大量的数据。那每秒两台broker机器之间大概会传输多大的数据量？

高峰期每秒大概会涌入6万条数据，约每天处理10000个请求，每个请求50kb，故每秒约进来488M数据，我们还有副本同步数据，故高峰期的时候需要488M * 2 = 976M/s的网络带宽，所以在高峰期的时候，使用千兆带宽，网络还是非常有压力的。

综上描述

10亿数据，6w/s的吞吐量，276T的数据，5台物理机

硬盘：11(SAS) * 7T，7200转

内存：64GB/128GB，JVM分配10G，剩余的给os cache

CPU：16核/32核

网络：千兆网卡，万兆更好

Kafka的集群搭建

进到Kafka的config文件夹下，会发现有很多很多的配置文件，可是都不需要你来修改，你仅仅需要点开一个叫作server.properties的文件就够了。

【broker.id】

每个broker都必须自己设置的一个唯一id，可以在0~255之间

【log.dirs】

这个极为重要，Kafka的所有数据就是写入这个目录下的磁盘文件中的，如果说机器上有多块物理硬盘，那么可以把多个目录挂载到不同的物理硬盘上，然后这里可以设置多个目录，这样Kafka可以数据分散到多块物理硬盘，多个硬盘的磁头可以并行写，这样可以提升吞吐量。ps：多个目录用英文逗号分隔

【zookeeper.connect】

连接Kafka底层的ZooKeeper集群的

【Listeners】

broker监听客户端发起请求的端口号，默认是9092

【num.network.threads】默认值为3

【num.io.threads】默认值为8

细心的朋友们应该已经发现了，这就是上一篇我们在网络架构上提到的processor和处理线程池的线程数目。

所以说掌握Kafka网络架构显得尤为重要。

现在你看到这两个参数，就知道这就是Kafka集群性能的关键参数了

【unclean.leader.election.enable】

默认是 false，意思就是只能选举ISR列表里的follower成为新的leader，1.0版本后才设为 false，之前都是 true，允许非ISR列表的follower选举为新的leader

【delete.topic.enable】

默认 true，允许删除topic

【log.retention.hours】

可以设置一下，要保留数据多少个小时，这个就是底层的磁盘文件，默认保留7天的数据，根据自己的需求来就行了

【min.insync.replicas】

acks=-1(一条数据必须写入ISR里所有副本才算成功)，你写一条数据只要写入leader就算成功了，不需要等待同步到follower才算写成功。但是此时如果一个follower宕机了，你写一条数据到leader之后，leader也宕机，会导致数据的丢失。

因为实际的集群搭建说真的没有太大难度，所以搭建的过程就不详细展开了，网上应该很多相关资料。

Kafka的简单集群操作

在操作Kafka集群的时候，不同的Kafka版本命令的写法是不一样的，所以其实如果需要了解一下，推荐直接到官网去查看。

上一篇 时也有提到说Kafka在0.8版本以前存在比较大的问题，1.x的算是目前生产环境中使用较多的版本。

在quickStart就能看到相关的命令，比如：

创建主题

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

将该命令修改一下

zookeeper localhost:2181 --replication-factor 2 --partitions 2 --topic tellYourDream

这时候就是zookeeper的地址为localhost:2181

两个分区，两个副本，一共4个副本，topic名称为“tellYourDream”了

还得注意，一般来说设置分区数建议是节点的倍数，这是为了让服务节点分配均衡的举措。

查看主题

bin/kafka-topics.sh --list --zookeeper localhost:2181

生产信息

bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test

This is a message

This is another message

消费信息

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

This is a message

This is another message

这里有个细节需要提及一下，就是我们0.8版本的Kafka找的是ZooKeeper，ZooKeeper上确实是也存在着元数据信息。

不过这存在着一些问题，ZooKeeper本身有一个过半服务的特性，这是一个限制，过半服务是指任何的请求都需要半数节点同意才能执行。每次有写请求，它都要投票，因为它要保持数据的强一致性，做到节点状态同步，所以高并发写的性能不好。不适合做高并发的事。ZooKeeper是Kafka存储元数据和交换集群信息的工具，主要是处理分布式一致性的问题。

集群测试

下面的命令就是生产50W条数据，每条数据200字节，这条命令一运行就会产生一条报告，可以很直观的看到集群性能，看不懂的情况搜索引擎也可以很好地帮助你解决问题。

每次消费2000条，集群没跑挂那就稳妥了。

KafkaManager

KafkaManager使用scala写的项目，非常不错。使用方法可以通过搜索引擎查找。

安装步骤可以参考：https://www.cnblogs.com/dadonggg/p/8205302.html

安装好了之后可以使用jps命令查看一下，会多出一个名字叫做ProdServerStart的服务。

功能介绍：

管理多个Kafka集群

便捷的检查Kafka集群状态(topics，brokers，备份分布情况，分区分布情况)

选择你要运行的副本

基于当前分区状况进行

可以选择topic配置并创建topic(0.8.1.1和0.8.2的配置不同)

删除topic(只支持0.8.2以上的版本并且要在broker配置中设置delete.topic.enable=true)

Topic list会指明哪些topic被删除(在0.8.2以上版本适用)

为已存在的topic增加分区

为已存在的topic更新配置

在多个topic上批量重分区

在多个topic上批量重分区(可选partition broker位置)

KafkaOffsetMonitor

KafkaOffsetMonitor就是一个jar包而已，是一个针对于消费者的工具。它可以用于监控消费延迟的问题，不过对于重复消费和消息丢失等就无法解决，因为之后如果需要讲解SparkStreaming，flink这些用于消费者的实践的话，会使用到这个工具，所以现在先不展开，了解一下即可。

启动命令：

java -cp KafkaOffsetMonitor-assembly-0.3.0-SNAPSHOT.jar

com.quantifind.kafka.offsetapp.OffsetGetterWeb

--offsetStorage kafka

--zk xx:2181,xx:2181,xx:2181/kafka_cluster

--port 8088

--refresh 60.seconds

--retain 2.days

还有一些跨机房同步数据的像MirrorMaker这些，酌情使用。