Kafka技术进阶

1 Kafka技术精讲

1.1 Controller选举

Controller，是Apache Kafka的核心组件。它的主要作用是在Apache ZooKeeper的帮助下管理和协调控制整个Kafka集群。

集群中的任意一台Broker都能充当Controller的角色，但是，在整个集群运行过程中，只能有一个Broker成为Controller。也就是说，每个正常运行的Kafka集群，在任何时刻都有且只有一个Controller。

最先在ZooKeeper上创建临时节点/controller成功的Broker就是Controller。Controller重度依赖Zookeeper，依赖ZooKeeper保存元数据，依赖Zookeeper进行服务发现。Controller大量使用Watcher功能实现对集群的协调管理。如果此时，作为Controller的Broker节点宕掉了。那么ZooKeeper的临时节点/controller就会因为会话超时而自动删除。而监控这个节点的Broker就会收到通知而向ZooKeeper发出创建/controller节点的申请，一旦创建成功，那么创建成功的Broker节点就成为了新的Controller。

有一种特殊的情况，就是Controller节点并没有宕掉，而是因为网络抖动不稳定，导致和ZooKeeper之间的会话超时，那么此时，整个Kafka集群就会认为之前的Controller已经下线从而选举出新的Controller，而之前的Controller的网络又恢复了，以为自己还是Controller了，继续管理整个集群，那么此时，整个Kafka集群就有两个Controller进行管理，那么其他的Broker就懵了，不知道听谁的了，这种情况，我们称之为脑裂现象，为了解决这个问题，Kafka通过一个任期（epoch:纪元）的概念来解决，也就是说，每一个Broker当选Controller时，会告诉当前Broker是第几任Controller，一旦重新选举时，这个任期会自动增1，那么不同任期的Controller的epoch值是不同的，那么旧的Controller一旦发现集群中有新任Controller的时候，那么它就会完成退出操作（清空缓存，中断和Broker的连接，并重新加载最新的缓存），让自己重新变成一个普通的Broker。

1.2 Broker上下线

Controller在初始化时，会利用ZooKeeper的Watcher机制注册很多不同类型的监听器，当监听的事件被触发时，Controller就会触发相应的操作。Controller在初始化时，会注册多种类型的监听器，主要有以下几种：

• 监听/admin/reassign_partitions节点，用于分区副本迁移的监听
• 监听/isr_change_notification节点，用于Partition ISR变动的监听
• 监听/admin/preferred_replica_election节点，用于需要进行Partition最优Leader选举的监听
• 监听/brokers/topics节点，用于Topic新建的监听
• 监听/brokers/topics/TOPIC_NAME节点，用于Topic Partition扩容的监听
• 监听/admin/delete_topics节点，用于Topic删除的监听
• 监听/brokers/ids节点，用于Broker上下线的监听

每台Broker在上线时，都会与ZooKeeper建立一个建立一个session，并在/brokers/ids下注册一个节点，节点名字就是brokerid，这个节点是临时节点，该节点内部会有这个Broker的详细节点信息。Controller会监听/brokers/ids这个路径下的所有子节点，如果有新的节点出现，那么就代表有新的Broker上线，如果有节点消失，就代表有旧的Broker下线，Controller会进行相应的处理，Kafka就是利用ZooKeeper的这种Watcher机制及临时节点的特性来完成集群Broker的上下线。无论Controller监听到的哪一种节点的变化，都会进行相应的处理，同步整个集群元数据。

1.3 数据偏移量定位

分区是一个逻辑工作单元，其中记录被顺序附加分区上（Kafka只能保证分区消息的有序性，而不能保证消息的全局有序性）。但是分区不是存储单元，分区进一步划分为Segment（段），这些段是文件系统上的实际文件。为了获得更好的性能和可维护性，可以创建多个段，而不是从一个巨大的分区中读取，消费者现在可以更快地从较小的段文件中读取。创建具有分区名称的目录，并将该分区的所有段作为各种文件进行维护。在理想情况下，数据流量分摊到各个Parition中，实现了负载均衡的效果。

每个数据日志文件会对应一个LogSegment对象，并且都有一个基准偏移量，表示当前LogSegment中第一条消息的偏移量offset。

偏移量是一个64位的长整形数，固定是20位数字，长度未达到用0进行填补，索引文件和日志文件都由该作为文件名命名规则：

