Kafka 基本原理、生产问题总结及性能优化实践 | 京东云技术团队
Kafka 是最初由 Linkedin 公司开发，是一个分布式、支持分区的（partition）、多副本的（replica），基于 zookeeper 协调的分布式消息系统，它的最大的特性就是可以实时的处理大量数据以满足各种需求场景：比如基于 hadoop 的批处理系统、低延迟的实时系统、Storm/Spark 流式处理引擎，web/nginx 日志、访问日志，消息服务等等，用 scala 语言编写，Linkedin 于 2010 年贡献给了 Apache 基金会并成为顶级开源项目。（分布式的流处理平台）
Kafka 的使用场景
日志收集：一个公司可以用 Kafka 收集各种服务的 log，通过 kafka 以统一接口服务的方式开放给各种 consumer，例如 hadoop、Hbase、Solr 等。
消息系统：解耦和生产者和消费者、缓存消息等。
用户活动跟踪：Kafka 经常被用来记录 web 用户或者 app 用户的各种活动，如浏览网页、搜索、点击等活动，这些活动信息被各个服务器发布到 kafka 的 topic 中，然后订阅者通过订阅这些 topic 来做实时的监控分析，或者装载到 hadoop、数据仓库中做离线分析和挖掘。
运营指标：Kafka 也经常用来记录运营监控数据。包括收集各种分布式应用的数据，生产各种操作的集中反馈，比如报警和报告。
Kafka 基本概念
kafka 是一个分布式的，分区的消息(官方称之为 commit log)服务。它提供一个消息系统应该具备的功能，但是确有着独特的设计。可以这样来说，Kafka 借鉴了 JMS 规范的思想，但是确并没有完全遵循 JMS 规范。
首先，让我们来看一下基础的消息(Message)相关术语：
因此，从一个较高的层面上来看，producer 通过网络发送消息到 Kafka 集群，然后 consumer 来进行消费，如下图：集群架构，broker 就相当于服务端，处理消息的节点；producer 将消息 push，consumer 将消息 pull
依赖第三方的 zk，类似 nameserver；topic 类似消息队列，最终存放的是在 topic 下面的 partition 里面。
服务端(brokers)和客户端(producer、consumer)之间通信通过 TCP 协议来完成。
kafka 基本使用(2.4)
https://kafka.apache.org/documentation/
安装前的环境准备
由于 Kafka 是用 Scala 语言开发的，运行在 JVM 上，因此在安装 Kafka 之前需要先安装 JDK。
kafka 依赖 zookeeper，所以需要先安装 zookeeper
# 启动 zookeeper
查看 zk 的目录树：
第一步：下载安装包
下载 2.4.1 release 版本，并解压：
第二步：修改配置
修改配置文件 config/server.properties:
第三步：启动服务
现在来启动 kafka 服务：
启动脚本语法：kafka-server-start.sh [-daemon] server.properties
可以看到，server.properties 的配置路径是一个强制的参数，-daemon 表示以后台进程运行，否则 ssh 客户端退出后，就会停止服务。(注意，在启动 kafka 时会使用 linux 主机名关联的 ip 地址，所以需要把主机名和 linux 的 ip 映射配置到本地 host 里，用 vim /etc/hosts)
kafka 启动的时候，会把自己的信息注册到 zk 里面去
server.properties 核心配置详解：
第四步：创建主题
现在我们来创建一个名字为“test”的 Topic，这个 topic 只有一个 partition，并且备份因子也设置为 1：
现在我们可以通过以下命令来查看 kafka 中目前存在的 topic，主题的元数据在 zk 中，具体的消息在 kafka 中保存
除了我们通过手工的方式创建 Topic，当 producer 发布一个消息到某个指定的 Topic，这个 Topic 如果不存在，就自动创建。
删除主题
第五步：发送消息
kafka 自带了一个 producer 命令客户端，可以从本地文件中读取内容，或者我们也可以以命令行中直接输入内容，并将这些内容以消息的形式发送到 kafka 集群中。在默认情况下，每一个行会被当做成一个独立的消息。
首先我们要运行发布消息的脚本，然后在命令中输入要发送的消息的内容：
第六步：消费消息
对于 consumer，kafka 同样也携带了一个命令行客户端，会将获取到内容在命令中进行输出，默认是消费最新的消息：
如果想要消费之前的消息可以通过--from-beginning 参数指定，如下命令：
如果你是通过不同的终端窗口来运行以上的命令，你将会看到在 producer 终端输入的内容，很快就会在 consumer 的终端窗口上显示出来。
以上所有的命令都有一些附加的选项；当我们不携带任何参数运行命令的时候，将会显示出这个命令的详细用法。
消费多主题
单播消费
一条消息只能被某一个消费者消费的模式，类似 queue 模式，只需让所有消费者在同一个消费组里即可
分别在两个客户端执行如下消费命令，然后往主题里发送消息，结果只有一个客户端能收到消息
同一个消费组中，同一条消息，只能有一个消费者消费消息
多播消费
一条消息能被多个消费者消费的模式，类似 publish-subscribe 模式，针对 Kafka 同一条消息只能被同一个消费组下的某一个消费者消费的特性，要实现多播只要保证这些消费者属于不同的消费组即可。我们再增加一个消费者，该消费者属于 testGroup-2 消费组，结果两个客户端都能收到消息
查看消费组名
查看消费组的消费偏移量
current-offset：当前消费组的已消费偏移量
log-end-offset：主题对应分区消息的结束偏移量(HW)
lag：当前消费组未消费的消息数
以组为单位保存消费的偏移量，组来决定消费情况
kafka 消息消费完不会删除，消息是存储在文件里面，消息会根据 offset 来进行逐个消费，还可以指定偏移量进行消费。每一个消费者，会维护自己的偏移量。在某一个地方存上一次消费的偏移量。
消费情况和组绑定，不同的消费者只影响当前的组
主题 Topic 和消息日志 Log
可以理解 Topic 是一个类别的名称，同类消息发送到同一个 Topic 下面。对于每一个 Topic，下面可以有多个分区(Partition)日志文件:
Partition 是一个有序的 message 序列，这些 message 按顺序添加到一个叫做 commit log 的文件中。每个 partition 中的消息都有一个唯一的编号，称之为 offset，用来唯一标示某个分区中的 message。
为什么要多个分区？
真正大数据的系统，会有很多很多消息的，不会主动删除（最多可以保存 7 天），这里面会放很多的消息，如果一个分区，放在一起，存在一个磁盘同一台机器，肯定存不下，所以这个思想就是分布式存储，把一个主题划分很多分区，一个分区在一个文件里面，放在不同的机器上面去，达到分布式存储的效率。
不同的分区可以由不同的消费者消费，同时也提高了消费的能力。
每个 partition，都对应一个 commit log 文件。一个 partition 中的 message 的 offset 都是唯一的，但是不同的 partition 中的 message 的 offset 可能是相同的。
kafka 一般不会删除消息，不管这些消息有没有被消费。只会根据配置的日志保留时间(log.retention.hours)确认消息多久被删除，默认保留最近一周的日志消息。
针对业务场景，设置这个值。
kafka 的性能与保留的消息数据量大小没有关系，因此保存大量的数据消息日志信息不会有什么影响。
每个 consumer 是基于自己在 commit log 中的消费进度(offset)来进行工作的。在 kafka 中，消费 offset 由 consumer 自己来维护；一般情况下我们按照顺序逐条消费 commit log 中的消息，当然我可以通过指定 offset 来重复消费某些消息，或者跳过某些消息。
这意味 kafka 中的 consumer 对集群的影响是非常小的，添加一个或者减少一个 consumer，对于集群或者其他 consumer 来说，都是没有影响的，因为每个 consumer 维护各自的消费 offset。
创建多个分区的主题：
查看下 topic 的情况
以下是输出内容的解释，第一行是所有分区的概要信息，之后的每一行表示每一个 partition 的信息。
leader 节点负责给定 partition 的所有读写请求。
replicas 表示某个 partition 在哪几个 broker 上存在备份。不管这个几点是不是”leader“，甚至这个节点挂了，也会列出。
isr 是 replicas 的一个子集，它只列出当前还存活着的，并且已同步备份了该 partition 的节点。
我们可以运行相同的命令查看之前创建的名称为”test“的 topic
之前设置了 topic 的 partition 数量为 1，备份因子为 1，因此显示就如上所示了。
消息日志文件的存储
可以进入 kafka 的数据文件存储目录查看我们创建的主题的消息日志文件：cd /tmpkafka-logs/
消息日志文件主要存放在分区文件夹里的以 log 结尾的日志文件里，如下是 test_topic-0 主题对应的分区 0 的消息日志：
分区的扩容
消费者消费肯定会发生变化，reblance 机制
当然我们也可以通过如下命令增加 topic 的分区数量(目前 kafka 不支持减少分区)：
那么分区扩容之后，消费者会不会发生变化呢？肯定会触发它 rebalance 机制。
可以这么来理解 Topic，Partition 和 Broker
一个 topic，代表逻辑上的一个业务数据集，比如按数据库里不同表的数据操作消息区分放入不同 topic，订单相关操作消息放入订单 topic，用户相关操作消息放入用户 topic，对于大型网站来说，后端数据都是海量的，订单消息很可能是非常巨量的，比如有几百个 G 甚至达到 TB 级别，如果把这么多数据都放在一台机器上可定会有容量限制问题，那么就可以在 topic 内部划分多个 partition 来分片存储数据，不同的 partition 可以位于不同的机器上，每台机器上都运行一个 Kafka 的进程 Broker。
为什么要对 Topic 下数据进行分区存储？
1、commit log 文件会受到所在机器的文件系统大小的限制，分区之后可以将不同的分区放在不同的机器上，相当于对数据做了分布式存储，理论上一个 topic 可以处理任意数量的数据。
2、为了提高并行度。
kafka 集群实战
对于 kafka 来说，一个单独的 broker 意味着 kafka 集群中只有一个节点。
所以 kafka 的集群，主要是根据分区来的，所以它单台机器也叫集群。
要想增加 kafka 集群中的节点数量，只需要多启动几个 broker 实例即可。为了有更好的理解，现在我们在一台机器上同时启动三个 broker 实例。注册到同一个 zk 上面。
首先，我们需要建立好其他 2 个 broker 的配置文件：
配置文件的需要修改的内容分别如下：
config/server-1.properties:
config/server-2.properties:
目前我们已经有一个 zookeeper 实例和一个 broker 实例在运行了，现在我们只需要在启动 2 个 broker 实例即可：
查看 zookeeper 确认集群节点是否都注册成功：
现在我们创建一个新的 topic，副本数设置为 3，分区数设置为 2：
查看下 topic 的情况
kafka 一启动，都会注册到 zk 的对应的节点下面去
以下是输出内容的解释，第一行是所有分区的概要信息，之后的每一行表示每一个 partition 的信息。
leader 节点负责给定 partition 的所有读写请求，同一个主题不同分区 leader 副本一般不一样(为了容灾)
leader 处理收发数据，其他的节点是不可以写的，会同步给其他的从节点。
replicas 表示某个 partition 在哪几个 broker 上存在备份。不管这个几点是不是”leader“，甚至这个节点挂了，也会列出。
isr 是 replicas 的一个子集（已同步的副本集），它只列出当前还存活着的，并且已同步备份了该 partition 的节点。
容灾
把每一个分区放到不同的 broker 上去，如果有一个节点挂了，那么其他的还可以继续写数据。
写消息都是往 leader 上面写
副本被分配到三台机器上
现在我们向新建的 my-replicated-topic 中发送一些 message，kafka 集群可以加上所有 kafka 节点：
现在开始消费：
现在我们来测试我们容错性，因为 broker1 目前是 my-replicated-topic 的分区 0 的 leader，所以我们要将其 kill
现在再执行命令：
我们可以看到，分区 0 的 leader 节点已经变成了 broker 0。要注意的是，在 Isr 中，已经没有了 1 号节点。leader 的选举也是从 ISR(in-sync replica)中进行的。
此时，我们依然可以 消费新消息：
查看主题分区对应的 leader 信息：
kafka 将很多集群关键信息记录在 zookeeper 里，保证自己的无状态，从而在水平扩容时非常方便。
集群消费
log 的 partitions 分布在 kafka 集群中不同的 broker 上，每个 broker 可以请求备份其他 broker 上 partition 上的数据。kafka 集群支持配置一个 partition 备份的数量。
针对每个 partition，都有一个 broker 起到“leader”的作用，0 个或多个其他的 broker 作为“follwers”的作用。leader 处理所有的针对这个 partition 的读写请求，而 followers 被动复制 leader 的结果，不提供读写(主要是为了保证多副本数据与消费的一致性)。如果这个 leader 失效了，其中的一个 follower 将会自动的变成新的 leader。
Producers
生产者将消息发送到 topic 中去，同时负责选择将 message 发送到 topic 的哪一个 partition 中。通过 round-robin 做简单的负载均衡。也可以根据消息中的某一个关键字来进行区分。通常第二种方式使用的更多。
Consumers
传统的消息传递模式有 2 种：队列( queue) 和（publish-subscribe）
queue 模式：多个 consumer 从服务器中读取数据，消息只会到达一个 consumer。
publish-subscribe 模式：消息会被广播给所有的 consumer。
Kafka 基于这 2 种模式提供了一种 consumer 的抽象概念：consumer group。
queue 模式：所有的 consumer 都位于同一个 consumer group 下。
publish-subscribe 模式：所有的 consumer 都有着自己唯一的 consumer group。
上图说明：由 2 个 broker 组成的 kafka 集群，某个主题总共有 4 个 partition(P0-P3)，分别位于不同的 broker 上。这个集群由 2 个 Consumer Group 消费， A 有 2 个 consumer instances ，B 有 4 个。
通常一个 topic 会有几个 consumer group，每个 consumer group 都是一个逻辑上的订阅者（ logical subscriber ）。每个 consumer group 由多个 consumer instance 组成，从而达到可扩展和容灾的功能。
消费顺序
一个 partition 同一个时刻在一个 consumer group 中只能有一个 consumer instance 在消费，从而保证消费顺序。
consumer group 中的 consumer instance 的数量不能比一个 Topic 中的 partition 的数量多，否则，多出来的 consumer 消费不到消息。
Kafka 只在 partition 的范围内保证消息消费的局部顺序性，不能在同一个 topic 中的多个 partition 中保证总的消费顺序性。
如果有在总体上保证消费顺序的需求，那么我们可以通过将 topic 的 partition 数量设置为 1，将 consumer group 中的 consumer instance 数量也设置为 1，但是这样会影响性能，所以 kafka 的顺序消费很少用。
Java 客户端访问 Kafka
引入 maven 依赖
消息发送端代码
作者：京东零售 王雷
来源：京东云开发者社区 转载请注明来源
版权声明: 本文为 InfoQ 作者【京东科技开发者】的原创文章。
原文链接:【http://xie.infoq.cn/article/dd49d456b7d04eefba81ac7b4】。文章转载请联系作者。
拥抱技术，与开发者携手创造未来！ 2018-11-20 加入
我们将持续为人工智能、大数据、云计算、物联网等相关领域的开发者，提供技术干货、行业技术内容、技术落地实践等文章内容。京东云开发者社区官方网站【https://developer.jdcloud.com/】，欢迎大家来玩
评论