02 Confluent_Kafka权威指南 第二章：安装kafka

CHAPTER 2 Installing Kafka kafka的安装配置

本章节描述了如何安装apache kafka的broker,以及如何设置apache zookeeper,zookeeper被用于存储broker的元数据。本章节还将介绍kafka的基本配置，以及broker运行的硬件标准。最后，我们将介绍多实例集群的安装方法，以及在生产环节中使用kafka时需要注意的一些问题。

First Things First

在安装和使用kafka之前，需要准备如下事项：

Choosing an Operating System

Apache kafka是一个基于java的应用程序，可以在许多操作系统上运行。包括windows、MacOS、Linux和其他的操作系统。本章的安装步骤主要对linux环境安装kafka进行讨论。因为这是安装kafka最常见的操作系统。这也是一般kafka所推荐的操作系统。有关于windows、MacOS上安装kafka的信息，参见附录A。

Installing Java

在安装kafka和zookeeper之前，你需要先安装和ava环境。jdk要求1.8及以上版本，可以是操作系统提供的版本，也可以从java.com下载。虽然zookeeper和kafka在jre环境下就能工作，但是在开发过程中，最好是安装完整的jdk。本文假定你已安装完整的jdk1.8.51版本。安装在/usr/java/ jdk1.8.0_51。

Installing Zookeeper

apache kafka使用zookeeper存储borker的元数据和客户端的详细信息。如下图所示。虽然可以使用kafka安装包中提供的zookeeper,但是通常发行版本提供的zookeeper都是很简单的。

kafka已经在zookeeper的稳定版本3.4.6之上进行了广泛的测试。可以从apache.org上下载http://bit.ly/2sDWSgJ.

Standalone Server

下面示例在/usr/local/zookeeper中安装基本的zookeeper。数据存储在/var/lib/zookeeper。

# tar -zxf zookeeper-3.4.6.tar.gz
# mv zookeeper-3.4.6 /usr/local/zookeeper
# mkdir -p /var/lib/zookeeper
# cat > /usr/local/zookeeper/conf/zoo.cfg << EOF
> tickTime=2000
> dataDir=/var/lib/zookeeper
> clientPort=2181
> EOF
# export JAVA_HOME=/usr/java/jdk1.8.0_51
# /usr/local/zookeeper/bin/zkServer.sh start
JMX enabled by default
Using config: /usr/local/zookeeper/bin/../conf/zoo.cfg
Starting zookeeper ... STARTED
#

现在，你可以通过客户端连接到服务端口发送四个字母的命令srvr来验证zookeeper在Standalone模式下是否正确运行：

# telnet localhost 2181
Trying ::1...
Connected to localhost.
Escape character is '^]'.
srvr
Zookeeper version: 3.4.6-1569965, built on 02/20/2014 09:09 GMT
Latency min/avg/max: 0/0/0
Received: 1
Sent: 0
Connections: 1
Outstanding: 0
Zxid: 0x0
Mode: standalone
Node count: 4
Connection closed by foreign host.
#

Zookeeper ensemble zookeeper集合

一个zookeeper集群被成称为一个集合。由于zookeeper的核心算法建议集群的成员个数为奇数如3、5等等。zookeeper如果要完成响应，大多数的集群成员必须工作。这意味着在3节点的集群中，你可以在缺少一个节点的情况下运行。在5节点的集群中，可以缺少2个节点仍然能运行。

Sizing Your Zookeeper Ensemble

建议在5个节点的集群中运行zookeeper。为了对集群进行设置你必须按个节点的配置。如果你的集群不能容忍多个节点宕机，那么维护工作将带来额外的风险。但是也不建议超过7个以上的节点，因为由于zookeeper底层协议的性质，节点过多可能会导致性能下降。 要在一个集群中配置zookeeper，他们必须要有一个列出所有服务器的公共配置，并且每个服务器都需要在数据牡蛎中存在一个myid文件。该文件用于指定服务器的ID号。假定服务器的主机名为 zoo1.example.com,zoo2.example.com,zoo3.example.com,配置文件则为如下：

tickTime=2000
dataDir=/var/lib/zookeeper
clientPort=2181
initLimit=20
syncLimit=5
server.1=zoo1.example.com:2888:3888
server.2=zoo2.example.com:2888:3888
server.3=zoo3.example.com:2888:3888

在这个配置中，tickTime表示zookeeper中的最小时间单位为20s,也就是说每个tickTimezookeeper各节点之间会向leader节点发送一个心跳。initLimit是允许follower节点与leader节点在初始连接的时候最多能容忍的心跳数。follower节点在启动的时候，如果超过initLimit的连接时间，即20个心跳时间，则连接失败。syncLimit则是集群follower节点与leader节点之间发送消息请求和应答的响应时间长度。如果在synclimit时间内不能完成与leader节点的通信，那么follower将被丢弃。 配置还列出了集群中的每个成员服务器。服务器是在配置文件中指定的。格式为：

server.X=hostname:peerPort:leaderPort