00000000000000000000.index：索引文件，记录偏移量映射到.log文件位置，此映射用于从任何特定偏移量读取记录

0000000000000000000.timeindex：时间戳索引文件，此文件包含时间戳到记录偏移量的映射

00000000000000000000.log：此文件包含实际记录，并将记录保持到特定偏移量，文件名描述了添加到此文件的起始偏移量，如果日志文件名为 00000000000000000004.log，则当前日志文件的第一条数据偏移量就是4（偏移量从0开始）

多个数据日志文件在操作时，只有最新的日志文件处于活动状态，拥有文件写入和读取权限，其他的日志文件只有只读的权限。

偏移量索引文件用于记录消息偏移量与物理地址之间的映射关系。时间戳索引文件则根据时间戳查找对应的偏移量。Kafka中的索引文件是以稀疏索引的方式构造消息的索引，并不保证每一个消息在索引文件中都有对应的索引项。每当写入一定量的消息时，偏移量索引文件和时间戳索引文件分别增加一个偏移量索引项和时间戳索引项。通过修改log.index.interval.bytes的值，改变索引项的密度。

数据位置索引保存在index文件中，log日志默认每写入4K（log.index.interval.bytes设定的），会写入一条索引信息到index文件中，因此索引文件是稀疏索引，它不会为每条日志都建立索引信息，索引文件的数据结构则是由相对offset（4byte）+position（4byte）组成，由于保存的是相对第一个消息的相对offset，只需要4byte就可以，节省空间，实际查找后还需要计算回实际的offset，这对用户是不可见的。

如果消费者想要消费某一个偏移量的数据，那么Kafka会通过Kafka中存在一个ConcurrentSkipListMap（跳跃表）定位到00000000000000000000.index索引文件，通过二分法在偏移量索引文件中找到不大于指定偏移量的最大索引项，然后从日志分段文件中的物理位置开始顺序查找偏移量为指定值的消息。

1.4 日志清理和压缩

Kafka软件的目的本质是用于传输数据，而不是存储数据，但是为了均衡生产数据速率和消费者的消费速率，所以可以将数据保存到日志文件中进行存储。默认的数据日志保存时间为7天，可以通过调整如下参数修改保存时间：

• log.retention.hours：小时（默认：7天，最低优先级）
• log.retention.minutes：分钟
• log.retention.ms：毫秒（最高优先级）
• log.retention.check.interval.ms：负责设置检查周期，默认5分钟

日志一旦超过了设置的时间，Kafka中提供了两种日志清理策略：delete和compact。

• delete：将过期数据删除

  • log.cleanup.policy = delete（所有数据启用删除策略）
  • 基于时间：默认打开。以Segment中所有记录中的最大时间戳作为该文件时间戳。
  • 基于大小：默认关闭。超过设置的所有日志总大小，删除最早的Segment。log.retention.bytes，默认等于-1，表示无穷大。

• compact：日志压缩

  • 基本思路就是将相同key的数据，只保留最后一个

  • log.cleanup.policy = compact（所有数据启用压缩策略）

    

  • 注意：因为数据会丢失，所以这种策略只适用保存数据最新状态的特殊场景。

1.5 页缓存

页缓存是操作系统实现的一种主要的磁盘缓存，以此用来减少对磁盘I/O的操作。具体来说，就是把磁盘中的数据缓存到内存中，把对磁盘的访问变为对内存的访问。为了弥补性能上的差异，现代操作系统越来越多地将内存作为磁盘缓存，甚至会将所有可用的内存用于磁盘缓存，这样当内存回收时也几乎没有性能损失，所有对于磁盘的读写也将经由统一的缓存。

当一个进程准备读取磁盘上的文件内容时，操作系统会先查看待读取的数据所在的页是否在页缓存（page cache）中，如果存在（命中）则直接返回数据，从而避免了对物理磁盘的I/O操作；如果没有命中，则操作系统会向磁盘发起读取请示并将读取的数据页写入页缓存，之后再将数据返回进程。同样，如果一个进程需要将数据写入磁盘，那么操作系统也会检测数据对应的页是否在页缓存中，如果不存在，则会先在页缓存中添加相应的页，最后将数据写入对应的页。被修改过后的页也就变成了脏页，操作系统会在合适的时间把脏页中的数据写入磁盘，以操作数据的一致性。

