Kafka 快速搭建

admin · 发表于 2022-12-30 09:08:01

在了解完 Kafka 的基本概念之后，我们通过搭建 Kafka 集群来进一步深刻认识一下 Kafka。

确保安装环境安装 Java 环境

在安装 Kafka 之前，先确保Linux 环境上是否有 Java 环境，使用 java -version 命令查看 Java 版本，推荐使用Jdk 1.8 ，如果没有安装 Java 环境的话，可以按照这篇文章进行安装（https://www.cnblogs.com/zs-notes/p/8535275.html）

安装 Zookeeper 环境

Kafka 的底层使用 Zookeeper 储存元数据，确保一致性，所以安装 Kafka 前需要先安装 Zookeeper，Kafka 的发行版自带了 Zookeeper ，可以直接使用脚本来启动，不过安装一个 Zookeeper 也不费劲

Zookeeper 单机搭建

Zookeeper 单机搭建比较简单，直接从 https://www.apache.org/dyn/closer.cgi/zookeeper/ 官网下载一个稳定版本的 Zookeeper ，这里我使用的是 3.4.10，下载完成后，在 Linux 系统中的 /usr/local 目录下创建 zookeeper 文件夹，使用xftp 工具(xftp 和 xshell 工具都可以在官网 https://www.netsarang.com/zh/xshell/ 申请免费的家庭版)把下载好的 zookeeper 压缩包放到 /usr/local/zookeeper 目录下。

如果下载的是一个 tar.gz 包的话，直接使用 tar -zxvf zookeeper-3.4.10.tar.gz解压即可

如果下载的是 zip 包的话，还要检查一下 Linux 中是否有 unzip 工具，如果没有的话，使用 yum install unzip 安装 zip 解压工具，完成后使用 unzip zookeeper-3.4.10.zip 解压即可。

解压完成后，cd 到 /usr/local/zookeeper/zookeeper-3.4.10 ，创建一个 data 文件夹，然后进入到 conf 文件夹下，使用 mv zoo_sample.cfg zoo.cfg 进行重命名操作

然后使用 vi 打开 zoo.cfg ，更改一下dataDir = /usr/local/zookeeper/zookeeper-3.4.10/data ，保存。

进入bin目录，启动服务输入命令 ./zkServer.sh start 输出下面内容表示搭建成功

关闭服务输入命令，./zkServer.sh stop

使用 ./zkServer.sh status 可以查看状态信息。

Zookeeper 集群搭建准备条件

准备条件：需要三个服务器，这里我使用了CentOS7 并安装了三个虚拟机，并为各自的虚拟机分配了1GB的内存，在每个 /usr/local/ 下面新建 zookeeper 文件夹，把 zookeeper 的压缩包挪过来，解压，完成后会有 zookeeper-3.4.10 文件夹，进入到文件夹，新建两个文件夹，分别是 data 和 log 文件夹

注：上一节单机搭建中已经创建了一个data 文件夹，就不需要重新创建了，直接新建一个 log 文件夹，对另外两个新增的服务需要新建这两个文件夹。

设置集群

新建完成后，需要编辑 conf/zoo.cfg 文件，三个文件的内容如下

1
2
3
4
5
6
7
8
9

tickTime=2000initLimit=10syncLimit=5dataDir=/usr/local/zookeeper/zookeeper-3.4.10/datadataLogDir=/usr/local/zookeeper/zookeeper-3.4.10/logclientPort=12181server.1=192.168.1.7:12888:13888server.2=192.168.1.8:12888:13888server.3=192.168.1.9:12888:13888

server.1 中的这个 1 表示的是服务器的标识也可以是其他数字，表示这是第几号服务器，这个标识要和下面我们配置的 myid 的标识一致可以。

192.168.1.7:12888:13888 为集群中的 ip 地址，第一个端口表示的是 master 与 slave 之间的通信接口，默认是 2888，第二个端口是leader选举的端口，集群刚启动的时候选举或者leader挂掉之后进行新的选举的端口，默认是 3888

现在对上面的配置文件进行解释

tickTime: 这个时间是作为 Zookeeper 服务器之间或客户端与服务器之间维持心跳的时间间隔，也就是每个 tickTime 时间就会发送一个心跳。

initLimit：这个配置项是用来配置 Zookeeper 接受客户端（这里所说的客户端不是用户连接 Zookeeper 服务器的客户端，而是 Zookeeper 服务器集群中连接到 Leader 的 Follower 服务器）初始化连接时最长能忍受多少个心跳时间间隔数。当已经超过 5个心跳的时间（也就是 tickTime）长度后 Zookeeper 服务器还没有收到客户端的返回信息，那么表明这个客户端连接失败。总的时间长度就是 5*2000=10 秒

syncLimit: 这个配置项标识 Leader 与Follower 之间发送消息，请求和应答时间长度，最长不能超过多少个 tickTime 的时间长度，总的时间长度就是5*2000=10秒

dataDir: 快照日志的存储路径

dataLogDir: 事务日志的存储路径，如果不配置这个那么事务日志会默认存储到dataDir指定的目录，这样会严重影响zk的性能，当zk吞吐量较大的时候，产生的事务日志、快照日志太多

clientPort: 这个端口就是客户端连接 Zookeeper 服务器的端口，Zookeeper 会监听这个端口，接受客户端的访问请求。

