kafka集群Broker端参数设置及调优准则建议-kafka 商业环境实战

本套系列博客从真实商业环境抽取案例进行总结和分享，并给出Spark商业应用实战指导，请持续关注本套博客。版权声明：本套Spark商业应用实战归做者（秦凯新）全部，禁止转载，欢迎学习。

• 秦凯新的技术社区-大数据商业实战系列全集目录
• kafka 商业环境实战-kafka生产环境规划
• kafka 商业环境实战-kafka生产者和消费者吞吐量测试
• kafka 商业环境实战-kafka生产者Producer参数设置及参数调优建议
• kafka 商业环境实战-kafka集群Broker端参数设置及调优准则建议

1 Distributed streaming platform

Apache Kafka® is a distributed streaming platform. What exactly does that mean?
A streaming platform has three key capabilities:
   -  Publish and subscribe to streams of records, similar to a message queue or enterprise messaging system.
   -  Store streams of records in a fault-tolerant durable way.
   -  Process streams of records as they occur.

Kafka is generally used for two broad classes of applications:
- Building real-time streaming data pipelines that reliably get data between systems or applications
- Building real-time streaming applications that transform or react to the streams of data

To understand how Kafka does these things, let's dive in and explore Kafka's capabilities from the bottom up.
First a few concepts:

- Kafka is run as a cluster on one or more servers that can span multiple datacenters.
- The Kafka cluster stores streams of records in categories called topics.
- Each record consists of a key, a value, and a timestamp.
复制代码

2 Kafka as a Storage System

Any message queue that allows publishing messages decoupled from consuming them 
is effectively acting as a storage system for the in-flight messages. What is 
different about Kafka is that it is a very good storage system.

- Data written to Kafka is written to disk and replicated for fault-tolerance. 
Kafka allows producers to wait on acknowledgement so that a write isn't considered
complete until it is fully replicated and guaranteed to persist even if the server 
written to fails.

- The disk structures Kafka uses scale well，Kafka will perform the same whether you 
have 50 KB or 50 TB of persistent data on the server.

- As a result of taking storage seriously and allowing the clients to control 
their read position, you can think of Kafka as a kind of special purpose 
distributed filesystem dedicated to high-performance, low-latency commit 
log storage, replication, and propagation.
复制代码

3 kafka实现高吞吐率的秘密

• 一个用户程序要把文件内容发送到网络这个用户程序是工做在用户空间，文件和网络socket属于硬件资源，二者之间有一个内核空间。所以在操做系统内部，整个过程为：

  

在Linux kernel2.2 以后出现了一种叫作"零拷贝(zero-copy)"系统调用机制，就是跳过“用户缓冲区”的拷贝，创建一个磁盘空间和内存的直接映射，数据再也不复制到“用户态缓冲区”

• kafka的队列topic被分为了多个区partition，每一个partition又分为多个段segment，因此一个队列中的消息其实是保存在N多个片断文件中，经过分段的方式，每次文件操做都是对一个小文件的操做，增长了并行处理能力

• kafka容许进行批量发送消息，先将消息缓存在内存中，而后经过一次请求批量把消息发送出去，好比：能够指定缓存的消息达到某个量的时候就发出去，或者缓存了固定的时间后就发送出去，如100条消息就发送，或者每5秒发送一次这种策略将大大减小服务端的I/O次数。
• kafka还支持对消息集合进行压缩，Producer能够经过GZIP或Snappy格式或LZ4对消息集合进行压缩,压缩的好处就是减小传输的数据量，减轻对网络传输的压力。

4 kafka集群Broker端全局参数设置

---

• broker. id

惟一的整数来标识每一个broker，不能与其余broker冲突，建议从0开始。

---

• log.dirs <= 吞吐量提高

确保该目录有比较大的硬盘空间。若是须要指定多个目录，以逗号分隔便可，好比/xin/kafka1,/xin/kafka2。这样作的好处是Kafka会力求均匀地在多个目录下存放分区(partition)数据。若是挂载多块磁盘，那么会有多个磁头同时执行写操做。对吞吐量具备很是强的提高。