Kafka中大量使用了页缓存，这是Kafka实现高吞吐的重要因此之一。虽然消息都是先被写入页缓存，然后由操作系统负责具体的刷盘任务，但在Kafka中同样提供了同步刷盘及间断性强制刷盘（fsync）的功能，这些功能可以通过log.flush.interval.message、log.flush.interval.ms等参数来控制。同步刷盘可以提高消息的可靠性，防止由于机器掉电等异常造成处于页缓存而没有及时写入磁盘的消息丢失。不过一般不建议这么做，刷盘任务就应交由操作系统去调配，消息的可靠性应该由多副本机制来保障，而不是由同步刷盘这种严重影响性能的行为来保障。

1.6 顺写日志

Kafka中消息是以Topic进行分类的，生产者生产消息，消费者消费消息，都是面向Topic的。在Kafka中，一个Topic可以分为多个Partition，一个Partition分为多个Segment，每个Segment对应三个文件：.index文件、.log文件、.timeindex文件。

Kafka底层采用的是FileChannel.wrtieTo进行数据的写入，写的时候并不是直接写入文件，而是写入ByteBuffer，然后当缓冲区满了，再将数据顺序写入文件，无需定位文件中的某一个位置进行写入，那么就减少了磁盘查询，数据定位的过程。所以性能要比随机写入效率高得多。

官网有数据表明，同样的磁盘，顺序写能到600M/s，而随机写只有100K/s。这与磁盘的机械结构有关，顺序写之所以快，是因为其省去了大量磁头寻址的时间。

2.Kafka集群优化

2.1 集群部署方案

2.1.1 操作系统

如果考虑操作系统与Kafka的适配性，Linux系统显然要比其他特别是Windows系统更加适合部署Kafka。虽然这个结论可能你不感到意外，但其中具体的原因你也一定要了解。主要是在下面这三个方面上，Linux 的表现更胜一筹。

• I/O模型的使用：实际上Kafka客户端底层使用了Java的selector，selector在Linux上的实现机制是epoll，而在Windows平台上的实现机制是select。因此在这一点上将Kafka部署在Linux上是有优势的，因为能够获得更高效的I/O性能。
• 数据网络传输效率：你知道的，Kafka生产和消费的消息都是通过网络传输的，而消息保存在哪里呢？肯定是磁盘。故Kafka需要在磁盘和网络间进行大量数据传输。如果你熟悉Linux，你肯定听过零拷贝技术，就是当数据在磁盘和网络进行传输时避免昂贵的内核态数据拷贝从而实现快速的数据传输。Linux平台实现了这样的零拷贝机制，但有些令人遗憾的是在Windows平台上必须要等到Java 8的60更新版本才能“享受”到这个福利。一句话总结一下，在Linux部署Kafka能够享受到零拷贝技术所带来的快速数据传输特性。
• 社区支持度：这一点虽然不是什么明显的差别，但如果不了解的话可能比前两个因素对你的影响更大。简单来说就是，社区目前对Windows平台上发现的Kafka Bug不做任何承诺。虽然口头上依然保证尽力去解决(但实际上Windows上的Bug一般是不会修复的)。因此，Windows平台上部署Kafka只适合于个人测试或用于功能验证，千万不要应用于生产环境。

2.1.2 磁盘选型

如果问哪种资源对Kafka性能最重要，磁盘无疑是要排名靠前的。在对Kafka集群进行磁盘规划时经常面对的问题是，我应该选择普通的机械磁盘还是固态硬盘？前者成本低且容量大，但易损坏；后者性能优势大，不过单价高。我给出的建议是使用普通机械硬盘即可。

Kafka大量使用磁盘不假，可它使用的方式多是顺序读写操作，一定程度上规避了机械磁盘最大的劣势，即随机读写操作慢。从这一点上来说，使用SSD似乎并没有太大的性能优势，毕竟从性价比上来说，机械磁盘物美价廉，而它因易损坏而造成的可靠性差等缺陷，又由Kafka在软件层面提供机制来保证，故使用普通机械磁盘是很划算的。