创建 myid 文件

在了解完其配置文件后，现在来创建每个集群节点的 myid ，我们上面说过，这个 myid 就是 server.1 的这个 1 ，类似的，需要为集群中的每个服务都指定标识，使用 echo 命令进行创建

1
2
3
4
5
6

# server.1echo "1" > /usr/local/zookeeper/zookeeper-3.4.10/data/myid# server.2echo "2" > /usr/local/zookeeper/zookeeper-3.4.10/data/myid# server.3echo "3" > /usr/local/zookeeper/zookeeper-3.4.10/data/myid

启动服务并测试

配置完成，为每个 zk 服务启动并测试，我在 windows 电脑的测试结果如下

启动服务（每台都需要执行）

1
2

cd /usr/local/zookeeper/zookeeper-3.4.10/bin./zkServer.sh start

检查服务状态

使用 ./zkServer.sh status 命令检查服务状态

192.168.1.7 — follower

192.168.1.8 — leader

192.168.1.9 — follower

zk集群一般只有一个leader，多个follower，主一般是相应客户端的读写请求，而从主同步数据，当主挂掉之后就会从follower里投票选举一个leader出来。

Kafka 集群搭建准备条件

搭建好的 Zookeeper 集群
Kafka 压缩包（https://www.apache.org/dyn/close ... afka_2.12-2.3.0.tgz）
( F1 j# f1 ]$ t

在 /usr/local 下新建 kafka 文件夹，然后把下载完成的 tar.gz 包移到 /usr/local/kafka 目录下，使用 tar -zxvf 压缩包进行解压，解压完成后，进入到 kafka_2.12-2.3.0 目录下，新建 log 文件夹，进入到 config 目录下

我们可以看到有很多 properties 配置文件，这里主要关注 server.properties 这个文件即可。

kafka 启动方式有两种，一种是使用 kafka 自带的 zookeeper 配置文件来启动（可以按照官网来进行启动，并使用单个服务多个节点来模拟集群http://kafka.apache.org/quickstart#quickstart_multibroker），一种是通过使用独立的zk集群来启动，这里推荐使用第二种方式，使用 zk 集群来启动

修改配置项

需要为每个服务都修改一下配置项，也就是server.properties，需要更新和添加的内容有

1
2
3
4
5
6
7
8
9
10

broker.id=0 //初始是0，每个 server 的broker.id 都应该设置为不一样的，就和 myid 一样我的三个服务分别设置的是 1,2,3log.dirs=/usr/local/kafka/kafka_2.12-2.3.0/log#在log.retention.hours=168 下面新增下面三项message.max.byte=5242880default.replication.factor=2replica.fetch.max.bytes=5242880#设置zookeeper的连接端口zookeeper.connect=192.168.1.7:2181,192.168.1.8:2181,192.168.1.9:2181

配置项的含义

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

broker.id=0 #当前机器在集群中的唯一标识，和zookeeper的myid性质一样port=9092 #当前kafka对外提供服务的端口默认是9092host.name=192.168.1.7 #这个参数默认是关闭的，在0.8.1有个bug，DNS解析问题，失败率的问题。num.network.threads=3 #这个是borker进行网络处理的线程数num.io.threads=8 #这个是borker进行I/O处理的线程数log.dirs=/usr/local/kafka/kafka_2.12-2.3.0/log #消息存放的目录，这个目录可以配置为“，”逗号分割的表达式，上面的num.io.threads要大于这个目录的个数这个目录，如果配置多个目录，新创建的topic他把消息持久化的地方是，当前以逗号分割的目录中，那个分区数最少就放那一个socket.send.buffer.bytes=102400 #发送缓冲区buffer大小，数据不是一下子就发送的，先回存储到缓冲区了到达一定的大小后在发送，能提高性能socket.receive.buffer.bytes=102400 #kafka接收缓冲区大小，当数据到达一定大小后在序列化到磁盘socket.request.max.bytes=104857600 #这个参数是向kafka请求消息或者向kafka发送消息的请请求的最大数，这个值不能超过java的堆栈大小num.partitions=1 #默认的分区数，一个topic默认1个分区数log.retention.hours=168 #默认消息的最大持久化时间，168小时，7天message.max.byte=5242880 #消息保存的最大值5Mdefault.replication.factor=2 #kafka保存消息的副本数，如果一个副本失效了，另一个还可以继续提供服务replica.fetch.max.bytes=5242880 #取消息的最大直接数log.segment.bytes=1073741824 #这个参数是：因为kafka的消息是以追加的形式落地到文件，当超过这个值的时候，kafka会新起一个文件log.retention.check.interval.ms=300000 #每隔300000毫秒去检查上面配置的log失效时间（log.retention.hours=168 ），到目录查看是否有过期的消息如果有，删除log.cleaner.enable=false #是否启用log压缩，一般不用启用，启用的话可以提高性能zookeeper.connect=192.168.1.7:2181,192.168.1.8:2181,192.168.1.9:2181 #设置zookeeper的连接端口

启动 Kafka 集群并测试

启动服务，进入到 /usr/local/kafka/kafka_2.12-2.3.0/bin 目录下
% } Y2 v( D9 C }- B" F1 q

1
2

# 启动后台进程./kafka-server-start.sh -daemon ../config/server.properties