X: 服务器的ID号，他必须是整数，但是不一定需要从0开始或者顺序的。 hostname: 服务器的主机名或者IP地址。 peerPort： 集群中服务器进行通信的TCP端口。 leaderPort： 执行服务选举的TCP端口。 客户端只需要能供通过clientPort连接到集群，但是集群成员必须通过所有的三个端口互相通信。 除了共享的配置文件之外，每个服务器的数据文件中还必须有一个myid的文件。该文件内容为服务器的ID编号。该ID号必须与配置文件一致。一旦这些步骤完成，服务器将启动并以一个集群的形式彼此通信。

Installing a Kafka Broker

一旦配置好了java和zookeeper，你就可以安装apache kafka。最新版本的kafka可以在http://kafka.apache.org/down loads.html下载。假定已下载了0.9.0.1版本，其scala版本为2.11.0。 下面示例在/usr/local/kafka中安装kafka,在zookeeper启动之后，进行配置。消息存储在/tmp/kafka-logs:

# tar -zxf kafka_2.11-0.9.0.1.tgz
# mv kafka_2.11-0.9.0.1 /usr/local/kafka
# mkdir /tmp/kafka-logs
# export JAVA_HOME=/usr/java/jdk1.8.0_51
# /usr/local/kafka/bin/kafka-server-start.sh -daemon
/usr/local/kafka/config/server.properties
#

启动kafkabroker之后，我们可以通过对集群执行一些简单的操作来验证它是否正常工作。这些操作包括创建一个测试topic,生产一些消息并消费这些消息。 创建并验证topic

# /usr/local/kafka/bin/kafka-topics.sh --create --zookeeper localhost:2181
--replication-factor 1 --partitions 1 --topic test
Created topic "test".
# /usr/local/kafka/bin/kafka-topics.sh --zookeeper localhost:2181
--describe --topic test
Topic:test PartitionCount:1 ReplicationFactor:1 Configs:
Topic: test Partition: 0 Leader: 0 Replicas: 0 Isr: 0
#

生产测试消息：

# /usr/local/kafka/bin/kafka-console-producer.sh --broker-list
localhost:9092 --topic test
Test Message 1
Test Message 2
^D
#

消费测试消息：

# /usr/local/kafka/bin/kafka-console-consumer.sh --zookeeper
localhost:2181 --topic test --from-beginning
Test Message 1
Test Message 2
^C
Consumed 2 messages
#

Broker Configuration

虽然kafka的standalone模式足够对kafka分布式的概念进行验证，但是这对于大多数kafka使用者来说这是不够的。kafka有许多配置参数可以对kafka进行设置和调优。许多选项可以使用默认配置，在kafka的调优方面，你只有在一个特定的使用例或者将这些设置调整为特定值。

General Broker broker通用配置

在把kafka从standaloan模式切换到集群模式时，对kafka的配置文件要反复检查。这些通用的配置参数中的大多数必须重新配置，以便与其他broker在集群环境下正确运行。

broker.id

每个kafka的broker必须有一个整数的标识符，该标识符在broker.id进行设置。默认情况下这个标识符被设置为0，但是他可以是任何值。最重要的一点就是在整个kafka集群中，这个值必须唯一。这个数字的选择可以是任意的，如果维护需要，可以在多个broker之间移动。一个好的策略死将这个值设置为主机的固定值，以便在维护的时候不需要费力的对主机和broker之间映射。例如，你可以在主机名中包含一个唯一的编号（host1.example.com,host2.example.com）这对broker.id来说是一个不错的选择。

port

kafka的tcp端口默认是9092，可以通过改变配置参数将其设置为任何值。但是，如果选择的端口小于1024，那么kafka必须做为root用户启动，这是不推荐的。

zookeeper.connect

用于存储borker的元数据的zookeeper位置是用zookeeper.connect配置的。示例运行在本地2181端口的zookeeper，被设置为 localhost:2181。这个参数的格式是一个以分号分隔的主机名:端口/路径字符串列表。包括：

• hostname,zookeeper服务器的主机名或者IP。
• port，服务器的客户端连接端口号。
• /path,一个可选的zookeeper路径，用于kafka集群的chroot环境。如果省略，则在zookeeper上使用root路径。

如果指定了chroot路径但是zookeeper上没有找到，那么在broker启动的时候将创建该路径。

Why Use a Chroot Path

通常来讲，对kafka集群使用chroot路径是一种很好的选择。这使得zookeeper集群可以与其他程序共享。包括其他kafka集群。而不会产生冲突。最好在配置中指定多个zookeeper成员（都是这个集群的一部分）。这允许kafka在zookeeper某个节点故障的时候进行转移到其他节点。

log.dirs

kafka将所有的消息持久化到磁盘，并且这些消息分段存储在log.dirs指定的目录中。这是在本地系统上用逗号分隔的路径列表，如果指定了多个路径broker将以最少使用的方式在这些路径上存储分区，将一个分区的日志存储在同一个路径中。注意，在如下情况下，broker将当前存储分区数量最少的路径放置一个新分区，而不是使用磁盘余量空间最大的路径。