---

• zookeeper.connect

该参数则彻底没有默认值，必需要配置。这个参数也能够是一个逗号分隔值的列表，好比zk1:2181,zk2:2181,zk3:2181/kafka。注意结尾的/kafka，它是zookeeper的chroot，是可选的配置，若是不指定的话就默认使用zookeeper的根路径。

---

• listeners

协议配置包括PLAINTEXT，SSL, SASL_SSL等，格式是[协议]://[主机名]:[端口],[[协议]://[主机名]:[端口]]，该参数是Brocker端开发给clients的监听端口。建议配置：

PLAINTEXT://hostname:port（未启用安全认证）
    SSL://hostname:port（启用安全认证）
复制代码

---

• unclean.leader.election.enable <= 数据的完整性保证

解决ISR全部副本为空，leader又出现宕机的状况。此时leader该如何选择呢？截止kafka 1.0.0版本，该参数默认为false，表示不容许选择非ISR副本集以外的broker。由于高可用性与数据的完整性，kafka官方选择了后者。

---

• delete.topic.enable

很少说，是否容许删除topic，鉴于0.9.0.0新增了ACL机制权限机制，误操做基本是不存在的。

---

• log.retention.{hours|minutes|ms} <=时间维度

优先选取ms的配置，minutes次之，hours最后，默认留存机制是7天。如何判断：

新版本：基于消息中的时间戳来进行判断。 老版本：根据日志文件的最新修改时间进行比较.

---

• log.retention.bytes <=空间维度

Kafka会按期删除那些大小超过该参数值的日志文件。默认值是-1，表示Kafka永远不会根据大小来删除日志

---

• min.insync.replicas <= 与acks=-1 搭配使用

持久化级别，用于最少须要多少副本同步。在acks=all(或-1) 时才有意义。min.insync.replicas指定了必需要应答写请求的最小数量的副本数。若是不能知足，producer将会抛出NotEnoughReplicas或NotEnoughReplicasAfterAppend异常。该参数用于实现更好的消息持久性。

举例以下：

5台broker ack =-1 min.insync.replicas = 3

上述表示最少须要3个副本同步后，Broker才可以对外提供服务,不然将会抛出异常。若3台Broker宕机，即便剩余2台所有同步结束，知足了 ack =-1也要报错。

---

• num.network.threads <= 请求转发线程数量

默认值为3，主要负责转发来自broker和clients发送过来的各类请求。强调这里只是转发，真实环境下，须要监听 NetWorkerProcessorAvgIdlePercent JMX指标，若指标低于0.3，则建议调高该值。

---

• num.io.threads <= 实际处理线程数量

默认是8，也即broker后端有8个线程以轮询的方式不停的监听转发过来的网络请求并进行实时处理。

---

• message.max.bytes

broker可以接收的最大消息大小，默认是977KB。所以注意，生产环境应该调高该值。

4 kafka集群Broker端Topic级别参数设置

• delete.topic.enable
• message.max.bytes
• log.retention.bytes

5 操做系统参数设置

• OS页缓存刷盘flush时间 <= 提高吞吐量

默认是5秒，间隔实在过短，适当增长该值能够很高的提行OS物理写入操做的性能。LinkedIn设置为2分钟来提高吞吐量。

• 文件系统选择

官方测试XFS文件系统写入时间为160秒，Ext4大约是250毫秒。建议生产环境最好使用XFS文件系统。

• OS文件描述符限制

OS系统最大打开的文件描述符是有上限的，举例：一个kafka集群主要有3个副本，50个分区，若每个分区文件大小为10G，而分区内日志段大小为1GB，则一个Broker须要维护1500个左右的文件描述符。所以根据须要设置：

ulimit -n 100000
复制代码

• OS 操做系统缓冲区设置 （尚不肯定）

总结

本机立足于Broker进行参数详细讲解，有问题，欢迎留言。

秦凯新 于深圳 2018