检查服务是否启动

1
2
3
4

# 执行命令 jps6201 QuorumPeerMain7035 Jps6972 Kafka

kafka 已经启动
创建 Topic 来验证是否创建成功. @7 g# v9 B( v& d2 _% n( h

1
2

# cd .. 往回退一层到 /usr/local/kafka/kafka_2.12-2.3.0 目录下bin/kafka-topics.sh --create --zookeeper 192.168.1.7:2181 --replication-factor 2 --partitions 1 --topic cxuan

对上面的解释

–replication-factor 2 复制两份
–partitions 1 创建1个分区
–topic 创建主题

查看我们的主题是否出创建成功

1	bin/kafka-topics.sh --list --zookeeper 192.168.1.7:2181

启动一个服务就能把集群启动起来

在一台机器上创建一个发布者

1
2

# 创建一个broker，发布者./kafka-console-producer.sh --broker-list 192.168.1.7:9092 --topic cxuantopic

在一台服务器上创建一个订阅者

1
2

# 创建一个consumer，消费者bin/kafka-console-consumer.sh --bootstrap-server 192.168.1.7:9092 --topic cxuantopic --from-beginning

注意：这里使用 –zookeeper 的话可能出现 zookeeper is not a recognized option 的错误，这是因为 kafka 版本太高，需要使用 --bootstrap-server 指令

测试结果

发布

消费

其他命令

显示 topic

1
2
3
4

bin/kafka-topics.sh --list --zookeeper 192.168.1.7:2181# 显示cxuantopic

查看 topic 状态

1
2
3
4
5
6
7
8

bin/kafka-topics.sh --describe --zookeeper 192.168.1.7:2181 --topic cxuantopic# 下面是显示的详细信息Topic:cxuantopic PartitionCount:1 ReplicationFactor:2 Configs:Topic: cxuantopic Partition: 0 Leader: 1 Replicas: 1,2 Isr: 1,2# 分区为为1 复制因子为2 主题 cxuantopic 的分区为0 # Replicas: 0,1 复制的为1，2

Leader 负责给定分区的所有读取和写入的节点，每个节点都会通过随机选择成为 leader。

Replicas 是为该分区复制日志的节点列表，无论它们是 Leader 还是当前处于活动状态。

Isr 是同步副本的集合。它是副本列表的子集，当前仍处于活动状态并追随Leader。

至此，kafka 集群搭建完毕。

验证多节点接收数据

刚刚我们都使用的是相同的ip 服务，下面使用其他集群中的节点，验证是否能够接受到服务

在另外两个节点上使用

1	bin/kafka-console-consumer.sh --bootstrap-server 192.168.1.7:9092 --topic cxuantopic --from-beginning

然后再使用 broker 进行消息发送，经测试三个节点都可以接受到消息。

配置详解

在搭建 Kafka 的时候我们简单介绍了一下 server.properties 中配置的含义，现在我们来详细介绍一下参数的配置和概念

常规配置

这些参数是 kafka 中最基本的配置

broker.id
0 I* @1 C; U$ q! z |! _

每个 broker 都需要有一个标识符，使用 broker.id 来表示。它的默认值是 0，它可以被设置成其他任意整数，在集群中需要保证每个节点的 broker.id 都是唯一的。

port! b$ V# t+ E b `% Z. {

如果使用配置样本来启动 kafka ，它会监听 9092 端口，修改 port 配置参数可以把它设置成其他任意可用的端口。

zookeeper.connect5 l/ H G' k) s. V

用于保存 broker 元数据的地址是通过 zookeeper.connect 来指定。 localhost:2181 表示运行在本地 2181 端口。该配置参数是用逗号分隔的一组 hostname:port/path 列表，每一部分含义如下：

hostname 是 zookeeper 服务器的服务名或 IP 地址

port 是 zookeeper 连接的端口

/path 是可选的 zookeeper 路径，作为 Kafka 集群的 chroot 环境。如果不指定，默认使用跟路径

log.dirs. m$ R1 i% f6 \5 r# _

Kafka 把消息都保存在磁盘上，存放这些日志片段的目录都是通过 log.dirs 来指定的。它是一组用逗号分隔的本地文件系统路径。如果指定了多个路径，那么 broker 会根据 “最少使用” 原则，把同一分区的日志片段保存到同一路径下。要注意，broker 会向拥有最少数目分区的路径新增分区，而不是向拥有最小磁盘空间的路径新增分区。

