Apache Kafka 入门

1.kafka简介和产生的背景

什么是 Kafka

Kafka 是一款分布式消息发布和订阅系统，具有高性能、高吞吐量、系统快速, 可扩展并且可持久化. 它的分区特性, 可复制和可容错都是其不错的特性而被广泛应用与大数据传输场景。它是由 LinkedIn 公司开发，使用 Scala 语言编写，之后成为 Apache 基金会的一个顶级项目。 kafka 提供了类似 JMS 的特性，但是在设计和实现上是完全不同的，而且他也不是 JMS 规范的实现。

kafka 产生的背景

kafka 作为一个消息系统，早起设计的目的是用作 LinkedIn 的活动流（ActivityStream）和运营数据处理管道（Pipeline）。活动流数据是所有的网站对用户的使用情况做分析的时候要用到的最常规的部分,活动数据包括页面的访问量（PV）、被查看内容方面的信息以及搜索内容。这种数据通常的处理方式是先
把各种活动以日志的形式写入某种文件，然后周期性的对这些文件进行统计分析。运营数据指的是服务器的性能数据（CPU、 IO 使用率、请求时间、服务日志等） 。

Kafka 的应用场景

由于 kafka 具有更好的吞吐量、内置分区、冗余及容错性的优点(kafka 每秒可以处理几十万消息)，让 kafka 成为了一个很好的大规模消息处理应用的解决方案。所以在企业级应用长，主要会应用于如下几个方面

• 行为跟踪： kafka 可以用于跟踪用户浏览页面、搜索及其他行为。通过发布-订阅模式实时记录到对应的 topic 中，通过后端大数据平台接入处理分析，并做更进一步的实时处理和监控

• 日志收集：日志收集方面，有很多比较优秀的产品，比如 Apache Flume，很多公司使用kafka 代理日志聚合。日志聚合表示从服务器上收集日志文件，然后放到一个集中的平台（文件服务器）进行处理。在实际应用开发中，我们应用程序的 log 都会输出到本地的磁盘上，排查问题的话通过 linux 命令来搞定，如果应用程序组成了负载均衡集群，并且集群的机器有几十台以上，那么想通过日志快速定位到问题，就是很麻烦的事情了。所以一般都会做一个日志统一收集平台管理 log 日志用来快速查询重要应用的问题。所以很多公司的套路都是把应用日志几种到 kafka 上，然后分别导入到 es 和 hdfs 上，用来做实时检索分析和离线统计数据备份等。而另一方面， kafka 本身又提供了很好的 api 来集成日志并且做日志收集

2.kafka的本身架构

一个典型的 kafka 集群包含若干 Producer（可以是应用节点产生的消息，也可以是通过Flume 收集日志产生的事件），若干个 Broker（kafka 支持水平扩展）、若干个 ConsumerGroup，以及一个 zookeeper 集群。 kafka 通过 zookeeper 管理集群配置及服务协同。Producer 使用 push 模式将消息发布到 broker， consumer 通过监听使用 pull 模式从broker 订阅并消费消息。多个 broker 协同工作， producer 和 consumer 部署在各个业务逻辑中。三者通过zookeeper 管理协调请求和转发。这样就组成了一个高性能的分布式消息发布和订阅系统。图上有一个细节是和其他 mq 中间件不同的点， producer 发送消息到 broker
的过程是 push，而 consumer 从 broker 消费消息的过程是 pull，主动去拉数
据。而不是 broker 把数据主动发送给 consumer

2.1kafka术语介绍

（1）Topics（主题）
属于特定类别的消息流称为主题。 数据存储在主题中。Topic相当于Queue。
主题被拆分成分区。 每个这样的分区包含不可变有序序列的消息。 分区被实现为具有相等大小的一组分段文件。
（2）Partition（分区）

kafka_Partition.jpg

• 一个Topic可以分成多个Partition，这是为了平行化处理。
• 每个Partition内部消息有序，其中每个消息都有一个offset序号。
• 一个Partition只对应一个Broker，一个Broker可以管理多个Partition。

