一个生产的配置

# ZooKeeper
zookeeper.connect=[list of ZooKeeper servers]

# Log configuration
num.partitions=8
default.replication.factor=3
log.dir=[List of directories. Kafka should have its own dedicated disk(s) or SSD(s).]

# Other configurations
broker.id=[An integer. Start with 0 and increment by 1 for each new broker.]
listeners=[list of listeners]
auto.create.topics.enable=false
min.insync.replicas=2
queued.max.requests=[number of concurrent requests]

> bin/kafka-mirror-maker.sh --consumer.config consumer.properties --producer.config producer.properties --whitelist my-topic

• 添加broker到集群很简单，就是分配一个独立的broker id并启动即可(connect同一个zk的都属于同一个cluster)。
• 新增的broker不会自动分配任何数据的分区，除非分区本移动到新broker上，否则如果没有新的topic创建，他就什么也不做, 因此需要迁移已有数据到新broker上。

• 建议监控服务器端的GC time，CPU使用，IO服务时间等。
• 建议监控客户端的全局的和单个topic的消息byte rate，

一开始听人说 kafka 是一个很牛逼的消息系统，所以有下面的一些问题，带着问题去看可能好一点。

• kafka 和其他消息队列有什么不同

目前市面上有好些消息系统，比如 RabbitMQ，Redis，ZeroMQ,ActiveMQ,Kafka/Jafka，他们各有什么特点， kafka 和他们有什么不同的地方?

• 为什么他的吞吐这么牛逼

kafka 的消息吞吐量可以达到 10W/s，为什么会有这样的能力?

• 整套系统的框架

kafka 有 Broker，Producer 和 Consumer，以及一个对整体进行管理的 Zookeeper，他们各自扮演什么角色， 又是怎么联系起来的？

• 如何使用

最后就是怎么使用这个东西？怎么监控它的运行情况？

先要明白几个和消息相关的概念，明白这些概念也就对整体有个了解了：

• 话题 Topic

Topic把Kafka中流通的消息进行分类, 它算是消息的载体。生产者把消息发送到对应的Topic上，消费者从这个 Topic获取消息进行处理。和其它大部分系统的topic一样，一个topic可以被0,1或者多个订阅者订阅使用。但 不同的是：在比如mqtt，rabbitmq等broker中，一个topic上同时只能有一条消息，后面的消息会把前面的消息 覆盖，但是kafka中的消息则会被保留起来，类似数据库，但是毕竟是一个broker，数据不可能一致留着占用资 源，因此可以通过设置消息产生时间期限和消息的数量大小限制来删除旧的消息。

• 分区 Partition

由于 kafka 中 topic 的特殊性，一个topic可能会接收到来自生产者的大量的消息，所以一个topic会承受很 大的压力, 因此在创建一个topic的时候会指定把这个topic分成几个区(Partition)，每个分区都有在这个 topic粒度上 的唯一编号，Producer 在发送消息的时候会由Producer对消息做 某种决策(或者说路由分流) ， 将这条消息归到某个分区。topic 的分区如下图所示。

Producer 确定某条消息该发往哪个Partition 的方法主要有两种 ，一个是随机确定，另一个是通过消息的一 个key来利用hash算法确定，通常的做法是 (hash(key) rem partition_num) 的结果作为选择的分区(所以 Producer首先要知道共有多少分区, zookeeper负责该功能)。这样通过分区就可以实现负载均衡, 增加kafka 的读写能力。

• 分段 Segment

每个broker上 给每个Partition都会产生一个目录 ，该目录下面保存这发往这个分区的消息。消息是按照顺 序存入磁盘的，但并不是作为一个文件保存在这个目录下面的，而是 分段保存成一个个文件(segment 文件) ， 每个segment以该segment的第一条消息的offset命名, 比如第一个文件是00000000000000000000.log，下一个 文件接着该文件最后一条消息的偏移往下排。kafka 中可以设置一些参数来控制 什么时候新建segment ， 什么时候删除旧的segment ， 后台会有线程隔多久检查一次 ，决定相关的处理。Partition 内的segment 如下图所示。

• 顺序写磁盘

Kafka是通过顺序写磁盘来持久化数据的，一般来说磁盘的操作效率是很低的，肯定很难得到这么高的吞吐量，但是 有人测试过顺序写磁盘有时候比随机写内存还要快。这种 顺序写磁盘的方式正式kafka高吞吐的一个重要原因 。