num.recovery.threads.per.data.dir

kafka使用一个可配置的线程池来处理日志的段，在如下情况下使用线程池：

• 正常启动的时候，打开每个分区的日志段
• 在故障后重新启动，检查并截断每个分区的日志段
• 当关闭kafka节点时，要关闭并清理日志段 默认情况下每个日志目录只使用一个线程。由于这些线程只在启动和关闭时使用，因此做了并行化处理，设置线程池是合理的。 具体来说，当从没有clean的关闭中恢复时，这可能意味着在重启具有大量分区的broker可能需要数小时。在设置此参数的时候，需要注意的是这个参数配置的是每个路线的线程数。 这意味着，果num.recovery.threads.per.data.dir 设置为8，在log.dirs存在3个路径的话，这将启动24个线程。

auto.create.topics.enable

默认的kafka的配置指定的broker应该在一下情况下自动创建topic:

• 当生产者开始向Topic写入消息的时候
• 当消费者开始从topic读消息的时候
• 当任何客户端请求该topic的元数据的时候 通常来讲，这是不希望出现的，特别是当无法通过kafka协议验证主题的存在时创建topic。如果你显式的管理topic的创建，无论是手动还是通过配置系统创建，都建议将auto.create.topics.enable设置为false。

Topic Defaults

kafka为创建的topic指定了许多默认的配置，其中一些参数，包括分区计数和消息保留。可以使用管理工具按topic设置（详见第9章）。服务器默认配置的值为基本配置，这些配置适合大多数topic的配置。

Using Per-Topic Overrides

在kafka早期版本中，可以使用参数log.retention.hours.per.topic，log.retention.bytes.per.topic,log.segment.bytes.per.topic。在broker的配置中为每个topic重写这个配置。这些配置在新版本中不再支持，必须使用管理工具指定参数进行覆盖重新。

num.partitions

参数num.partitions 确定创建topic的分区数量，主要是在启动自动创建topic的时候使用（这是默认配置）。这个参数默认值为1，需要注意的是topic的分区一旦设置，只能增加，不能减少。这意味着如果一个topic的分区数量少于num.partitions ，在手动创建时尤其需要注意。 如第一章所述，分区时kafka集群中对topic进行扩展的方式，这使得在添加broker的时候使用分区数量来平衡整个集群的消息负载非常重要。 许多用户将topic分区数设置为与集群中的broker数量相等。这将允许将负载均匀的分配给bioker。但是，这不是必须的，因为你还可以通过设置多个topic来进行消息的平衡负载。

How to Choose the Number of Partitions

考虑分区数量时，有如下几个因素需要考虑：

• 你希望该topic的吞吐量时多少？例如，希望某个topic的吞吐量为100kB/s或者1kB/s
• 你希望的当个分区的最大吞吐量是多少？你需要知道的是，在大多数情况下，一个消费者从topic消费数据，一个较慢的消费者将数据写入数据库，这个数据库的处理能力在每个线程上不超过50MB/s,那么你这个topic的吞吐量将不会超过60MB/s。
• 你可以通过相同的方式来估算每个生产者的最大吞吐量，但是在通常情况下，由于生产者通常比消费者快得多，因此通常不用考虑生产者的吞吐量。
• 如果你正在基于key向分区发送消息，那么以后扩展分区将非常有挑战性，因此需要根据预期的未来使用情况而不是目前的情况来计算吞吐量。
• 综合考虑每个borker的分区数量和每个borker可用的磁盘空间和网络带宽。
• 避免将分区数量估计过高，每个分区需要使用broker的内存以及其他资源，并将增加leader的选举时间。

综合考虑上述因素，很明显分区需要一定的数量，但并不是越多越好。如果你对topic的目标吞吐量和使用吞吐量有一些预期的话，那么可以将目标的吞吐量除以预期每个消费者的吞吐量，这样得到分区的数量。 因此如果我们需要能够独写1GB/s的topic，并且每个消费者只能处理50MB/s,那么我们能计算得到分区至少需要20个，这样我们可以让20个消费者订阅主题，总吞吐量达到1GB/s。 如果你没有这方面的详细消息，那么我们的经验告诉你，将每个分区磁盘上的文件大小限制在每天小于6GB通常会得到较好的效果。

log.retention.ms

对于kafka最常用的配置就是日志保存时间。可以通过指定log.retention.hours进行配置，通常被设置为168小时，或者一周。另外还有两个参数也可以对此进行控制，log.retention.minutes和 log.retention.ms。这三个参数都可以对消息保留时间进行配置，但是如果同时配置了三个参数，则只有log.retention.ms会生效，这是因为较小的时间单位会让较大时间单位的配置失效。系统会确保较小的时间单位配置优先。 ####### Retention By Time and Last Modified Times 保留时间是通过检查磁盘上每个日志段文件的最后修改时间来确定的。在正常的集群操作中，这是关闭日志段的时间，表示文件中最后一条消息的时间戳。但是，当我们使用管理工具对broker的分区进行移动操作之后，这个时间将不准确，会导致这些分区超额保留。在第九章将对分区的移动操作进行详细的讨论。