（3）Partition offset（分区偏移）
每个分区消息具有称为 offset 的唯一序列标识。
（4）Replicas of partition（分区备份）
副本只是一个分区的备份。 副本从不读取或写入数据。 它们用于防止数据丢失。
（5）Brokers（经纪人）

• 代理是负责维护发布数据的简单系统。 每个代理可以每个主题具有零个或多个分区。 假设，如果在一个主题和N个代理中有N个分区，每个代理将有一个分区。
• 假设在一个主题中有N个分区并且多于N个代理(n + m)，则第一个N代理将具有一个分区，并且下一个M代理将不具有用于该特定主题的任何分区。
• 假设在一个主题中有N个分区并且小于N个代理(n-m)，每个代理将在它们之间具有一个或多个分区共享。 由于代理之间的负载分布不相等，不推荐使用此方案。

（6）Kafka Cluster（Kafka集群）
Kafka有多个代理被称为Kafka集群。 可以扩展Kafka集群，无需停机。 这些集群用于管理消息数据的持久性和复制。
（7）Producers（生产者）
生产者是发送给一个或多个Kafka主题的消息的发布者。 生产者向Kafka经纪人发送数据。 每当生产者将消息发布给代理时，代理只需将消息附加到最后一个段文件。实际上，该消息将被附加到分区。 生产者还可以向他们选择的分区发送消息。
（8）Consumers（消费者）
Consumers从经纪人处读取数据。 消费者订阅一个或多个主题，并通过从代理中提取数据来使用已发布的消息。

• Consumer自己维护消费到哪个offet
• 每个Consumer都有对应的group
• group内是queue消费模型：各个Consumer消费不同的partition，因此一个消息在group内只消费一次
• group间是publish-subscribe消费模型：各个group各自独立消费，互不影响，因此一个消息被每个group消费一次。

2.2 kafka架构图

kafka架构.png

3.kafka安装部署和基本操作

3.1 搭建

3.1.1 hosts 映射(可选, 建议)

echo "192.168.1.10 kfk1" >> /etc/hosts
echo "192.168.1.11 kfk2" >> /etc/hosts
echo "192.168.1.12 kfk3" >> /etc/hosts

3.1.2 搭建 Zookeeper (单机 or 集群)

1. 需要先启动 zookeeper，如果没有搭建 zookeeper 环境，可以直接运行
   kafka 内嵌的 zookeeper
   启动命令： bin/zookeeper-server-start.sh config/zookeeper.properties &

Broker, Producer, Consumer 的运行都需要 ZooKeeper