关于磁盘选择另一个经常讨论的话题就是到底是否应该使用磁盘阵列（RAID）。使用RAID的两个主要优势在于：提供冗余的磁盘存储空间和提供负载均衡。以上两个优势对于任何一个分布式系统都很有吸引力。不过就Kafka而言，一方面Kafka自己实现了冗余机制来提供高可靠性；另一方面通过分区的概念，Kafka也能在软件层面自行实现负载均衡。如此说来RAID的优势就没有那么明显了。当然，我并不是说RAID不好，实际上依然有很多大厂确实是把Kafka底层的存储交由RAID的，只是目前Kafka在存储这方面提供了越来越便捷的高可靠性方案，因此在线上环境使用RAID似乎变得不是那么重要了。

综合以上的考量，我给出的建议是：

• 追求性价比的公司可以不搭建RAID，使用普通磁盘组成存储空间即可。
• 使用机械磁盘完全能够胜任Kafka线上环境。

2.1.3 磁盘容量

Kafka集群到底需要多大的存储空间？这是一个非常经典的规划问题。Kafka需要将消息保存在底层的磁盘上，这些消息默认会被保存一段时间然后自动被删除。虽然这段时间是可以配置的，但你应该如何结合自身业务场景和存储需求来规划Kafka集群的存储容量呢？

我举一个简单的例子来说明该如何思考这个问题。假设你所在公司有个业务每天需要向Kafka集群发送1亿条消息，每条消息保存两份以防止数据丢失，另外消息默认保存两周时间。现在假设消息的平均大小是1KB，那么你能说出你的Kafka集群需要为这个业务预留多少磁盘空间吗？

我们来计算一下：每天1亿条1KB大小的消息，保存两份且留存两周的时间，那么总的空间大小就等于1亿 * 1KB * 2 / 1000 / 1000 = 200GB。一般情况下Kafka集群除了消息数据还有其他类型的数据，比如索引数据等，故我们再为这些数据预留出10%的磁盘空间，因此总的存储容量就是220GB。既然要保存两周，那么整体容量即为220GB * 14，大约3TB左右。Kafka支持数据的压缩，假设压缩比是0.75，那么最后你需要规划的存储空间就是0.75 * 3 = 2.25TB。

总之在规划磁盘容量时你需要考虑下面这几个元素：

• 新增消息数
• 消息留存时间
• 平均消息大小
• 备份数
• 是否启用压缩

2.1.4 带宽

与其说是带宽资源的规划，其实真正要规划的是所需的Kafka服务器的数量。假设你公司的机房环境是千兆网络，即1Gbps，现在你有个业务，其业务目标或SLA是在1小时内处理1TB的业务数据。那么问题来了，你到底需要多少台Kafka服务器来完成这个业务呢？

让我们来计算一下，由于带宽是1Gbps，即每秒处理1Gb的数据，假设每台Kafka服务器都是安装在专属的机器上，也就是说每台Kafka机器上没有混部其他服务，毕竟真实环境中不建议这么做。通常情况下你只能假设Kafka会用到70%的带宽资源，因为总要为其他应用或进程留一些资源。

根据实际使用经验，超过70%的阈值就有网络丢包的可能性了，故70%的设定是一个比较合理的值，也就是说单台Kafka服务器最多也就能使用大约700Mb的带宽资源。

稍等，这只是它能使用的最大带宽资源，你不能让Kafka服务器常规性使用这么多资源，故通常要再额外预留出2/3的资源，即单台服务器使用带宽700Mb / 3 ≈ 240Mbps。需要提示的是，这里的2/3其实是相当保守的，你可以结合你自己机器的使用情况酌情减少此值。

好了，有了240Mbps，我们就可以计算1小时内处理1TB数据所需的服务器数量了。根据这个目标，我们每秒需要处理2336Mb的数据，除以240，约等于10台服务器。如果消息还需要额外复制两份，那么总的服务器台数还要乘以3，即30台。

2.2 集群参数配置

2.2.1 Broker端参数

• 针对存储信息:

  • log.dirs:指定了Broker需要使用的若干个文件目录路径，在线上生产环境中一定要为log.dirs配置多个路径，具体格式是一个CSV格式，也就是用逗号分隔的多个路径，比如/home/kafka1,/home/kafka2,/home/kafka3这样。

    如果有条件的话你最好保证这些目录挂载到不同的物理磁盘上。这样做有两个好处：

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

• 针对ZooKeeper相关配置:

  • zookeeper.connect:这也是一个CSV格式的参数，比如我可以指定它的值为zk1:2181,zk2:2181,zk3:2181。

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