num.recovery.threads.per.data.dir
; B ? `/ t) o2 Q

对于如下 3 种情况，Kafka 会使用可配置的线程池来处理日志片段

服务器正常启动，用于打开每个分区的日志片段；

服务器崩溃后启动，用于检查和截断每个分区的日志片段；

服务器正常关闭，用于关闭日志片段

默认情况下，每个日志目录只使用一个线程。因为这些线程只是在服务器启动和关闭时会用到，所以完全可以设置大量的线程来达到井行操作的目的。特别是对于包含大量分区的服务器来说，一旦发生崩愤，在进行恢复时使用井行操作可能会省下数小时的时间。设置此参数时需要注意，所配置的数字对应的是 log.dirs 指定的单个日志目录。也就是说，如果 num.recovery.threads.per.data.dir 被设为 8，并且 log.dir 指定了 3 个路径，那么总共需要 24 个线程。

auto.create.topics.enable3 o. V9 K# ^# E, m2 S1 p" y# m: b1 u& }

默认情况下，Kafka 会在如下 3 种情况下创建主题

当一个生产者开始往主题写入消息时

当一个消费者开始从主题读取消息时

当任意一个客户向主题发送元数据请求时

delete.topic.enable
0 y0 m$ a, e4 \! V

如果你想要删除一个主题，你可以使用主题管理工具。默认情况下，是不允许删除主题的，delete.topic.enable 的默认值是 false 因此你不能随意删除主题。这是对生产环境的合理性保护，但是在开发环境和测试环境，是可以允许你删除主题的，所以，如果你想要删除主题，需要把 delete.topic.enable 设为 true。

主题默认配置

Kafka 为新创建的主题提供了很多默认配置参数，下面就来一起认识一下这些参数

num.partitions: I& `6 M# m" e2 v" F. c8 R

num.partitions 参数指定了新创建的主题需要包含多少个分区。如果启用了主题自动创建功能（该功能是默认启用的），主题分区的个数就是该参数指定的值。该参数的默认值是 1。要注意，我们可以增加主题分区的个数，但不能减少分区的个数。

default.replication.factor; D8 Z" K% S8 g

这个参数比较简单，它表示 kafka保存消息的副本数，如果一个副本失效了，另一个还可以继续提供服务default.replication.factor 的默认值为1，这个参数在你启用了主题自动创建功能后有效。

log.retention.ms
) @0 o5 v3 ]1 F! m. x

Kafka 通常根据时间来决定数据可以保留多久。默认使用 log.retention.hours 参数来配置时间，默认是 168 个小时，也就是一周。除此之外，还有两个参数 log.retention.minutes 和 log.retentiion.ms 。这三个参数作用是一样的，都是决定消息多久以后被删除，推荐使用 log.retention.ms。