自己搭建见 [3.2.2.1 单机模式](#3.2.2.1 单机模式) or 见 [3.2.2.3 集群模式

3.1.3 Broker 的配置

tar xzf kafka_2.11-0.9.0.1.tgz -C /usr/local/
cd /usr/local/kafka_2.11-0.9.0.1

config 文件夹下是各个组件的配置文件, server.properties 是 Broker 的配置文件, 需要修改的有

######################### Server Basics #########################
broker.id=0                    # 本 Broker 的 id, 只要非负数且各 Broker 的 id 不同即可, 一般依次加 1

##################### Socket Server Settings #####################
listeners=PLAINTEXT://:9092    # Broker 监听的端口, Producer, Consumer 会连接这个端口
port=9092                      # 同上
host.name=kfk1                 # 本 Broker 的 hostname

########################## Topic Basics ##########################
delete.topic.enable=true       # 配置为可以使用 delete topic 命令

########################### Log Basics ###########################
log.dirs=/var/lib/kafka-logs   # log 目录, 此目录要存在且有足够权限

###################### Log Retention Policy ######################
log.roll.hours=2               # 开始一个新的 log 文件片段的最大时间
log.retention.hours=24         # 控制一个 log 文件保留多长个小时
log.retention.bytes=1073741824 # 所有 log 文件的最大大小
log.segment.bytes=104857600    # 单一的 log 文件最大大小
log.cleanup.policy=delete      # log 清除策略
log.retention.check.interval.ms=60000

############################ Zookeeper ############################
zookeeper.connect=zoo1:2181,zoo2:2181,zoo3:2181 # Zookeeper 的连接信息

注意: broker.id 和 host.name 在每台机器上是不一样的, 要按实际填写

即在 kfk2, kfk3 上

broker.id=1
host.name=kfk2

broker.id=2
host.name=kfk3

PS: 在 kafka 安装目录下的 ./site-docs 目录下有 kafka_config.html, producer_config.html, consumer_config.html 三个文件, 分别讲解 broker, producer, consumer 配置参数含义. 附录 A 是其翻译.

3.1.4 开放端口

把 Kafka 用到的端口开放出来

firewall-cmd --zone=public --add-port=9100/tcp --permanent     # 永久开启 9100 端口(kafka manager)
firewall-cmd --zone=public --add-port=9092/tcp --permanent     # 永久开启 9092 端口(brokers)
firewall-cmd --zone=public --add-port=9999/tcp --permanent     # 永久开启 9999 端口(JMX)
firewall-cmd --reload                                          # 重新加载防火墙规则

3.1.5 Broker 运行与终止

Broker 运行与终止命令如下.

# 运行: 如下有两种方式
# 以守护进程的方式启动(推荐)
bin/kafka-server-start.sh -daemon config/server.properties
# 将 Broker 放到后台执行, 且不受终端关闭的影响, 标准输出和错误输出定向到 `./logs/kafka-server-boot.log`, 有问题时可以去看这个文件
nohup bin/kafka-server-start.sh config/server.properties > logs/kafka-server-boot.log 2>&1 &

# 终止
bin/kafka-server-stop.sh config/server.properties

3.1.6 测试

我们使用 Kafka 自带的基于 终端 的 Producer 和 Consumer 脚本做测试.

先只启动一台机器上的 Broker. 在 kfk1 上运行

nohup bin/kafka-server-start.sh config/server.properties > logs/kafka-server-boot.log 2>&1 &

1. 创建 Topic

创建一个 名为"TestCase"的 单分区 单副本 的 Topic.

bin/kafka-topics.sh --create --zookeeper localhost:2181 --topic TestCase --replication-factor 1 --partitions 1

查看有哪些 Topic:

$ bin/kafka-topics.sh --list --zookeeper localhost:2181
TestCase

运行describe topics命令, 可以知道 Topic 的具体分配:

$ bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic test
Topic:test    PartitionCount:1    ReplicationFactor:1    Configs:
    Topic: test    Partition: 0    Leader: 0    Replicas: 0    Isr: 0

解释一下输出的内容. 第一行给出了所有 partition 的一个摘要, 每行给出一个 partition 的信息. 因为我们这个 topic 只有一个 partition 所以只有一行信息.

• "leader" 负责所有 partition 的读和写请求的响应. "leader" 是随机选定的.
• "replicas" 是备份节点列表, 包含所有复制了此 partition log 的节点, 不管这个节点是否为 leader 也不管这个节点当前是否存活, 只是显示.
• "isr" 是当前处于同步状态的备份节点列表. 即 "replicas" 列表中处于存活状态并且与 leader 一致的节点.

可以发现这个 Topic 没有副本而且它在 [我们创建它时集群仅有的一个节点] Broker 0 上.

另外, 除去手工创建 Topic 以外, 你也可以将你的 Brokers 配置成当消息发布到一个不存在的 Topic 时自动创建此 Topic.

2. 启动 生产者

Kafka 附带一个 终端生产者 可以从文件或者标准输入中读取输入然后发送这个消息到 Kafka 集群. 默认情况下每行信息被当做一个消息发送.

运行生产者脚本然后在终端中输入一些消息, 即可发送到 Broker.

$ bin/kafka-console-producer.sh --broker-list localhost:9092 --topic TestCase
This is a message
This is another message

PS: 通过键入 Ctrl-C 来终止终端生产者.

3. 启动 消费者

Kafka 也附带了一个 终端生产者 可以导出这些消息到标准输出.

$ bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic TestCase --from-beginning
This is a message
This is another message

--from-beginning 参数使得可以接收到 topic 的所有消息, 包括 consumer 启动前的. 去掉后则为仅接收 consumer 启动后 kafka 收到的消息.

如果你在不同的终端运行生产者和消费者这两个命令, 那么现在你就应该能再生产者的终端中键入消息同时在消费者的终端中看到.

所有的命令行工具都有很多可选的参数; 不添加参数直接执行这些命令将会显示它们的使用方法, 更多内容可以参考他们的手册.

PS: 通过键入 Ctrl-C 来终止终端消费者.

4. 配置一个多节点集群

我们已经成功的以单 Broker 的模式运行起来了, 但这并没有意思. 对于 Kafka 来说, 一个单独的 Broker 就是一个大小为 1 的集群, 所以集群模式就是多启动几个 Broker 实例.

我们将我们的集群扩展到3个节点. 在另外两台机器 kfk2, kfk3 上运行

nohup bin/kafka-server-start.sh config/server.properties > logs/kafka-server-boot.log 2>&1 &

现在我们可以创建一个新的 Topic 并制定副本数量为 3:

bin/kafka-topics.sh --create --zookeeper localhost:2181 --topic my-replicated-topic --replication-factor 3 --partitions 1

运行describe topics命令, 可以知道每个 Broker 具体的工作:

$ bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic my-replicated-topic
Topic:my-replicated-topic    PartitionCount:1    ReplicationFactor:3    Configs:
    Topic: my-replicated-topic    Partition: 0    Leader: 1    Replicas: 1,2,0    Isr: 1,2,0

注意本例中 Broker 1 是这个有一个 partition 的 topic 的 leader.

现在我们发布几个消息到我们的新 topic 上:

$ bin/kafka-console-producer.sh --broker-list localhost:9092 --topic my-replicated-topic
my test message 1
my test message 2

现在让我们消费这几个消息:

$ bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic my-replicated-topic --from-beginning
my test message 1
my test message 2

现在让我们测试一下集群容错. Broker 1 正在作为 leader, 所以我们杀掉它:

$ ps | grep server.properties
...
7564 ttys002    0:15.91 /System/Library/Frameworks/JavaVM.framework/Versions/1.8/Home/bin/java...
...

$ kill -9 7564

或在 kfk1 机器上运行

bin/kafka-server-stop.sh config/server.properties

此时, 集群领导已经切换到一个从服务器上, Broker 1 节点也不在出现在同步副本列表中了:

$ bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic my-replicated-topic
Topic:my-replicated-topic    PartitionCount:1    ReplicationFactor:3    Configs:
    Topic: my-replicated-topic    Partition: 0    Leader: 2    Replicas: 1,2,0    Isr: 2,0

而且现在消息的消费仍然能正常进行, 即使原来负责写的节点已经失效了.

$ bin/kafka-console-consumer.sh --zookeeper localhost:2181 --from-beginning --topic my-replicated-topic
...
my test message 1
my test message 2

5. 使用 Kafka Connect 进行数据导入导出

从终端写入数据, 数据也写回终端是默认的. 但是你可能希望从一些其它的数据源或者导出 Kafka 的数据到其它的系统. 相比其它系统需要自己编写集成代码, 你可以直接使用Kafka的 Connect 直接导入或者导出数据. Kafka Connect 是 Kafka 自带的用于数据导入和导出的工具. 它是一个扩展的可运行连接器(runsconnectors)工具, 可实现自定义的逻辑来实现与外部系统的集成交互. 在这个快速入门中我们将介绍如何通过一个简单的从文本导入数据, 导出数据到文本的连接器来调用 Kafka Connect. 首先我们从创建一些测试的基础数据开始:

echo -e "foo\nbar" > test.txt

接下来我们采用standalone模式启动两个 connectors, 也就是让它们都运行在独立的, 本地的, 不同的进程中. 我们提供三个参数化的配置文件, 第一个提供共有的配置用于 Kafka Connect 处理, 包含共有的配置比如连接哪个 Kafka broker 和数据的序列化格式. 剩下的配置文件制定每个 connector 创建的特定信息. 这些文件包括唯一的 connector 的名字, connector 要实例化的类和其它的一些 connector 必备的配置.

bin/connect-standalone.sh config/connect-standalone.properties config/connect-file-source.properties config/connect-file-sink.properties

上述简单的配置文件已经被包含在 Kafka 的发行包中, 它们将使用默认的之前我们启动的本地集群配置创建两个 connector: 第一个作为源 connector 从一个文件中读取每行数据然后将他们发送 Kafka 的 topic, 第二个是一个输出(sink)connector 从 Kafka 的 topic 读取消息, 然后将它们输出成输出文件的一行行的数据. 在启动的过程你讲看到一些日志消息, 包括一些提示 connector 正在被实例化的信息. 一旦 Kafka Connect 进程启动以后, 源 connector 应该开始从 test.txt 中读取数据行, 并将他们发送到 topic connect-test 上, 然后输出 connector 将会开始从 topic 读取消息然后把它们写入到 test.sink.txt 中.

我们可以查看输出文件来验证通过整个管线投递的数据:

$ cat test.sink.txt
foo
bar

注意这些数据已经被保存到了 Kafka 的 connect-test topic 中, 所以我们还可以运行一个终端消费者来看到这些数据(或者使用自定义的消费者代码来处理数据):

$ bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic connect-test --from-beginning
{"schema":{"type":"string","optional":false},"payload":"foo"}
{"schema":{"type":"string","optional":false},"payload":"bar"}
...

connector 在持续的处理着数据, 所以我们可以向文件中添加数据然后观察到它在这个管线中的传递:

echo "Another line" >> test.txt

你应该可以观察到新的数据行出现在终端消费者中和输出文件中.

3.2 Zookeeper

3.2.1 简介

Apache Zookeeper 是 Hadoop 的一个子项目, 是一个致力于开发和管理开源服务器, 并且能实现高可靠性的分布式协调框架. 它包含一个简单的原语集, 分布式应用程序可以基于它实现同步服务, 配置维护和命名服务等.

Zookeeper 保证 2n + 1 台机器的集群最大允许 n 台机器挂掉而事务不中断.

3.2.2 搭建

3.2.2.1 单机模式

此模式主要用于开发人员本地环境下测试代码

1. 解压 Zookeeper 并进入其根目录

tar xzf zookeeper-3.4.9.tar.gz -C /usr/local/
cd /usr/local/zookeeper-3.4.9

2. 创建配置文件 conf/zoo.cfg

cp conf/zoo_sample.cfg conf/zoo.cfg

3. 修改内容如下:

tickTime=2000
initLimit=10
syncLimit=5
dataDir=/var/lib/zookeeper/data
dataLogDir=/var/lib/zookeeper/logs
clientPort=2181

• tickTime: 是 zookeeper 的最小时间单元的长度(以毫秒为单位), 它被用来设置心跳检测和会话最小超时时间(tickTime 的两倍)
• initLimit: 初始化连接时能容忍的最长 tickTime 个数
• syncLimit: follower 用于同步的最长 tickTime 个数
• dataDir: 服务器存储 数据快照 的目录
• dataLogDir: 服务器存储 事务日志 的目录
• clientPort: 用于 client 连接的 server 的端口

其中需要注意的是dataDir和dataLogDir, 分别是 zookeeper 运行时的数据目录和日志目录, 要保证 这两个目录已创建 且 运行 zookeeper 的用户拥有这两个目录的所有权

4.测试

• 启动/关闭 Zookeeper:

bin/zkServer.sh start
bin/zkServer.sh stop

• 查看 Zookeeper 状态:

bin/zkServer.sh status

显示 mode: standalone, 单机模式.

• 使用 java 客户端连接 ZooKeeper

./bin/zkCli.sh -server 127.0.0.1:2181

然后就可以使用各种命令了, 跟文件操作命令很类似, 输入 help 可以看到所有命令.

3.2.2.2 集群模式

此模式是 生产环境中实际使用的模式

因为 zookeeper 保证 2n + 1 台机器最大允许 n 台机器挂掉, 所以配置集群模式最好是奇数台机器: 3, 5, 7...

最少 3 台构成集群

1 hosts 映射(可选)

echo "192.168.1.1 zoo1" >> /etc/hosts
echo "192.168.1.2 zoo2" >> /etc/hosts
echo "192.168.1.3 zoo3" >> /etc/hosts

2 修改 zookeeper-3.4.9/conf/zoo.cfg 文件

tickTime=2000
initLimit=10
syncLimit=5
dataDir=/var/lib/zookeeper/data
dataLogDir=/var/lib/zookeeper/log
clientPort=2181
server.1=192.168.1.1:2888:3888
server.2=192.168.1.2:2888:3888
server.3=192.168.1.3:2888:3888

与单机模式的不同就是最后三条: server.X=host:portA:portB

server.1=192.168.1.1:2888:3888
server.2=192.168.1.2:2888:3888
server.3=192.168.1.3:2888:3888

或

server.1=zoo1:2888:3888
server.2=zoo2:2888:3888
server.3=zoo3:2888:3888

X 为标识为 X 的机器, host 为其 hostname 或 IP, portA 用于这台机器与集群中的 Leader 机器通信, portB 用于 server 选举 leader.

PS: 要配单机伪分布式的话, 可以修改这里为

server.1=localhost:2888:3888
server.2=localhost:2889:3889
server.3=localhost:2890:3890

然后每个 zookeeper 实例的 dataDir 和 dataLogDir 配置为不同的即可

3 myid 文件

在标示为 X 的机器上, 将 X 写入 ${dataDir}/myid 文件, 如: 在 192.168.1.2 机器上的 /var/lib/zookeeper/data 目录下建立文件 myid, 写入 2

echo "2" > /var/lib/zookeeper/data/myid

4 开放端口

CentOS 7 使用 firewalld 代替了原来的 iptables, 基本使用如下

systemctl start firewalld                                      # 启动防火墙
firewall-cmd --state                                           # 检查防火墙状态
firewall-cmd --zone=public --add-port=2888/tcp --permanent     # 永久开启 2888 端口
firewall-cmd --reload                                          # 重新加载防火墙规则
firewall-cmd --list-all                                        # 列出所有防火墙规则

把 Zookeeper 用到的端口开放出来

firewall-cmd --zone=public --add-port=2181/tcp --permanent     # 永久开启 2181 端口
firewall-cmd --zone=public --add-port=2888/tcp --permanent     # 永久开启 2888 端口
firewall-cmd --zone=public --add-port=3888/tcp --permanent     # 永久开启 3888 端口
firewall-cmd --reload                                          # 重新加载防火墙规则

5 测试

• 在 集群中所有机器上 启动 zookeeper(尽量同时):

bin/zkServer.sh start

• 查看状态, 应该有一台机器显示mode: leader, 其余为mode: follower

bin/zkServer.sh status

• 使用 java 客户端连接 ZooKeeper

./bin/zkCli.sh -server 192.168.1.1:2181

然后就可以使用各种命令了, 跟文件操作命令很类似, 输入help可以看到所有命令.

• 关闭 zookeeper:

./bin/zkServer.sh stop

3.2.3 Zookeeper 常见问题

查看状态时, 应该有一台机器显示mode: leader, 其余为mode: follower

bin/zkServer.sh status

当显示Error contacting service. It is probably not running.时, 可以查看日志

cat zookeeper.out

查看 zookeeper.out 日志可以看到是那些机器连不上, 可能是 网络, ip, 端口, 配置文件, myid 文件 的问题.
正常应该是: 先是一些 java 异常, 这是因为 ZooKeeper 集群启动的时候, 每个结点都试图去连接集群中的其它结点, 先启动的肯定连不上后面还没启动的, 所以上面日志前面部分的异常是可以忽略的, 当集群所有的机器的 zookeeper 都启动起来, 就没有异常了, 并选举出来了 leader.

PS: 因为 zkServer.sh 脚本中是用 nohup 命令启动 zookeeper 的, 所以 zookeeper.out 文件是在调用 zkServer.sh 时的路径下, 如:用 bin/zkServer.sh start 启动则 zookeeper.out 文件在 zookeeper-3.4.9/ 下; 用 zkServer.sh start 启动则 zookeeper.out 文件在 zookeeper-3.4.9/bin/ 下.

4.kafka的应用

未完待续！！