kafka 给每条消息都会置一个相对这个Partition第一条消息的偏移(offset) 。给每一条消息都编号并按顺序append 到上一条消息后面，这也方便Consumer更方便地消费数据。

• 消费方式 push vs pull

大部分消息系统的消费有两种方式：push vs pull。前者是服务器一有消息就直接推送给Consumer，不管 Consumer 目前有没有能力处理；pull则是Consumer有能力处理的时候自动向broker拉取消息，如果broker没有消息就会等待 消息的到来。 kafka的设计是生产者向broker push消息，而消费者向broker pull消息 。这样消息的使用决定权在 Consumer，所以不太会因为消息压力太大把Consumer搞挂。

同时，由于消息存储的时候有一个offset值， Consumer消费的时候也可以维持消费到的offset值，按顺序消费 ， 这样消费速度就会更快。当然如果消费者想消费已经消费过的消息，只要Consumer改变这个值就可重复消费了。

• 消费组 Consumer Group

kafka通过消费组把Consumer归在某些组里面，实现broker上消息对Consumer的单播和广播。 原则是一条消息只 要被一个组里面的一个 Consumer 消费，就不会再给这个组的其他成员了 。这样如果我们要对所有Consumer 广播，那就 他们单独成一个组，如果单播，就把多个 Consumer 放在一个 Group 里面。

• 复本集 Replicas

为容错性考虑，避免因为一个 broker 挂了就无法正常工作的情况，通常在使用时会将多个broker布置为一个cluster。 首先，这种考虑可以真正的实现负载均衡，让多个broker来分担压力；其次，一个Partition写到一个broker上， 一旦这个broker挂了，数据就可能丢失了。因此kafka 在Partition粒度上设置复制集 。我们在建立topic的时候也 会设置一个Replicas个数的参数，那么这个 topic 的所有分区都会有这么多个副本(注意并不是所有broker上都有所有 Partition的副本)，而且这些副本分散在不同的broker上，即使一个broker挂了或者不工作了，还有另外的broker保存 相同的数据，可以保证数据不丢失。

一个 Partition 的所有副本集中 有一个是leader，其他是follower。 Producer向cluster推送消息和Consumer从 broker拉取消息都只和leader交互 ， 然后follower像Consumer消费数据一样从leader上拷贝数据 。

• In-Sync Replicas (ISR)

和大部分分布式系统一样，要自动处理故障就必须对一个节点node的"alive"有一个精确的定义。kafka的存活node(也就 是In Sync node)需要满足两个条件：

leader会跟踪Partition的ISR集合(就是一个记录In Sync的列表)，如果某个follower挂了或者同步数据落后了，leader就 会把这个node从ISR列表中删除, 确定node挂了或者数据落后由参数replica.lag.time.max.ms确定。

如果一条消息被所有ISR节点写入各自的log，那么这个消息就是commited的消息，只有这样的消息才会给Consumer使用 , 也就是说Consumer看到的消息一定不会因为leader失效而丢失； 另一方面，用户可以设置是否让Producer等待它发送的 消息被标记为commited (request.required.acks设置)，这就需要在数据延迟和数据持久性上做出权衡了(做法是通过Producer使用ack来判断)。

Producer把一条消息发送到该Partition的leader之后，只有所有ISR都把这条消息写入log，然后给出ack确认之后这条消 息才叫commited的，此外leader的最后一条消息的offset叫做log end offset（LEO）。 当一个ISR的offset和leader的 LEO差值大于某个值（通过replica.lag.max.messages设置）时，这个broker就会被从ISR列表中剔除。 此时follower还 可以从leader上复制数据(其实就是消费数据)，当数据差值符合条件是还会被添加到ISR列表中.

需要注意的是ISR的数量并不是和该Partition的Replicas个数一样的，ISR数量是可以配置的， 这样就只需要有这个数量 的副本集对消息进行确认，消息才能认为是commited的。

• Kafka 对消息投递的保障

kafka 提供消息投递过程中的一些保证：

• 如何管理整个系统--zookeeper

它是一个快速、高可用、容错、分布式的协调服务。你可以使用 ZooKeeper 构建可靠的、分布式的数据结构，用于群组 成员、leader选举、协同工作流和配置服务，以及广义的分布式数据结构如锁、队列、屏障（Barrier）和锁存器（Latch）。 许多知名且成功的项目依赖于 ZooKeeper，其中包括 HBase、Hadoop 2.0、Solr Cloud、Neo4J、Apache Blur（Incubating） 和 Accumulo。