• 针对Topic管理相关配置:

  • auto.create.topics.enable:建议最好设置成false，即不允许自动创建Topic。在我们的线上环境里面有很多名字稀奇古怪的Topic，我想大概都是因为该参数被设置成了true的缘故。

  • unclean.leader.election.enable:关闭Unclean Leader选举，何谓Unclean？还记得Kafka有多个副本这件事吗？每个分区都有多个副本来提供高可用。在这些副本中只能有一个副本对外提供服务，即所谓的Leader副本。那么问题来了，这些副本都有资格竞争Leader吗？显然不是，只有保存数据比较多的那些副本才有资格竞选，那些落后进度太多的副本没资格做这件事。

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

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

  • auto.leader.reblance.enable:是否开启分区Leader的自平衡，后台线程会定期检查分区Leader的分布情况(可以通过leader.imbalance.check.interval.seconds参数配置检查时间间隔)，如果分区Leader的不平衡程度超过了设定的参数leader.imbalance.per.broker.percentage则会触发Leader重平衡到分区的Prederred Leader。换一次Leader代价很高的，原本向A发送请求的所有客户端都要切换成向B发送请求，而且这种换Leader本质上没有任何性能收益，因此我建议你在生产环境中把这个参数设置成false。

• 针对数据留存相关配置:

  • log.retention.{hours|minutes|ms}：这是个”三兄弟”，都是控制一条消息数据被保存多长时间。从优先级上来说ms设置最高、minutes次之、hours最低。

    虽然ms设置有最高的优先级，但是通常情况下我们还是设置hours级别的多一些，比如log.retention.hours=168表示默认保存7天的数据，自动删除7天前的数据。很多公司把Kafka当作存储来使用，那么这个值就要相应地调大。

  • log.retention.bytes:指定Broker为消息保存的总磁盘容量大小，这个值默认是-1，表明你想在这台Broker上保存多少数据都可以，至少在容量方面Broker绝对为你开绿灯，不会做任何阻拦。

    这个参数真正发挥作用的场景其实是在云上构建多租户的Kafka集群：设想你要做一个云上的Kafka服务，每个租户只能使用100GB的磁盘空间，为了避免有个”恶意”租户使用过多的磁盘空间，设置这个参数就显得至关重要了。

  • message.max.bytes:控制Broker能够接收的最大消息大小，默认的1000012太少了，还不到1MB。实际场景中突破1MB的消息都是屡见不鲜的，因此在线上环境中设置一个比较大的值还是比较保险的做法。毕竟它只是一个标尺而已，仅仅衡量Broker能够处理的最大消息大小，即使设置大一点也不会耗费什么磁盘空间的。

2.2.2 Topic级别参数

说起Topic级别的参数，你可能会有这样的疑问：如果同时设置了Topic级别参数和全局Broker参数，到底听谁的呢？哪个说了算呢？答案就是Topic级别参数会覆盖全局Broker参数的值，而每个Topic都能设置自己的参数值，这就是所谓的Topic级别参数。

举个例子说明一下，上一期我提到了消息数据的留存时间参数，在实际生产环境中，如果为所有Topic的数据都保存相当长的时间，这样做既不高效也无必要。更适当的做法是允许不同部门的Topic根据自身业务需要，设置自己的留存时间。如果只能设置全局Broker参数，那么势必要提取所有业务留存时间的最大值作为全局参数值，此时设置Topic级别参数把它覆盖，就是一个不错的选择。

下面我们依然按照用途分组的方式引出重要的Topic级别参数。从保存消息方面来考量的话，下面这组参数是非常重要的：

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

上面这些是从保存消息的维度来说的。如果从能处理的消息大小这个角度来看的话，有一个参数是必须要设置的，即max.message.bytes。它决定了Kafka Broker能够正常接收该Topic的最大消息大小。我知道目前在很多公司都把Kafka作为一个基础架构组件来运行，上面跑了很多的业务数据。如果在全局层面上，我们不好给出一个合适的最大消息值，那么不同业务部门能够自行设定这个Topic级别参数就显得非常必要了。在实际场景中，这种用法也确实是非常常见的。

我们先来看看如何在创建Topic时设置这些参数。我用上面提到的retention.ms和max.message.bytes举例。设想你的部门需要将交易数据发送到Kafka进行处理，需要保存最近半年的交易数据，同时这些数据很大，通常都有几MB，但一般不会超过5MB。现在让我们用以下命令来创建Topic：