log.retention.bytes

另外一种使消息过期的方法是基于保留消息的总字节数。这个值通过log.retention.bytes参数配置，按分区生效。这意味着如果你的topic有8个分区，如果log.retention.bytes设置为1GB，那么topic的字节数量最多为8GB。需要注意的是所有的保留设置都是针对分区进行配置的，而不是topic，这意味着如果扩展了topic的分区，在使用了log.retention.bytes参数的情况下，那么保留率也会增加。 ####### Configuring Retention by Size and Time 如果 log.retention.bytes 与log.retention.ms（或者其他按时间保留的参数）都进行了配置，当满足任何一个条件时，消息将被删除。 例如log.retention.ms 设置为86400000(一天)，log.reten tion.bytes设置为 1000000000(1GB)，如果1天之内的数据量大于1GB那么不到1天消息就可能被删除。想反如果消息小于1G，但是在1天之后这些消息也将被删除。

log.segment.bytes

前面提到消息保留设置操作日志的段而不是单个消息。当消息被生产者写入到kafkabroker之后，它们将被附加到分区的日志段中，一旦达到了log.segment.bytes设置的大小（这大小默认1GB）,broker将会关闭日志段打开一个新的段。一旦关闭了日志段，过期操作就可以执行。如果日志段设置得较小，那么意味着频繁的关闭并分配段文件，这将降低磁盘的写操作的效率。（也就是说日志保留的操作都是在完整的段文件上进行的，如果段文件未关闭，则不会参与日志保留计算） 如果topic的生成率很低，那么调整日志段的大小就很重要，如果一个topic每条接受100MB的消息， log.segment.bytes设置为默认值，一个段要被写慢大概需要10天。由于日志段未关闭之前消息不会过期，如果log.retention.ms 设置为604800000（一周），在关闭的日志段到期之前，实际上部分消息最后会被保留17天。这是因为一旦日志段用来当前10天的消息未关闭，根据时间策略，该日志段必须保留7天。（因为在该日志段最后一天消息过期之前，不能删除该段）。 ####### Retrieving Offsets by Timestamp 日志段的大小还会根据时间戳影响对offset的获取。当在特定的时间戳请求分区的offset时，kafka会找到当时正在写入的日志段文件，它使用文件的创建和最后修改的时间，并查找在指定时间戳之前 创建和最后修改的文件。在响应中返回该日志段开头的offset（这也是文件名）。

log.segment.ms

另外一种控制日志段关闭的方式是使用log.segment.ms参数。它指定日志段被关闭的时间。与log.retention.bytes 和 log.retention.ms类似，log.segment.bytes 和 log.segment.ms同时生效。当达到大小限制或者达到时间限制的时候，kafka将关闭一个日志段，无论哪个先出现。默认情况下，没有对log.segment.ms进行设置，那么只会根据日志段的大小来关闭日志段。 ####### Disk Performance When Using Time-Based Segments 在使用基于时间的日志段限制时，必须要考虑同时关闭多个日志段对磁盘性能的影响。当有许多分区从未达到日志段的大小限制，就会发生这种情况，因为时间限制的时钟源在broker中，会同时对这些未满的分区执行关闭操作。

message.max.bytes

kafka的broker限制了生产者可以写入的最大消息的大小，通过message.max.bytes进行配置，默认大小是1M，如果生产者试图发送大于此数值的消息，则会收到返回错误，消息将不被接受。这个字节大小指的是压缩消息之后的大小，这意味着如果生产者使用压缩，就可以发送比这个配置大很多的数据。只需要保证将消息压缩后再此参数配置值以下。 增加对允许消息大小限制会对生产者的性能产生明显的影响，更大的消息意味着处理网络连接和请求broker的线程在每个请求上的工作时间会更长。更大的消息还会增加磁盘写操作的大小，这将影响I/O的吞吐量。 ####### Coordinating Message Size Configurations kafkabroker上配置的消息大小必须与fetch.message.max.bytes协调一致，如果这个值小于message.max.bytes，那么遇到更大的消息消费者将无法消费这些消息。导致消费者陷入困境，无法继续下去。同样的规则也适用于replica.fetch.max.bytes在集群配置的broker上的配置。也就是说replica.fetch.max.bytes与fetch.message.max.bytes都尽量与message.max.bytes一致，否则将导致kafka无法正常工作。

Hardware Selection 硬件选择

为kafka的broker选择合适的硬件配置可能更像一门艺术而不是一门科学。kafka本身对特定的硬件配置没有严格的要求，并且可以在任何系统上毫无问题的允许。然而，在考虑性能之后，有几个因素会对总体性能产生影响：磁盘吞吐量和容量，内存、网络和CPU。 一旦确定了你的环境中最关键的性能指标，那么你将在适合的预算内优化硬件配置。

