1 简介

Apache Kafka 是一个分布式、多分区、多副本，基于Zookeeper协调的分布式消息系统和流式处理平台。它有高吞吐、可持久化、可水平扩展、支持流数据处理等多种特点。（最初是由LinkedIn公司开发，现已被捐献给Apache基金会）。

三种系统特性：

• 作为消息系统，可以发布和订阅流式记录。和传统消息中间件都具备系统解耦、冗余、异步通信、流量削峰、扩展性、可恢复性等功能。此外，Kafka还提供了保障消息顺序性及回溯消费的功能。
• 作为存储系统，可以把消息持久化到磁盘，并且进行备份以便容错。
• 作为流式处理平台，可以进行实时的数据流式处理。

2 基本概念

2.1 Kafka体系结构

Kafka体系结构

• Producer：生产者，发送消息的一端。创建信息并且投递信息到Kafka Broker中。
• Consumer：消费者，消费消息的一端。从Kafka Broker拉取（pull）订阅的消息，并进行相应的业务逻辑处理。
• Broker：服务代理节点。负责将接收到的消息存储到磁盘中，可以是一个独立的Kafka服务节点或者服务实例。
  Kafka典型的体系架构包括若干个Producer、若干个Broker、若干个Consumer。Kafka通过Zookeeper来负责集群元数据的管理、控制器选举等操作。

2.2 主题和分区

Kafka核心概念：主题（Topic）和分区（Partition）。
每一条发送到Kafka集群中的消息都需要指定一个Topic，用来归类消息。Topic是一个逻辑上的概念，用来区分不同的业务消息。

• 一个Topic可以细分为多个分区，且每个分区包含的消息是不同的。
• 一个分区只属于单个Topic。
• 同一个主题的分区可以分布在不同的broker上。
• 每个主题的分区可以看作一个可追加的日志文件。消息在被追加到分区日志文件时都会分配一个特定的偏移量（offset），offset是消息在分区的唯一标识。通过它保证了消息在分区内的顺序性。不保证消息在Topic内有序。
• Kafka消费端具备一定的容灾能力，从broker服务节点pull消息并且会保存消费的具体位置（offset），当消费者宕机恢复上线时可以根据之前保存的消费位置重新拉取消息进行消费，保证消息不会丢失。

消息追加到分区日志文件

分区的用途

• 当文件日志大小超过了单台服务器的限制时，可通过增加分区数量实现水平扩展解决。
• 当文件所在的机器I/O成为这个主题的性能瓶颈时，也可通过增加分区数量实现水平扩展解决。

2.3 副本

Kafka为分区引入了多副本（Replica）机制。

• 一个分区有一个或多个副本。同一分区的不同副本保存的是相同的信息（大部分时候）。
• 副本分为一主多从。leader副本负责对外服务，处理与生产者/消费者的读写请求。follower副本只负责与leader副本的消息同步。

• Kafka通过多副本机制实现故障自动转移。副本分布于不同broker中，若leader副本出现故障时，会从follower副本中重新选举新的leader副本对外提供服务。

  
  多分区多副本结构图

消息复制机制

先解释几个概念：

• ISR（In-Sync Replicas）：分区中所有与leader副本保持一定程度同步的副本，包括leader副本。
• OSR（Out-of-Sync Replicas）：分区中与leader副本同步滞后过多的副本。
• “一定程度的同步”：指可忍受的滞后范围，这个范围可通过参数【follower副本复制消息落后leader副本的最大条数】（replica.lag.max.messages）或【follower副本未向leader副本发送fetch请求的最大时间】（replica.lag.time.max.ms）进行配置。
• AR（Assigned Replicas）：分区中的所有副本。AR=ISR+OSR。

消息复制机制：消息会先发送到leader副本，然后follower副本从leader副本拉取消息进行同步，如果follower副本落后太多或失效时，那么leader副本会把它从ISR集合中移除。如果OSR集合中有follower副本追上了leader副本，那么leader副本会将它转移至ISR集合。默认情况下，当leader副本发生故障时，只有ISR集合中的follower副本才有资格被选举为新的leader，保证了数据可靠性。

HW与LEO