Kafka将元数据信息保存在Zookeeper中，但是 发送给Topic本身的数据是不会发到Zk上的 ，否则Zk就疯了。 kafka 使用zookeeper来实现动态的集群扩展 ，不需要更改客户端（producer和consumer）的配置。 broker会在zookeeper 注册并保持相关的元数据（topic，partition信息等）更新 。 而 客户端会在zookeeper上注册相关的watcher 。一 旦zookeeper发生变化，客户端能及时感知并作出相应调整。 这样就保证了添加或去除broker时，各broker间仍能自动 实现负载均衡。这里的客户端指的是Kafka的消息生产端(Producer)和消息消费端(Consumer), Producer端使用zookeeper 用来"发现"broker列表 ,以及和Topic下每个partition的leader建立socket连接并发送消息。也就是说每个Topic的partition 是由leader角色的Broker端使用zookeeper来注册broker信息,以及监测partition leader存活性。 Consumer端使用 zookeeper用来注册consumer信息 ,其中包括consumer消费的partition列表等,同时也用来发现broker列表,并和partition leader 建立socket连接,并获取消息。

kafka 需要 java-jdk 提供支持，同时还需要 zookeeper 来对集群进行管理。因此需要安装这些东西。

目前对 kafka 的监控程序有好几个，

前几个都没怎么更新了，比较新的是 kafka-monitor 是 LinkIn 支持的，kafka-manager 是 yahoo 支持的. 但是前者功能目前还很不完善。 这里使用雅虎的 kafka-manager。

如果想要获得更详细的监控信息，需要在启动kafka的时候打开JMX端口，在kafka-manager创建集群的时候也打开JMX Polling。 在 kafka-server-start.sh中修改为：

if [ "x$KAFKA_HEAP_OPTS" = "x" ]; then
    export KAFKA_HEAP_OPTS="-Xmx1G -Xms1G"
    export JMX_PORT="9999"
fi

就是加上export JMX_PORT="9999"

• 需要事先安装 java 环境。

$ git clone https://github.com/yahoo/kafka-manager.github
$ ./sbt clean dist #这个过程需要花费很长时间下载很多东西。

安装完之后会产生安装二进制包，在 target/universal 文件夹里面，是一个 zip 文件包。 解压这个 zip 文件，修改 conf/application.conf 文件。

kafka-manager.zkhosts="localhost:2181"
kafka-manager.zkhosts=${?ZK_HOSTS}

指定 zookeeper 的 ip 和 host，如果不想硬编码，可以设置环境变量 ZK_HOSTS="localhost:2181"

• 启动

$ ./bin/kafka-manager
# 如果要使用特定的配置文件或者打开特定的端口可以使用下面的命令(默认端口是 9000)：
$ ./bin/kafka-manager -Dconfig.file=/path/to/application.conf -Dhttp.port=8080
# 如果 java 的路径不在 path 中，可以如下制定
$ bin/kafka-manager -java-home /usr/local/oracle-java-8

完了就可以使用浏览器查看了。

• 新建 cluster

进入添加 cluster 界面，给集群一个命令，然后把 cluster 的 zookeeper 的 host 填好， 如果使用了多个 zookeeper，可以使用逗号隔开：

localhost:2181

• 可以新建 topic
• 参数含义

preferred replica: 每个 partitiion 的所有 replicas 叫做"assigned replicas"，"assigned replicas"中的 第一个replicas叫"preferred replica"，刚创建的topic一般"preferred replica"是leader。

zk中的目录结构和算法用于协调消费者和kafka节点。

[zk: 192.168.0.45:2181(CONNECTED) 40] ls / [cluster, controller_epoch, controller, brokers, zookeeper, kafka-manager, admin, isr_change_notification, consumers, config]

• Apache Kafka Best Practices https://www.slideshare.net/HadoopSummit/apache-kafka-best-practices
• Kafka Best Practices https://community.hortonworks.com/articles/80813/kafka-best-practices-1.html
• How many topics can be created in Apache Kafka? https://www.quora.com/How-many-topics-can-be-created-in-Apache-Kafka
• 某互联网大厂kafka最佳实践 http://www.jianshu.com/p/8689901720fd