Disk Throughput 磁盘吞吐量

用于存储日志段的broker的磁盘的吞吐量将最直接的影响生产者客户端的写入性能。kafka消息在生产的时候必须提交到本地村春中，大多数客户端将等待至少一个broker确认消息已提交完成之后才算发送成功。者意味着更快的磁盘写操作将等价于更低的生产者写入延迟。 对于磁盘吞吐量而言，最明显的区别是使用传统的机械硬盘HDD还是固态硬盘SSD。SSD具有非常低的查找和访问时间，能提供最佳的性能。另外一方面，HDD更加经济，能提供更多的容量。你还可以在broker使用更多的HDD来提高性能，无论通过多个数据目录，还是通过都列的磁盘冗余整列raid进行配置。另外一个因素，比如具体驱动技术（如串行附加存储器或者串行ATA）,以及驱动控制器的质量，也将影响吞吐量。

Disk Capacity 磁盘容量

容量是存储讨论的另外一个方面，所需的磁盘容量取决于在任何时候需要保留多少消息。如果预计broker每天将接收1TB的消息，并且将保留7天。那么broker将至少需要7TB的可用存储来存储日志文件。出路为流量波动活随着时间增长带来的维护缓冲之外，还应该考虑至少10%的其他系统文件开销。 在确定kafka的集群规模和决定何时进行扩展时，存储的容量是要考虑的因素之一。集群的总通信量可以通过为每个topic设置多个分区来实现平衡，这将允许额外的broker在单个broker密度不足以满足性能要求的情况下增加可用容量。 为集群选择合适的复制策略也将决定了需要选择多少磁盘容量（这将在第6章详细讨论）。

Memory 内存

kafka消费者的正常操作模式是从分区的末端读取数据，在这个过程中，消费者几乎不会落后于生产者。在这种情况下，消费者读取的消息被最佳的存储在系统的页缓存中，这样的读取速度比broker必须从磁盘重新读取消息的速度要快很多。因此，为系统提供更多的可用内存用于页缓存将提高消费者客户端的性能。 kfka本身不需要为jvm虚拟机分配太多的堆内存。即使每秒处理X条消息和每秒处理X MB数据的broker，也可以在5G堆内存中运行。 系统内存的其余部分将由页缓存使用，并且允许系统缓存正在使用中的日志段，这将使kafka从中受益。 这是不建议将kafka与其他重要的应用程序混合部署的主要原因，因为他们必须共享页缓存，这将降低kafka的性能。

Networking 网络

可用网络吞吐量将限制kafka的最大可处理的流量。与磁盘存储一起，这通常是影响集群规模的因素之一。由于kafka对于多个消费者的支持，使得入站和出站网络使用之间的不平衡使得这个问题更加复杂化。生产者每秒写入1MB，但是有任意数量的消费者导致了数倍的出站数据。其他操作如集群的副本和镜像等也会增加网络带宽需求。如果网络接口饱和导致集群副本落后的情况并不少见，这会使集群异常脆弱。

CPU

处理器的计算能力不像磁盘内存那样重要，但是他会在一定程度上影响broker的总体性能。理想情况下，客户端应该压缩消息以优化网络磁盘的使用。但是kafka的broker必须解压缩全部批次的消息，才能对单个消息进行校验和分配offset。然后，它需要重新批次压缩消息，以便将其存储在磁盘上。这就是kafka对CPU处理能力的大部分需求来源。然而这不是选择硬件的主要因素。

Kafka in the Cloud

kafka一个常见的安装就是在云计算的环境中，如亚马逊的AWS。AWS提供了血多计算实例，每个实例都具有不同的CPU、内存和磁盘组合，因此必须优先考虑kafka的各种性能特征，以便选择合适的实例配置。首先要考虑的是所需的数据保留量，然后考虑的是生产者所需的写入性能。如果需要低延迟，则可能需要具有SSD的IO优化实例。否则，通用的存储就能够满足需求了。如AWS提供的EBS。一旦做出了这些选择，可用的CPU和内存选项将提供更加合适的性能。 实际上，这意味着对于AWS而言，M4或者R3实例类型就是常见的选择。M4实例将允许更大的保留期，但是磁盘吞吐量比较小，因为它使用弹性块存储。R3实例在本地使用SSD，因此会有更好的吞吐量。但是保留的数据可能会有限制。如果两者都需要考虑的话，有必要升级到I2或者D2类型，他们的成本会高很多。

Kafka Clusters

单个kafka服务器可以很好的用于本地开发工作，或者用于概念验证，但是将多个broker配置为要给集群有显著的好处，如下图所示。最大的好处是能够跨多个服务器扩展系统负载。第二个好处是使用副本防止由于单个系统故障造成的数据丢失。复制还允许在kafka或底层系统上执行维护工作，同时任然保持对客户端的可用性。本节主要讨论如何配置kafka集群，第6章包含了更多的数据复制的信息。

How Many Brokers?