• HW（High Watermark）：高水位。标识了一个特定的消息offset，消费者只能拉取到的这个offset之前的消息。
• LEO（Log End Offset）：日志末尾偏移量。标识当前日志文件中下一条待写入消息的offset。
• 每个副本都会维护自身的LEO。而ISR集合中最小的LEO即为分区HW。所以消费者只能拉取ISR集合中最小的LEO之前的消息。

HW和LEO

3 安装及配置

Kafka运行在JVM之上且依赖于Zookeeper，需要先安装JDK及Zookeeper（我这里不再描述）。Kafka 2.0版本之后不再支持JDK7及以下版本，本文环境基于Mac系统、采用JDK8来示例。

3.1 下载安装包

• 首先从官网下载安装包，下载地址：http://kafka.apache.org/downloads
  ，这里我选用的安装版本是kafka_2.13-2.6.0.tgz。下载后将其复制到工作目录解压。

# 解压
tar zxvf kafka_2.13-2.6.0.tgz 

3.2 修改配置

修改broker的配置文件$KAFKA_HOME/conf/server.properties。

# 此broker的id，如果集群部署，每个broker需要设置一个唯一id。
broker.id=0

# broker提供对外的服务地址入口，默认取java.net.InetAddress.getCanonicalHostName()返回的值
listeners=PLAINTEXT://:9092

# 存放日志文件的目录，如果有多个日志目录，则用逗号隔开
log.dirs=/tmp/kafka-logs

# 每个主题的日志分区个数
num.partitions=1

# 连接zk地址，如果是zk集群使用host:port对，用逗号分隔，例如 "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002"
# 可以追加个chroot，作为所有kafka znode的根目录
zookeeper.connect=localhost:2181/kafka

# 连接zk超时时间，单位ms
zookeeper.connection.timeout.ms=18000

如果是单机模式，可以修改为上面的配置即可启动服务。
如果是集群模式，只需修改每个broker对应的broker.id及listeners。

3.3 启动Kafka服务

# 在$KAFKA_HOME目录执行启动命令
bin/kafka-server-start.sh config/server.properties 

# 后台运行Kafka服务，追加&方式
bin/kafka-server-start.sh config/server.properties &
# 后台运行Kafka服务，加-daemon参数方式
bin/kafka-server-start.sh -daemon config/server.properties

Kafka 服务启动成功，如下图所示。

服务启动成功

后台运行Kafka服务，可以通过jps命令查看服务进程是否已经启动。如下图所示出现Kafka，表示Kafka服务已启动。

jps查看启动的JVM进程

4 消息生产与消费

在$KAFKA_HOME的bin目录下，存放着许多实用的脚本工具，我们通过这些小工具来演示消息的生产和消费过程。

第一步 使用kafka-topics.sh创建名称为topic_test的topic。

# --replication-factor指定副本因子，--partitions指定分区个数，--zookeeper指定连接的zk地址，--topic指定创建的主题
# 创建主题topic_test,包括3个分区，每个分区副本为1个。
bin/kafka-topics.sh --create --topic topic_test --replication-factor 1 --partitions 3 --zookeeper localhost:2181/kafka

# 使用--describe指令查看创建的topic详细信息
bin/kafka-topics.sh --describe --topic topic_test  --zookeeper localhost:2181/kafka

第二步 打开一个shell终端，使用kafka-console-consumer.sh模拟消费者订阅主题topic_test。

# --bootstrap-server指定连接的Kafka服务地址， --topic指定订阅的主题
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic topic_test 

第三步 再打开一个shell终端，使用kafka-console-producer.sh模拟消息的生产, 指定投递的消息主题名topic_test。

# --broker-list指定连接的Kafka服务地址， --topic指定投递的消息主题
bin/kafka-console-producer.sh --broker-list localhost:9092 --topic topic_test

在生产端控制台输入消息“Hello”和“World”并且换行。切换到kafka-console-consumer.sh终端出现了刚刚输入的消息“Hello”和“World”。如下图所示，演示了消息的生产和消费。

生产消息 消费消息

至此，对Kafka的概念、服务的搭建、消息的生产和消费过程等有了个初步的认知，这仅仅只是一个开始，后续会深入探索Kafka其他功能及用法。