bin/kafka-topics.sh --bootstrap-server localhost:9092 --create --topic transaction --partitions 1 --replication-factor 1 --config retention.ms=15552000000 --config max.message.bytes=5242880

我们只需要知道Kafka开放了kafka-topics命令供我们来创建Topic即可。对于上面这样一条命令，请注意结尾处的--config设置，我们就是在config后面指定了想要设置的Topic级别参数。

下面看看使用另一个自带的命令kafka-configs来修改Topic级别参数。假设我们现在要发送最大值是10MB 的消息，该如何修改呢？命令如下：

bin/kafka-configs.sh --zookeeper localhost:2181 --entity-type topics --entity-name transaction --alter --add-config max.message.bytes=10485760

2.2.3 JVM参数

先无脑给出一个通用的建议：将你的JVM堆大小设置成6GB吧，这是目前业界比较公认的一个合理值(当然最好还是监控一下实时的堆大小，特别是GC之后的live data大小，通常将HeapSize设置成其1.5~2倍就足以了)。我见过很多人就是使用默认的Heap Size来跑Kafka，说实话默认的1GB有点小，毕竟Kafka Broker在与客户端进行交互时会在JVM堆上创建大量的ByteBuffer实例，Heap Size不能太小。

JVM端配置的另一个重要参数就是垃圾回收器的设置，也就是平时常说的GC设置。如果你依然在使用Java 7，那么可以根据以下法则选择合适的垃圾回收器：

• 如果Broker所在机器的CPU资源非常充裕，建议使用CMS收集器。启用方法是指定-XX:+UseCurrentMarkSweepGC。
• 否则，使用吞吐量收集器。开启方法是指定-XX:+UseParallelGC。

当然了，如果你在使用Java 8，那么可以手动设置使用G1收集器。在没有任何调优的情况下，G1表现得要比CMS出色，主要体现在更少的Full GC，需要调整的参数更少等，所以使用G1就好了。

现在我们确定好了要设置的JVM参数，我们该如何为Kafka进行设置呢？其实设置的方法也很简单，你只需要设置下面这两个环境变量即可：

• KAFKA_HEAP_OPTS：指定堆大小。
• KAFKA_JVM_PERFORMANCE_OPTS：指定GC参数。

比如你可以这样启动Kafka Broker，即在启动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

2.2.4 操作系统参数

• 文件描述符限制:首先是ulimit -n。我觉得任何一个Java项目最好都调整下这个值。实际上，文件描述符系统资源并不像我们想象的那样昂贵，你不用太担心调大此值会有什么不利的影响。通常情况下将它设置成一个超大的值是合理的做法。
• 文件系统类型:其次是文件系统类型的选择。这里所说的文件系统指的是如ext3、ext4或XFS这样的日志型文件系统。根据官网的测试报告，XFS的性能要强于ext4，所以生产环境最好还是使用XFS。
• Swappiness:网上很多文章都提到设置其为0，将swap完全禁掉以防止Kafka进程使用swap空间。我个人反倒觉得还是不要设置成0比较好，我们可以设置成一个较小的值。为什么呢？因为一旦设置成0，当物理内存耗尽时，操作系统会触发OOM killer这个组件，它会随机挑选一个进程然后kill掉，即根本不给用户任何的预警。但如果设置成一个比较小的值，当开始使用swap空间时，你至少能够观测到Broker性能开始出现急剧下降，从而给你进一步调优和诊断问题的时间。基于这个考虑，我个人建议将swappniess配置成一个接近0但不为0的值，比如1。
• 提交时间:向Kafka发送数据并不是真要等数据被写入磁盘才会认为成功，而是只要数据被写入到操作系统的页缓存（Page Cache）上就可以了，随后操作系统根据LRU算法会定期将页缓存上的“脏”数据落盘到物理磁盘上。这个定期就是由提交时间来确定的，默认是5秒。一般情况下我们会认为这个时间太频繁了，可以适当地增加提交间隔来降低物理磁盘的写操作。当然你可能会有这样的疑问：如果在页缓存中的数据在写入到磁盘前机器宕机了，那岂不是数据就丢失了。的确，这种情况数据确实就丢失了，但鉴于Kafka在软件层面已经提供了多副本的冗余机制，因此这里稍微拉大提交间隔去换取性能还是一个合理的做法。