kafka集群的大小由如下几个因素决定。需要考虑的抵押给因素就是保留消息的磁盘容量，以及单个broker上的可用存储空间。如果集群要保留10TB数据，而一个broker可以存储2TB，那么最小的集群大小是5个broker。此外由于副本会增加存储的100%。取决于选择的副本因子。这意味着相同的集群，如果配置了副本，那么集群现在至少需要10个broker。需要考虑的另外一个因素就是集群处理请求的能力。例如，网络接口的容量是多少，如果数据有多个使用者，或者在数据保留期间流量的不平稳，他们能处理消费者的流量吗？如果单个broker上的网络接口在峰值使用了80%的容量，并且有两个数据的消费者。那么消费者将无法跟上峰值的流量，除非有两个broker。如果在集群中使用副本，则必须考虑数据的额外使用者。未来处理较少的磁盘吞吐量或者可用的系统内存引起的性能问题，者可能希望在集群扩展中得到更多的broker。

Broker Configuration

为了允许多个kafka的broker加入单个集群，broker配置中只有两个需求，首先，对于zookeeper，所有的broker必须都具有相同的配置，连接参数。这指定了zookeeper集群存储kafka集群元数据的存储路径。 第二个要求是集群中所有的broker必须为该broker提供一个唯一的ID值。如果两个broker试图用同一个brokerID加入同一个集群，第二个broker将启动失败并报错。在运行集群时还需要使用其他的配置参数，特别是控制复制的副本因子。这些参数将在后面章节介绍。

OS Tuning 操作系统调优

虽然大多数linux发行版本都为内核调优参数提供了开箱即用的配置，这些配置在大多数的应用程序中都能很好的工作，但是可以对kafka的broker上的参数进行修改，从而进一步提高性能。这些问题主要涉及虚拟内存和网络子系统，以及存储日志段的磁盘挂载点的具体问题。 这些参数通常在/etc/susctl.conf文件中配置，但是你应该参考linux发现版的文档以连接关于如何调整内核配置的具体细节。

Virtual Memory

通常linux虚拟内存会根据系统的工作负载自动调整。我们可以对swap交换分区的处理方式以及内存page进行一些调整，以适应kafka的工作负载。 与大多数应用程序一样(特别是考虑吞吐量的应用程序)最好避免以所有代价进行交换。将内存的分页交换到磁盘所产值的成本将对kafka的各方面使用性能产生显著的影响。此外，kafka大量使用系统的page缓存，如果虚拟内存要交换到磁盘，就没有足够的内存分配给page缓存。 避免交换的一种办法就是根本上不使用交换分区，交换不是必要的条件，如果系统发生故障，它能提供一个安全保障。使用交换分区可以防止操作系统的内存不足而突然终止进程。因此建议将系统的虚拟交换分区设置为一个非常小的值，例如1。该参数是虚拟内存使用的交换分区空间，而不是从page缓存中删除page可能性的百分百。最好减少页缓存的大小，而不是使用交换分区。

Why Not Set Swappiness to Zero?

之前，我们建议将参数vm.swappiness设置为0，这个值的意思是除非内存不足，否则不要使用交换分区。然而，这个值的含义在linux内核版本为3.5-rc1中发生了变化，这个变化被移植到了许多发行版本中，包括red hat 企业版本（内核为2.6.32-303）。将0的值的含义改变为在任何情况下都不要使用交换分区。正因为如此，推荐将值改为1。 调整内核如何处理必须刷新到磁盘的脏页也有好处，kafka依靠磁盘IO的性能为生产者提供良好的响应时间。这也是日志段通常放在快速的磁盘上的原因，无论是响应时间快的单个磁盘（如SSD）,还是具有重要NVRAM用于缓存的磁盘系统(如RAID)。结果是在刷新后台进程开始将脏页写入磁盘之前，允许的脏页的数量可以减少。 设置=vm.dirty_background_ratio的值低于默认值10。该值是系统内存总量的百分比，在许多情况下将该值设置为5是合适的。但是不应该将该值设置为0，这将导致内核不断刷新page。这将消除内核缓冲磁盘的写操作，以应对底层设备性能的临界值峰值问题。 内核强制同步操作将脏页刷新到磁盘之前允许的脏页总数也可以通过更改vm.dirty_ratio的值来增加。将其增加到高于默认值20(也是系统总内存的百分比)。这个设置的可能值比较宽泛，介于60-80之间是一个合理的数字，这个设置确实引入了一些风险，包括未刷新的磁盘活动的数量。以及如果强制同步刷新，可能会出现长时间的I/O暂停。如果vm.dirty_ratio设置较高。强烈建议在kafka集群中设置跟多的副本以防止系统故障。 在为这些参数选择值时，明智的做法是查看kafka集群在负载下的运行脏页的数量（无论是生产环境还是模拟环境）。当前脏页的数量可以通过检查/proc/vmstat来确定：

# cat /proc/vmstat | egrep "dirty|writeback"
nr_dirty 3875
nr_writeback 29
nr_writeback_temp 0
#

Disk