log.retention.bytes
, G s; e Q0 o( Q. C2 M3 O

另一种保留消息的方式是判断消息是否过期。它的值通过参数 log.retention.bytes 来指定，作用在每一个分区上。也就是说，如果有一个包含 8 个分区的主题，并且 log.retention.bytes 被设置为 1GB，那么这个主题最多可以保留 8GB 数据。所以，当主题的分区个数增加时，整个主题可以保留的数据也随之增加。

log.segment.bytes n! K& x3 z$ n _+ ?% |

上述的日志都是作用在日志片段上，而不是作用在单个消息上。当消息到达 broker 时，它们被追加到分区的当前日志片段上，当日志片段大小到达 log.segment.bytes 指定上限（默认为 1GB）时，当前日志片段就会被关闭，一个新的日志片段被打开。如果一个日志片段被关闭，就开始等待过期。这个参数的值越小，就越会频繁的关闭和分配新文件，从而降低磁盘写入的整体效率。

log.segment.ms
; c% v) [8 G9 p4 f( C* r

上面提到日志片段经关闭后需等待过期，那么 log.segment.ms 这个参数就是指定日志多长时间被关闭的参数和，log.segment.ms 和 log.retention.bytes 也不存在互斥问题。日志片段会在大小或时间到达上限时被关闭，就看哪个条件先得到满足。

message.max.bytes( N$ U3 I4 ]. ~4 _

broker 通过设置 message.max.bytes 参数来限制单个消息的大小，默认是 1000 000，也就是 1MB，如果生产者尝试发送的消息超过这个大小，不仅消息不会被接收，还会收到 broker 返回的错误消息。跟其他与字节相关的配置参数一样，该参数指的是压缩后的消息大小，也就是说，只要压缩后的消息小于 mesage.max.bytes，那么消息的实际大小可以大于这个值

这个值对性能有显著的影响。值越大，那么负责处理网络连接和请求的线程就需要花越多的时间来处理这些请求。它还会增加磁盘写入块的大小，从而影响 IO 吞吐量。

admin · 发表于 2022-12-30 09:09:23

1、软件下载1.1 kakfa 下载

地址：http://kafka.apache.org/downloads

1.2 zookeeper 下载

（1）因为 kafka 要依赖于 zookeeper 做调度，kafka 中实际自带的有 kafka，但是一般建议使用独立的 zookeeper，方便后续升级及公用。

（2）下载地址：

http://zookeeper.apache.org/

1.3 下载说明

文件都不大，zk 是 9m 多，kafka 是 50 多兆

2、 kafka 单机部署及集群部署

**说明：**北游在本地弄了三台虚拟机，ip 分别为：

192.168.85.158192.168.85.168192.168.85.178

复制代码

2.1 单机部署

（1）上传 jar 包，就不再新建用户了，直接在 root 账户下执行，将 kafka 和 zookeeper 的 tar 包上传到/root/tools 目录下。

（2）解压

[root@ruanjianlaowang158 tools]# tar -zxvf kafka_2.12-2.4.1.tgz [root@ruanjianlaowang158 tools]# tar -zxvf apache-zookeeper-3.5.7-bin.tar.gz

复制代码

（3）配置 zookeeper 及启动

[root@ruanjianlaowang158 apache-zookeeper-3.5.7-bin]# cd /root/tools/apache-zookeeper-3.5.7-bin#北游，首先创建个空文件夹，在接下来的配置文件中配置[root@ruanjianlaowang158 apache-zookeeper-3.5.7-bin]# mkdir data[root@ruanjianlaowang158 conf]# cd /root/tools/apache-zookeeper-3.5.7-bin/conf[root@ruanjianlaowang158 conf]# cp zoo_sample.cfg zoo.cfg [root@ruanjianlaowang158 conf]# vi zoo.cfg #单机只改一个值，保存退出。#dataDir=/tmp/zookeeperdataDir=/root/tools/apache-zookeeper-3.5.7-bin/data
#启动zookeeper[root@ruanjianlaowang158 bin]# cd /root/tools/apache-zookeeper-3.5.7-bin/bin[root@ruanjianlaowang158 bin]# ./zkServer.sh start

复制代码

（4）配置 kafka 及启动

[root@ruanjianlaowang158 kafka_2.12-2.4.1]# cd /root/tools/kafka_2.12-2.4.1
#北游，新建个空文件夹[root@ruanjianlaowang158 kafka_2.12-2.4.1]# mkdir data
#北游，更改配置文件[root@ruanjianlaowang158 config]# cd /root/tools/kafka_2.12-2.4.1/config[root@ruanjianlaowang158 config]# vi server.properties
#需要改3个值#log.dirs=/tmp/kafka-logslog.dirs=/root/tools/kafka_2.12-2.4.1/data#listeners = PLAINTEXT://your.host.name:9092listeners=PLAINTEXT://192.168.85.158:9092#zookeeper.connect=localhost:2181zookeeper.connect=192.168.85.158:2181
#启动kafka[root@ruanjianlaowang158 bin]# cd /root/tools/kafka_2.12-2.4.1/bin[root@ruanjianlaowang158 bin]# ./zookeeper-server-start.sh ../config/server.properties &

复制代码

启动完毕，单机验证就不验证了，直接在集群中进行验证。

2.2 集群部署

（1）集群方式，首先把上面的单机模式，再在 192.168.85.168 和 192.168.85.178 服务器上先解压配置一遍。

（2）zookeeper 是还是更改 zoo.cfg

158,168,178 三台服务器一样：

[root@ruanjianlaowang158 conf]# cd /root/tools/apache-zookeeper-3.5.7-bin/conf
[root@ruanjianlaowang158 conf]# vi zoo.cfg#其他不变，最后面新加，三行，三台服务器配置一样，北游server.1=192.168.85.158:2888:3888server.2=192.168.85.168:2888:3888server.3=192.168.85.178:2888:3888
158服务器执行：echo "1" > /root/tools/apache-zookeeper-3.5.7-bin/data/myid168服务器执行：echo "2" > /root/tools/apache-zookeeper-3.5.7-bin/data/myid178服务器执行：echo "3" > /root/tools/apache-zookeeper-3.5.7-bin/data/myid

复制代码

（3）kafka 集群配置

[root@ruanjianlaowang158 config]# cd /root/tools/kafka_2.12-2.4.1/config[root@ruanjianlaowang158 config]# vi server.properties #broker.id 三台服务器不一样，158服务器设置为1,168服务器设置为2,178服务器设置为3
broker.id=1#三个服务器配置一样zookeeper.connect=192.168.85.158:2181,192.168.85.168:2181,192.168.85.178:2181

复制代码

Kafka 常用 Broker 配置说明：

192.168.85.168:2181,192.168.85.178:2181 | ZooKeeper 服务器地址，多台用","间隔 |

（4）集群启动

启动方式跟单机一样：

#启动zookeeper[root@ruanjianlaowang158 bin]# cd /root/tools/apache-zookeeper-3.5.7-bin/bin[root@ruanjianlaowang158 bin]# ./zkServer.sh start
#启动kafka[root@ruanjianlaowang158 bin]# cd /root/tools/kafka_2.12-2.4.1/bin[root@ruanjianlaowang158 bin]# ./zookeeper-server-start.sh ../config/server.properties &

复制代码

（5）注意点

集群启动的时候，单机那台服务器（158）可能会报：Kafka：Configured broker.id 2 doesn't match stored broker.id 0 in meta.properties.方案：在158服务器data中有个文件：meta.properties，文件中的broker.id也需要修改成与server.properties中的broker.id一样，所以造成了这个问题。

复制代码

（6）创建个 topic，后面 springboot 项目测试使用。

[root@ruanjianlaowang158 bin]# cd /root/tools/kafka_2.12-2.4.1/bin[root@ruanjianlaowang158 bin]# ./kafka-topics.sh --create --zookeeper 192.168.85.158:2181,192.168.85.168:2181,192.168.85.178:2181 --replication-factor 3 --partitions 5 --topic aaaa

复制代码

3、结合 springboot 项目3.1 pom 文件

<?xml version="1.0" encoding="UTF-8"?><project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd"> <modelVersion>4.0.0</modelVersion> <parent> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-parent</artifactId> <version>2.2.0.RELEASE</version> <relativePath/>  </parent> <groupId>com.itany</groupId> <artifactId>kafka</artifactId> <version>0.0.1-SNAPSHOT</version> <name>kafka</name> <description>Demo project for Spring Boot</description>
<properties> <java.version>1.8</java.version> </properties> <dependencies> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-web</artifactId> </dependency> <dependency> <groupId>org.springframework.kafka</groupId> <artifactId>spring-kafka</artifactId> </dependency> </dependencies> <build> <plugins> <plugin> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-maven-plugin</artifactId> </plugin> </plugins> </build>
</project>

复制代码

说明：

主要就两个 gav，一个是 spring-boot-starter-web，启动 web 服务使用；一个是 spring-kafka,这个是 springboot 集成额 kafka 核心包。

3.2 application.yml

spring:  kafka: # 北游，kafka集群服务器地址 bootstrap-servers: 192.168.85.158:9092,192.168.85.168:9092,192.168.85.178:9092 producer:    key-serializer: org.apache.kafka.common.serialization.StringSerializer    value-serializer: org.apache.kafka.common.serialization.StringSerializer consumer:    group-id: test    key-deserializer: org.apache.kafka.common.serialization.StringDeserializer    value-deserializer: org.apache.kafka.common.serialization.StringDeserializer

复制代码

3.3 producer（消息生产者）

@RestControllerpublic class KafkaProducer { @Autowired private KafkaTemplate template;    //北游，topic使用上测试创建的aaaa @RequestMapping("/sendMsg") public String sendMsg(String topic, String message){       template.send(topic,message);       return "success"; }}

复制代码

3.4 consumer（消费者）

@Componentpublic class KafkaConsumer { //北游，这里是监控aaaa这个topic，直接打印到idea中，北游 @KafkaListener(topics = {"aaaa"}) public void listen(ConsumerRecord record){       System.out.println(record.topic()+":"+record.value()); }}

复制代码

3.5 验证结果

（1）浏览器上输入

http://localhost:8080/sendMsg?topic=aaaa&message=bbbb

复制代码

（2）北游的 idea 控制台打印信息

完整版kafka学习笔记领取

1 H$ w- b. w6 q8 w! t5 V. W" ]" e

二、Kafka 副本机制1、什么是副本机制：

通常是指分布式系统在多台网络互联的机器上保存有相同的数据拷贝

2、副本机制的好处：2.1 提供数据冗余

系统部分组件失效，系统依然能够继续运转，因而增加了整体可用性以及数据持久性

2.2 提供高伸缩性

支持横向扩展，能够通过增加机器的方式来提升读性能，进而提高读操作吞吐量

2.3 改善数据局部性

允许将数据放入与用户地理位置相近的地方，从而降低系统延时。

3、kafka 的副本

（1）、本质就是一个只能追加写消息的日志文件

（2）、同一个分区下的所有副本保存有相同的消息序列

（3）、副本分散保存在不同的 Broker 上，从而能够对抗部分 Broker 宕机带来的数据不可用（Kafka 是有若干主题概，每个主题可进一步划分成若干个分区。每个分区配置有若干个副本）

如下：有 3 台 Broker 的 Kafka 集群上的副本分布情况

4、kafka 如何保证同一个分区下的所有副本保存有相同的消息序列：

基于领导者（Leader-based）的副本机制

工作原理如图：

（1）、Kafka 中分成两类副本：领导者副本（Leader Replica）和追随者副本（Follower Replica）。每个分区在创建时都要选举一个副本，称为领导者副本，其余的副本自动称为追随者副本。

（2）、Kafka 中，追随者副本是不对外提供服务的。追随者副本不处理客户端请求，它唯一的任务就是从领导者副本，所有的读写请求都必须发往领导者副本所在的 Broker，由该 Broker 负责处理。（因此目前 kafka 只能享受到副本机制带来的第 1 个好处，也就是提供数据冗余实现高可用性和高持久性）

（3）、领导者副本所在的 Broker 宕机时，Kafka 依托于 ZooKeeper 提供的监控功能能够实时感知到，并立即开启新一轮的领导者选举，从追随者副本中选一个作为新的领导者。老 Leader 副本重启回来后，只能作为追随者副本加入到集群中。

5、kafka 追随者副本到底在什么条件下才算与 Leader 同步

Kafka 引入了 In-sync Replicas，也就是所谓的 ISR 副本集合。ISR 中的副本都是与 Leader 同步的副本，相反，不在 ISR 中的追随者副本就被认为是与 Leader 不同步的

6、kafka In-sync Replicas（ISR）

（1）、ISR 不只是追随者副本集合，它必然包括 Leader 副本。甚至在某些情况下，ISR 只有 Leader 这一个副本

（2）、通过 Broker 端 replica.lag.time.max.ms 参数（Follower 副本能够落后 Leader 副本的最长时间间隔）值来控制哪个追随者副本与 Leader 同步？只要一个 Follower 副本落后 Leader 副本的时间不连续超过 10 秒，那么 Kafka 就认为该 Follower 副本与 Leader 是同步的，即使此时 Follower 副本中保存的消息明显少于 Leader 副本中的消息。

（3）、ISR 是一个动态调整的集合，而非静态不变的。

某个追随者副本从领导者副本中拉取数据的过程持续慢于 Leader 副本的消息写入速度，那么在 replica.lag.time.max.ms 时间后，此 Follower 副本就会被认为是与 Leader 副本不同步的，因此不能再放入 ISR 中。此时，Kafka 会自动收缩 ISR 集合，将该副本“踢出”ISR。

倘若该副本后面慢慢地追上了 Leader 的进度，那么它是能够重新被加回 ISR 的。

（4）、ISR 集合为空则 leader 副本也挂了，这个分区就不可用了，producer 也无法向这个分区发送任何消息了。（反之 leader 副本挂了可以从 ISR 集合中选举 leader 副本）

7、kafka leader 副本所在 broker 挂了，leader 副本如何选举

（1）、ISR 不为空，从 ISR 中选举

（2）、ISR 为空，Kafka 也可以从不在 ISR 中的存活副本中选举，这个过程称为 Unclean 领导者选举，通过 Broker 端参数unclean.leader.election.enable控制是否允许 Unclean 领导者选举。

开启 Unclean 领导者选举可能会造成数据丢失，但好处是，它使得分区 Leader 副本一直存在，不至于停止对外提供服务，因此提升了高可用性。反之，禁止 Unclean 领导者选举的好处在于维护了数据的一致性，避免了消息丢失，但牺牲了高可用性。

一个分布式系统通常只能同时满足一致性（Consistency）、可用性（Availability）、分区容错性（Partition tolerance）中的两个。显然，在这个问题上，Kafka 赋予你选择 C 或 A 的权利。

强烈建议不要开启 unclean leader election，毕竟我们还可以通过其他的方式来提升高可用性。如果为了这点儿高可用性的改善，牺牲了数据一致性，那就非常不值当了。

ps1：leader 副本的选举也可以理解为分区 leader 的选举

ps2：broker 的 leader 选举与分区 leader 的选举不同，

Kafka 的 Leader 选举是通过在 zookeeper 上创建/controller 临时节点来实现 leader 选举，并在该节点中写入当前 broker 的信息

{“version”:1,”brokerid”:1,”timestamp”:”1512018424988”}

复制代码

利用 Zookeeper 的强一致性特性，一个节点只能被一个客户端创建成功，创建成功的 broker 即为 leader，即先到先得原则，leader 也就是集群中的 controller，负责集群中所有大小事务。

当 leader 和 zookeeper 失去连接时，临时节点会删除，而其他 broker 会监听该节点的变化，当节点删除时，其他 broker 会收到事件通知，重新发起 leader 选举

再给你们留个小问题：如果允许 Follower 副本对外提供读服务，你觉得应该如何避免或缓解因 Follower 副本与 Leader 副本不同步而导致的数据不一致的情形？

三、实时日志统计流程1、项目流程

在整合这套方案的时候，项目组也是经过一番讨论，在讨论中，观点很多，有人认为直接使用 Storm 进行实时处理，去掉 Kafka 环节；也有认为直接使用 Kafka 的 API 去消费，去掉 Storm 的消费环节等等，但是最终组内还是一致决定使用这套方案，原因有如下几点：

业务模块化
功能组件化
9 H) Y/ h* z3 E$ Z

我们认为，Kafka 在整个环节中充当的职责应该单一，这项目的整个环节她就是一个中间件，下面用一个图来说明这个原因，如下图所示：

整个项目流程如上图所示，这样划分使得各个业务模块化，功能更加的清晰明了。

Data Collection
9 f1 b4 r ]9 [$ [& |

负责从各个节点上实时收集用户上报的日志数据，我们选用的是 Apache 的 Flume NG 来实现。

Data Access
+ f$ s, y! g" a

由于收集的数据的速度和数据处理的速度不一定是一致的，因此，这里添加了一个中间件来做处理，所使用的是 Apache 的 Kafka，关于 Kafka 集群部署。另外，有一部分数据是流向 HDFS 分布式文件系统了的，方便于为离线统计业务提供数据源。

Stream Computing
& l6 H9 X9 ]: q' R, m+ s

在收集到数据后，我们需要对这些数据做实时处理，所选用的是 Apache 的 Storm。关于 Storm 的集群搭建部署博客后面补上，较为简单。

Data Output
$ m6 s8 z2 R9 T4 |

在使用 Storm 对数据做处理后，我们需要将处理后的结果做持久化，由于对响应速度要求较高，这里采用 Redis＋MySQL 来做持久化。整个项目的流程架构图，如下图所示：

2、Flume

Flume 是一个分布式的、高可用的海量日志收集、聚合和传输日志收集系统，支持在日志系统中定制各类数据发送方（如：Kafka，HDFS 等），便于收集数据。Flume 提供了丰富的日志源收集类型，有：Console、RPC、Text、Tail、Syslog、Exec 等数据源的收集，在我们的日志系统中目前我们所使用的是 spooldir 方式进行日志文件采集，配置内容信息如下所示：

producer.sources.s.type = spooldirproducer.sources.s.spoolDir = /home/hadoop/dir/logdfs

复制代码

当然，Flume 的数据发送方类型也是多种类型的，有：Console、Text、HDFS、RPC 等，这里我们系统所使用的是 Kafka 中间件来接收，配置内容如下所示：

producer.sinks.r.type = org.apache.flume.plugins.KafkaSinkproducer.sinks.r.metadata.broker.list=dn1:9092,dn2:9092,dn3:9092producer.sinks.r.partition.key=0producer.sinks.r.partitioner.class=org.apache.flume.plugins.SinglePartitionproducer.sinks.r.serializer.class=kafka.serializer.StringEncoderproducer.sinks.r.request.required.acks=0producer.sinks.r.max.message.size=1000000producer.sinks.r.producer.type=syncproducer.sinks.r.custom.encoding=UTF-8producer.sinks.r.custom.topic.name=test

复制代码

3、Kafka

Kafka 是一种提供高吞吐量的分布式发布订阅消息系统，她的特性如下所示：

通过磁盘数据结构提供消息的持久化，这种结构对于即使数据达到 TB+级别的消息，存储也能够保持长时间的稳定。
搞吞吐特性使得 Kafka 即使使用普通的机器硬件，也可以支持每秒数 10W 的消息。
能够通过 Kafka Cluster 和 Consumer Cluster 来 Partition 消息。

- r8 r* C0 F0 Z

Kafka 的目的是提供一个发布订阅解决方案，他可以处理 Consumer 网站中的所有流动数据，在网页浏览，搜索以及用户的一些行为，这些动作是较为关键的因素。这些数据通常是由于吞吐量的要求而通过处理日志和日志聚合来解决。对于 Hadoop 这样的日志数据和离线计算系统，这样的方案是一个解决实时处理较好的一种方案。

关于 Kafka 集群的搭建部署和使用，上面已经写了，不会的朋友翻上去再看一下，这里就不赘述了。

4、Storm

Twitter 将 Storm 开源了，这是一个分布式的、容错的实时计算系统，已被贡献到 Apache 基金会，下载地址如下所示：

http://storm.apache.org/downloads.html

复制代码

Storm 的主要特点如下：

简单的编程模型。类似于 MapReduce 降低了并行批处理复杂性，Storm 降低了进行实时处理的复杂性。
可以使用各种编程语言。你可以在 Storm 之上使用各种编程语言。默认支持 Clojure、Java、Ruby 和 Python。要增加对其他语言的支持，只需实现一个简单的 Storm 通信协议即可。
容错性。Storm 会管理工作进程和节点的故障。
水平扩展。计算是在多个线程、进程和服务器之间并行进行的。
可靠的消息处理。Storm 保证每个消息至少能得到一次完整处理。任务失败时，它会负责从消息源重试消息。
快速。系统的设计保证了消息能得到快速的处理，使用ØMQ 作为其底层消息队列。
本地模式。Storm 有一个本地模式，可以在处理过程中完全模拟 Storm 集群。这让你可以快速进行开发和单元测试。

7 E9 A' A4 O1 r3 d/ |

Storm 集群由一个主节点和多个工作节点组成。主节点运行了一个名为“Nimbus”的守护进程，用于分配代码、布置任务及故障检测。每个工作节点都运行了一个名为“Supervisor”的守护进程，用于监听工作，开始并终止工作进程。

Nimbus 和 Supervisor 都能快速失败，而且是无状态的，这样一来它们就变得十分健壮，两者的协调工作是由 Apache 的 ZooKeeper 来完成的。

Storm 的术语包括Stream、Spout、Bolt、Task、Worker、Stream Grouping和Topology。

Stream 是被处理的数据。
Spout 是数据源。
Bolt 处理数据。
Task 是运行于 Spout 或 Bolt 中的线程。
Worker 是运行这些线程的进程。
Stream Grouping 规定了 Bolt 接收什么东西作为输入数据。数据可以随机分配（术语为 Shuffle），或者根据字段值分配（术语为 Fields），或者广播（术语为 All），或者总是发给一个 Task（术语为 Global），也可以不关心该数据（术语为 None），或者由自定义逻辑来决定（术语为 Direct）。
Topology 是由 Stream Grouping 连接起来的 Spout 和 Bolt 节点网络。在 Storm Concepts 页面里对这些术语有更详细的描述。
: Z7 p- |0 Y1 m' |/ d" E

关于 Storm 集群的搭建部署，博客在下一篇中更新，到时候会将更新地址附在这里，这里就先不对 Storm 集群的搭建部署做过多的赘述了。

5、总结

Kafka 日志消息保存时间总结 Kafka 作为一个高吞吐的消息中间件和传统的消息中间件一个很大的不同点就在于它的日志实际上是以日志的方式默认保存在/kafka-logs 文件夹中的。虽然默认有 7 天清楚的机制，但是在数据量大，而磁盘容量不足的情况下，经常出现无法写入的情况。如何调整 Kafka 的一些默认参数就显得比较关键了。这里笔者整理了一些常见的配置参数供大家参考：

分段策略属性

日志刷新策略

Kafka 的日志实际上是开始是在缓存中的，然后根据策略定期一批一批写入到日志文件中去，以提高吞吐率。

日志保存清理策略

这里特别说明一下，日志的真正清楚时间。当删除的条件满足以后，日志将被“删除”，但是这里的删除其实只是将该日志进行了“delete”标注，文件只是无法被索引到了而已。

但是文件本身，仍然是存在的，只有当过了 log.segment.delete.delay.ms 这个时间以后，文件才会被真正的从文件系统中删除。

文章写到这里差不多了，比我预计要写得短一些，因为还有一些东西要写出来难免长篇大论，篇幅不允许，想更透彻的掌握 kafka 的同学可以领取我整理的完整版kafka学习笔记，最近要准备面试的同学可以看看我这份kafka高频面试题整理。

后面我会把另外两个中间件也分别写文章分析，可以给我点个关注第一时间接到通知

		自动登录	找回密码
密码			注册

Kafka 快速搭建

浏览过的版块