除了选择磁盘设备的硬件以及RAID的配置（如果要选择使用了RAID）之外。重要性仅次于此的是对使用文件系统的选择对性能的影响。有许多不同的文件系统可以选择，但是最常用的选择就是EXT4或者XFS。目前，XFS已经成为许多人默认的文件系统，这是一个很好的原因，对于大多数工作的负载，他的性能优于EXT4。只需要很少的调优。EXT的性能可能很好，但是它需要使用被认为不太安全的调优参数。这包括将提交间隔设置为比默认值5更长的时间以及强制更少的刷新频率。EXT4还引入了延迟块的分配，在系统故障时带来了更大的数据丢失和文件系统损坏的可能性。XFS文件系统虽然也使用了延迟分配算法，但是它通常比EXT4使用的算法更安全。对于kafka的工作负载，XFS也有更好的性能，无需执行文件系统的自动调优之外的调优。当批处理写操作的hi收，它也更加高效。无论为保存日志段的挂载磁盘选择哪个文件系统，建议为挂载点设置noatime挂载选项。文件的元数据包括三个时间戳，创建时间(ctime)，最后修改时间(mtime)，和最后访问时间(atime)。默认情况下，atime在每次读取文件的时候更新。这会生成大量的磁盘写操作。atime通常被认为没有什么用，除非应用程序需要指定文件在最后修改之后是否被访问(这种情况下可以使用realtime选项)。atime不被kafka使用，所以禁用它是安全的，在挂载上设置了noatime，将组织这些时间戳更新的发生。但不会影响对ctime和mtime属性的正确处理。

Networking

对于任何有大量网络通信操作的应用程序而言，调整linux网络堆栈的默认调优都很常见。因为内核在默认情况下不会为大型告诉的数据传输进行调优。实际上，建议更改kafka与大多数web服务器和其他网络应用程序的配置相同。第一个调整时更改每个socket发送和接收缓冲区分配的默认最大内存。这将显著提高大型传输的性能。每个socket发送和接收的缓冲区默认大小相关的参数时net.core.wmem_default和net.core.rmem_default。这些参数合理的配置为131072或者128KB。发送和接收缓冲区的大小的参数时net.core.wmem_max和net.core.rmem_ma。合理设置时2097152或者2MB。请记住，最大大小并不代表每个socket将分配这么多缓冲空间。它只在需要的时候会分配这么多。 除了socket接收设置之外，tcp socket的发送和接收缓冲区大小必须使用net.ipv4.tcp_wmem 和net.ipv4.tcp_rmem 设置。他们使用三个空格分隔的整数来设置。分别指定最小，默认和最大大小。最大大小不能小于使用net.core.wmem_max和 net.core.rmem_max的设置的值。一个参数实例为：“4096 65536 2048000” 这是要给最小4KB,默认64KB，最大2M的缓冲区。根据kafka broker的实际工作负载，你可能需要增加最大大小，已运行网络连接有更大的缓冲区。 还有其他几个网络调优参数需要设置，启动TCP窗口缩放通过设置net.ipv4.tcp_window_scaling 为1 将运行客户端更有效的传输数据，并允许在broker端缓冲数据。 增加net.ipv4.tcp_max_syn_backlog的值，高于默认的1024，将允许接收更多的并发连接。将net.core.netdev_max_backlog的值增加到默认值1000以上可以帮助处理突发的网络流量。特别实在使用千兆网络的连接速度时，通过允许更多的包排队等待内核来处理他们。

Production Concerns 生产环境问题

一旦你决定将kafka环境从测试转移到生产，需要注意如下情况。他们将有助于建立可靠的消息传递服务。

Garbage Collector Options GC选择

调优应用程序的java GC选项一直是一门艺术。需要掌握应用程序如何使用内存的详细信息，以及大量的观察和尝试。幸运的是，java自1.7之后引入了G1垃圾收集器改变了这一点。G1可以根据不同的工作负载自动调整，并在应用程序生命周期内为垃圾收集提供一致的暂停时间。他还可以轻松处理堆内存分配很大的GC，方法是将堆分割为更小的区域，并且每次暂停都不收集整个堆。 G1在正常运行中只需要很少的配置就可以正常工作。G1有两种配置选项来调整其性能：

MaxGCPauseMillis

此选项指定每个GC操作周期的首选暂停时间。这不是一个固定的上限，如果需要将超过这个时间。这个值默认为200ms，G1将尝试安排GC周期的频率，以及每个周期收集区域的数量，这样每个周去将花费大约200ms。

InitiatingHeapOccupancyPercent

此选项指定以前可能正在使用的堆占总堆的百分比，G1将开始一个GC周期。默认值是45,这意味着在45%的堆被使用之前，G1不会启动收集周期，这包括老年代和年前代的使用情况。 kafka对堆的利用率相当高效，因此可以适当降低堆内存。本节提供的GC调优选项适用于内存为64GB，堆内存分配为5GB的kafka服务器。 对于MaxGCPauseMillis, 这个broker可以配置为20ms，InitiatingHeapOccupancyPercent这个值可以设置为35.这将导致GC比默认值更早的运行。 kafka默认的启动脚本使用的不是G1收集器，而是默认的parallel new收集器和cms收集器。使用环境变量很容易修改。在start文件中修改如下：

# export JAVA_HOME=/usr/java/jdk1.8.0_51
# export KAFKA_JVM_PERFORMANCE_OPTS="-server -XX:+UseG1GC
-XX:MaxGCPauseMillis=20 -XX:InitiatingHeapOccupancyPercent=35
-XX:+DisableExplicitGC -Djava.awt.headless=true"
# /usr/local/kafka/bin/kafka-server-start.sh -daemon
/usr/local/kafka/config/server.properties
#

Datacenter Layout 数据中心布局

对于开发环境，kafka的broker再数据中心的物理位置不是很重要，因为如果集群在短时间内部分或者完全不可用，不会造成严重的影响。当然，如果服务应用于生产环境的时候，停机意味着资金的损失。无论是通过对用户的服务损失，还是通过对用户正在做的事情的结果造成损失。这时在kafka集群中配置的副本机制就变得至关重要（详见第六章）。同时考虑broker在数据中心的物理位置也很重要。如果在部署kafka之前没有考虑这些问题，那么以后可能需要花费昂贵的维护费用来移动服务器。 为了broker重新分配新的分区时，kafka的broker没有跨机架安全意识。这意味着它不能考虑两个broker可能位于相同的机架或者相同的可用区域。因此可以轻松的将一个分区的所有副本分配给在同一个机架中共享相同电源的broker。如果该机架发生故障，这个分区将失效，客户端无法访问。此外，由于不安全的leader选举也将导致数据丢失（详见第六章）。 最佳的实践是将集群中的每个kafka的broker安装在不同的机架上，或者至少不共享电力和网络的基础设施服务的单故障点。这通常意味着需要部署将运行具有双重的电源连接（连接到两个不同的电路）和双重的网络交换机（在服务器本身具有一个绑定接口，可以无缝的进行故障转移）的broker服务器。 即使是双重连接，将broker放在完全独立的机架上也是有好处的，有时，可能需要对机架或者机柜进行物理维护，使其处于离线状态（如移动服务器或者重新布线）。

Colocating Applications on Zookeeper 在zookeeper上应用的配置

kafka利用zookeeper存储关于broker、topic和分区的元数据信息。低于zookeeper的写操作仅在对消费者组成员的更改或对kafka集群本身的更改时进行。这样对kafka的数据流量是最小的。但是它不能说明一个kafka集群部署一个zookeeper就是合理的。事实上，许多认将多个kafka集群使用相同的zookeeper集群。（为每个集群使用chroot 的zookeeper的路径）。

Kafka Consumers and Zookeeper kafka的消费者和zookeeper

在apache kafka 0.9.0.0之前的版本，除了broker之外，消费者还将消费者组的组成、正在使用的topic以及定期提交的分区offerset信息等存储在zookeeper。（消费者组之间可以实现故障转移）。在版本0.9.0.0中，引入了新的消费者接口，该接口允许使用kafka的broker直接管理消费者的用户。将消费者的相关数据存储在broker中。这将在第4章讨论。 然而，在某些配置下，消费者和zookeeper存在一个问题。消费者可以选择的使用zookeeper或者kafka来提交offset。他们还可以配置提交offset之间的时间间隔。 如果使用zookeeper来获取offset，那么每个消费者都将为它所使用的分区在每个时间间隔上执行一个zookeeper写操作。offset的提交合理间隔时间设置为1分钟。因为在消费者出现故障的情况下，在这段时间内可能导致消费者重复消费。这些提交可能瞬间增大zookeeper的流量，特别是在有许多消费者的集群中尤其需要考虑。如果zookeeper集群无法响应这些流量，则这个时间间隔可能需要配置得更长。但是，建议使用最新的kafka版本，将offset提交到kafka。降低对zookeeper的依赖。 另外，除了为多个kafka集群使用的单一zookeeper集群之外，如果可能的话，不建议其他的应用共享使用zookeeper集群。kafka对zookeeper的延迟和超时很敏感，与zookeeper集群的通信中断将导致broker的行为不可预测。这很容易导致多个broker同时脱机，如果他们失去了zookeeper连接，这将导致分区的不可用。它还会给集群控制器带来压力，在中断过去很长的一段时间之后，比如在尝试执行有控制的broker关闭时，这些压力会造成错误异常。通过大量使用或不当操作，其他可能给zookeeper集群带来压力的应用程序应该隔离到他们自己的集群中。

Summary 概要

在本章中，我们学习了如何启动和运行apache kafka。我们还讨论了为broker选择合适的硬件，以及在生产环节中的配置问题。现在你有了一个kafka集群，我们将具体介绍kafka的客户端应用程序的基础知识。接下来两个章节，我们将介绍如何创建生产者来发送消息（第三章）,以及将这些消息消费出去（第四章）。