kafka

发表评论
6010 views

目录:

环境介绍:

http://kafka.apache.org/

Kafka是一种高吞吐量的分布式发布订阅消息系统，它可以处理消费者规模的网站中的所有动作流数据

软件版本:

  • Kafka 2.12

  • Zookeeper 3.4.13

Zookeeper+Kafka下载地址：

# wget http://mirrors.bit.edu.cn/apache/zookeeper/zookeeper-3.4.14/zookeeper-3.4.14.tar.gz

# wget http://mirrors.bit.edu.cn/apache/kafka/1.1.0/kafka_2.12-1.1.0.tgz

服务器:

  • 192.168.137.50 master
  • 192.168.137.60 node1
  • 192.168.137.70 node2
  • kafka: 192.168.137.50,192.168.137.60,192.168.137.70

Kafka集群部署(内置zookeeper)

部署 Kafka 可以分为以下几个步骤：

  • 下载 Kafka：从官网下载 Kafka 的二进制文件（tar.gz）。

  • 解压 Kafka：将下载的 Kafka 文件解压到任意目录下，例如 /opt。

$ tar -xzf kafka_2.13-2.8.0.tgz -C /opt

  • 配置 ZooKeeper：Kafka 依赖于 ZooKeeper 进行协调和管理，因此需要先启动一个 ZooKeeper 集群。我们通常使用单机模式运行 ZooKeeper，编辑该文件夹下conf/zookeeper.properties文件，并按照实际情况进行修改。例如，更改数据目录、客户端连接端口、创建myid等等。和单独部署zookeeper的步骤是一样的

  • 启动 Zookeeper：在 Kafka 解压目录下执行以下命令以启动 ZooKeeper。

$ bin/zookeeper-server-start.sh config/zookeeper.properties

在使用内置的 Kafka 时，要查看 ZooKeeper 的状态，可以通过以下步骤进行操作：

1、 首先，您需要登录到运行 Kafka 的服务器。

2、 打开 Kafka 安装目录，进入 bin 目录。

3、 在命令行中运行以下命令来启动 ZooKeeper Shell：

./zookeeper-shell.sh localhost:2181

如果您使用的是 Windows 系统，使用以下命令：

zookeeper-shell.bat localhost:2181

4、 运行该命令后，您将进入 ZooKeeper Shell。接下来，您可以使用各种 ZooKeeper 命令来获取状态信息。例如，您可以运行以下命令来列出 ZooKeeper 中的所有节点：

ls /

这将显示 ZooKeeper 根节点的子节点列表。

5、 您还可以使用其他命令来查看更多有关 ZooKeeper 状态的信息。例如，使用 stat 命令来获取有关特定节点的详细信息。例如：

stat /brokers/ids/0

这将显示有关 Kafka 的 Broker ID 为 0 的节点的详细信息。

请注意，上述命令假定您的 ZooKeeper 实例正在本地主机上的默认端口 2181 上运行。如果您的配置不同，请相应地更改命令中的 ZooKeeper 连接字符串。

  • 配置 Kafka：编辑 Kafka 的配置文件（conf/server.properties），并按照实际情况进行修改。例如，更改监听地址、通信端口、存储路径等等。

  • 启动 Kafka：在 Kafka 解压目录下执行以下命令以启动 Kafka。

$ bin/kafka-server-start.sh config/server.properties
  • 创建主题：通过 Kafka 的命令行工具创建一个主题。
$ bin/kafka-topics.sh --create --bootstrap-server localhost:9092 --replication-factor 1 --partitions 1 --topic my-topic
  • 生产者发送消息：通过 Kafka 的命令行工具往主题中发送一些消息。
$ bin/kafka-console-producer.sh --bootstrap-server localhost:9092 --topic my-topic
> Hello, World!
  • 消费者消费消息：通过 Kafka 的命令行工具从主题中读取消息。
$ bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic my-topic --from-beginning
Hello, World!

以上是在单机上部署 Kafka 的基本步骤。当然，在实际的生产环境中，我们需要考虑更多因素，包括高可用性、容错性、安全性等等。如果你想了解更多关于 Kafka 的详细信息，可以查看官方文档以获取更多帮助。

Kafka集群部署(外置zookeeper)

Zookeeper集群部署

Kafka集群是把状态保存在Zookeeper中的，首先要搭建Zookeeper集群。

1-1、Linux服务器一台、三台、五台、（2*n + 1），Zookeeper集群的工作是超过半数才能对外提供服务，3台中超过两台超过半数，允许1台挂掉 ，是否可以用偶数，其实没必要。

如果有四台那么挂掉一台还剩下三台服务器，如果在挂掉一个就不行了，这里记住是超过半数。

1-2、zookeeper是用java写的所以他的需要JAVA环境，java是运行在java虚拟机上的

1.下载软件包

cd /usr/local/elk
[root@master elk]# wget http://mirrors.bit.edu.cn/apache/zookeeper/zookeeper-3.4.14/zookeeper-3.4.14.tar.gz

2.安装zookeeper

[root@master elk]# tar xf zookeeper-3.4.14.tar.gz
[root@master elk]# ln -s zookeeper-3.4.14 zookeeper
[root@master elk]# cd zookeeper/conf/
[root@master conf]# cp zoo_sample.cfg zoo.cfg

[root@master conf]# cat zoo.cfg
tickTime=2000
initLimit=10
syncLimit=5
dataDir=/tmp/zookeeper
clientPort=2181
admin.serverPort=8080

server.1=192.168.137.50:2888:3888
server.2=192.168.137.60:2888:3888
server.3=192.168.137.70:2888: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 会监听这个端口，接受客户端的访问请求。修改他的端口改大点

#admin.serverPort：
管理后台端口

#server.1 这个1是服务器的标识也可以是其他的数字， 表示这个是第几号服务器，用来   标识服务器，这个标识要写到快照目录下面myid文件里
#192.168.137.50为集群里的IP地址，第一个端口是master和slave之间的通信端口， 默认是2888，第二个端口是leader选举的端口，集群刚启动的时候选举或者leader挂掉之后进行新的选举的端口默认是3888

3.创建dataDir目录

# 在master节点上
[root@master conf]# mkdir /tmp/zookeeper
[root@master conf]# touch /tmp/zookeeper/myid

创建myid编号，依次在每台Server上的dataDir目录执行，注意每台Server的myid要对应正确的编号：

[root@master conf]# echo 1 > /tmp/zookeeper/myid

直接将zookeeper-3.4.14目录复制到另外两个节点：

[root@master elk]# scp -r /usr/local/elk/zookeeper-3.4.14 192.168.137.60:/usr/local/elk
[root@master elk]# scp -r /usr/local/elk/zookeeper-3.4.14 192.168.137.70:/usr/local/elk

4.在两个slave节点创建目录和文件

#192.168.137.60（node01）节点:
[root@node01 elk]# ln -s zookeeper-3.4.14/ zookeeper
[root@node01 elk]# mkdir /tmp/zookeeper
[root@node01 elk]# touch /tmp/zookeeper/myid
[root@node01 elk]# echo 2 > /tmp/zookeeper/myid

#192.168.137.70（node02）节点:
[root@node01 elk]# ln -s zookeeper-3.4.14/ zookeeper
[root@node01 elk]# mkdir /tmp/zookeeper
[root@node01 elk]# touch /tmp/zookeeper/myid
[root@node01 elk]# echo 3 > /tmp/zookeeper/myid

5.配置环境变量并使环境变量立即生效

[root@master zookeeper]# vim /etc/profile
export ZOOKEEPER_HOME=/usr/local/elk/zookeeper
export PATH=$ZOOKEEPER_HOME/bin:$PATH
[root@master zookeeper]# source /etc/profile

同样，在其它Server也要做同样的操作

6.启动zookeeper并且进行监控

注意下：各台Server的防火墙要关闭，要不可能会报错。关闭防火墙方法：

[root@master elk]# systemctl stop firewalld

执行后查询是否关闭

[root@master elk]# systemctl status firewalld

[root@master elk]# zkServer.sh --help
/usr/bin/java
ZooKeeper JMX enabled by default
Using config: /usr/local/elk/zookeeper/bin/../conf/zoo.cfg
Usage: /usr/local/elk/zookeeper/bin/zkServer.sh [--config <conf-dir>] {start|start-foreground|stop|restart|status|print-cmd}

启动ZooKeeper： 先前台启动，不报错就后台启动

[root@master elk]# zkServer.sh start-foreground
/usr/bin/java
ZooKeeper JMX enabled by default
Using config: /usr/local/elk/zookeeper/bin/../conf/zoo.cfg
Starting zookeeper ... STARTED

启动ZooKeeper： 后台启动

[root@master elk]# zkServer.sh start
/usr/bin/java
ZooKeeper JMX enabled by default
Using config: /usr/local/elk/zookeeper/bin/../conf/zoo.cfg
Starting zookeeper ... STARTED

检查ZooKeeper的状态

root@master elk]# zkServer.sh status
ZooKeeper JMX enabled by default
Using config: /usr/local/elk/zookeeper/bin/../conf/zoo.cfg
Mode: follower

查看监听端口和进程

[root@master ~]# netstat -nutlp | grep 2181
tcp        0      0 0.0.0.0:2181            0.0.0.0:*               LISTEN      55188/java
[root@master ~]# ps aux | grep zookeeper

ZooKeeper常见配置

  • tickTime:CS通信心跳数；以毫秒为单位，可以使用默认配置。

  • initLimit:LF初始通信时限；

  • syncLimit:LF同步通信时限；数值不宜过高。

  • dataDir:数据文件目录；

  • dataLogDir:日志文件目录；

  • clientPort:客户端连接端口；

  • server.N:服务器名称与地址（服务编号，服务地址，LF通信端口，选举端口）

ZooKeeper高级配置

  • gloabalOutstandingLimit:最大请求堆积属，默认1000；

  • preAllocSize:预分配的Transaction log空间大小；

  • snapCount:每进行snapCount次事务日志输出后，触发一次快照；

  • maxClientCnxns:最大并发客户端数；

  • forceSync:是否提交事务的同时同步到磁盘；

  • leaderServes:是否禁止leader读功能；

  • traceFile:是否记录所有请求的log；不建议使用

至此zookeeper集群安装成功

kafka集群部署

kafka 基本概念:

  • kafka可以以一个或多个服务以集群的方式去浏览多个数据中心

  • kafka集群存储 记录流 是以主题(topic)类别的 方式 去存储的

  • 每条 记录 都 以 键,值,时间戳 组成

四个核心的api:

  • Productor api:(生产者) 允许 应用去发布 1个或多个的 主题(topic) 的记录流

  • Consumer api(消费者): 允许 应用去 订阅 1个或多个 主题(topic), 处理 生产的 记录流 给 消费者

  • Streams api(流): 允许 应用 去 扮演一个 处理流的 角色, 消费 一个 来自1个或者多个的主题的 输入流, 和 生产 1个或多个主题的 输出流, 有效的 改造 输入输出流

  • Connector api(连接器): 允许 构建 和重复 运行 消费者和 生产者 去 连接到 kafka 存在的 应用或者数据系统, 例如 关系数据库的连接器可能捕获对表的每个更改

  • 下载软件包并部署

cd /usr/local/elk
[root@master elk]# wget http://mirrors.bit.edu.cn/apache/kafka/2.3.1/kafka_2.12-2.3.1.tgz
[root@master elk]# tar xf kafka_2.12-2.3.1.tgz
[root@master elk]# ln -sv kafka_2.12-2.3.1 kafka
  1. 修改server.properties文件
[root@master elk]# cd kafka/config
[root@master config]# vim server.properties
broker.id=0
#kafka的监听地址
listeners=PLAINTEXT://192.168.137.50:9092
advertised.listeners=PLAINTEXT://192.168.137.50:9092
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
log.dirs=/tmp/kafka-logs
num.partitions=1
num.recovery.threads.per.data.dir=1
offsets.topic.replication.factor=1
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1
log.retention.hours=168
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
#配置kafka连接zookeeper的地址
zookeeper.connect=192.168.137.50:2181,192.168.137.60:2181,192.168.137.70:2181
zookeeper.connection.timeout.ms=6000
group.initial.rebalance.delay.ms=0
#可删除topic
delete.topic.enable=true
#自动创建topic
auto.create.topics.enable=true

参数配置：

  • broker.id(标示当前server在集群中的id，从0开始)

  • log.dirs(log的存储目录，需要提前创建)

  • advertised.listeners，这个是会发布给producer和consumer的地址，producer和consumer程序访问kafka

  • broker.id=0

当前机器在集群中的唯一标识，和zookeeper的myid性质一样,每台服务器的broker.id都不能相同

  • port=19092

当前kafka对外提供服务的端口默认是9092

  • host.name=192.168.30.204

这个参数默认是关闭的，在0.8.1有个bug，DNS解析问题，失败率的问题。

  • num.network.threads=3

这个是borker进行网络处理的线程数

  • num.io.threads=8

这个是borker进行I/O处理的线程数

  • log.dirs=/data/kafka/kafkalogs/

消息存放的目录，这个目录可以配置为“，”逗号分割的表达式，上面的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

消息保存的最大值5M

  • default.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.30.204:12181,192.168.30.205:12181,192.168.30.206:12181

设置zookeeper的连接端口，这个是zookeeper服务器列表，以逗号分隔

  1. 将 kafka_2.12-2.3.1 文件夹复制到另外两个节点下
[root@master elk]# scp -r /usr/local/elk/kafka_2.12-2.3.1 192.168.137.60:/usr/local/elk
[root@master elk]# scp -r /usr/local/elk/kafka_2.12-2.3.1 192.168.137.70:/usr/local/elk

并修改每个节点对应的server.properties文件的broker.id和listeners、advertised.listeners的名称

192.168.137.60（node01）主机配置文件如下：

[root@node01 elk]# ln -sv kafka_2.12-2.3.1/ kafka
broker.id=1
listeners=PLAINTEXT://192.168.137.60:9092
advertised.listeners=PLAINTEXT://192.168.137.60:9092

192.168.137.70（node02）主机配置文件如下：

[root@node01 elk]# ln -sv kafka_2.12-2.3.1/ kafka
broker.id=2
listeners=PLAINTEXT://192.168.137.70:9092
advertised.listeners=PLAINTEXT://192.168.137.70:9092
  1. 配置环境变量并使环境变量立即生效
[root@master elk]# vim /etc/profile
export ZOOKEEPER_HOME=/usr/local/elk/kafka
export PATH=$ZOOKEEPER_HOME/bin:$PATH
[root@master zookeeper]# source /etc/profile

同样，在其它Server也要做同样的操作

  1. 每个节点启动服务
[root@master elk]# nohup kafka-server-start.sh kafka/config/server.properties &

# 第一种方式（推荐）
kafka-server-start.sh -daemon ../config/server.properties
# 第二种方式
nohup kafka-server-start.sh config/server.properties &
  1. 停止
kafka-server-stop.sh

Kafka 安装包中 bin 目录下面的脚本是 Kafka 的命令行工具，它们实现了 Kafka 的各种功能，比如创建主题、发送消息、消费消息等。以下是 bin 目录中常用的几个脚本：

  • kafka-server-start.sh: 启动 Kafka 服务器。
  • kafka-server-stop.sh: 停止 Kafka 服务器。
  • kafka-topics.sh: 创建、列出、删除、描述 Kafka 的主题。
  • kafka-console-producer.sh: 以控制台的方式向 Kafka 主题中发送消息。
  • kafka-console-consumer.sh: 以控制台的方式消费 Kafka 主题中的消息。
  • zookeeper-server-start.sh: 启动 ZooKeeper 服务器。
  • zookeeper-server-stop.sh: 停止 ZooKeeper 服务器。

这些脚本都需要在命令行中执行，并且需要指定一些参数以完成相应的操作。例如，要启动 Kafka 服务器，可以在命令行中执行以下命令：

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

其中 config/server.properties 是 Kafka 的配置文件路径。其他的脚本也类似，都需要指定相应的参数来完成对应的操作。

总之，在使用 Kafka 之前，需要先熟悉这些命令行工具的使用方法，才能更好地管理和维护 Kafka 集群。

  1. 查看服务是否正常启动
[root@node01 ~]# ps aux | grep kafka
[root@node01 ~]# netstat -nutlp | grep 9092
tcp        0      0 0.0.0.0:46338           0.0.0.0:*               LISTEN      214516/java
tcp        0      0 192.168.137.60:9092     0.0.0.0:*               LISTEN      214516/java

[root@node01 ~]# jps        #QuorumPeerMain 是 zookeeper 进程
185856 QuorumPeerMain
218740 Kafka
219256 Jps
  1. systemctl管理zookeeper、kafka
# cat  /usr/lib/systemd/system/zookeeper.service 
[Unit]
Description=zookeeper server
After=network.target

[Service]
User=root
Group=root
Type=forking
#Environment=JAVA_HOME=/opt/jdk1.8.0_333
ExecStart=/data/kafka/bin/zookeeper-server-start.sh -daemon /data/kafka/config/zookeeper.properties
ExecStop=/data/kafka/bin/zookeeper-server-stop.sh
Restart=on-failure
RestartPreventExitStatus=1
RestartSec=10
StartLimitInterval=300
StartLimitBurst=2
TimeoutStartSec=30
TimeoutStopSec=30
PrivateTmp=true

[Install]
WantedBy=multi-user.target

# cat  /usr/lib/systemd/system/kafka.service 
[Unit]
Description=kafka server
#Requires=zookeeper.service
#After=network.target zookeeper.service
After=network.target

[Service]
User=root
Group=root
Type=simple
#Environment=JAVA_HOME=/opt/jdk1.8.0_333
ExecStart=/data/kafka/bin/kafka-server-start.sh  /data/kafka/config/server.properties
ExecStop=/data/kafka/bin/kafka-server-stop.sh
Restart=on-failure
RestartPreventExitStatus=1
RestartSec=10
StartLimitInterval=300
StartLimitBurst=2
TimeoutStartSec=30
TimeoutStopSec=30
PrivateTmp=true

[Install]
WantedBy=multi-user.target

Zookeeper+Kafka集群测试

创建topic：

--zookeeper是zookeeper服务器列表，--repication-factor是备份数量，--partitions是该topic的分区数量，--topic是创建的topic的名字

创建一个名为test的Topic，只有3个分区和3个备份：

[root@master ~]# kafka-topics.sh --create --zookeeper 192.168.137.50:2181,192.168.137.60:2181,192.168.137.70:2181 --replication-factor 3 --partitions 3 --topic test
Created topic test.

显示topic：

[root@master ~]# kafka-topics.sh --describe --zookeeper 192.168.137.50:2181, 192.168.137.60:2181, 192.168.137.70:2181 --topic test
Topic:test      PartitionCount:3        ReplicationFactor:3     Configs:
        Topic: test     Partition: 0    Leader: 0       Replicas: 0,2,1 Isr: 0,2,1
        Topic: test     Partition: 1    Leader: 1       Replicas: 1,0,2 Isr: 1,0,2
        Topic: test     Partition: 2    Leader: 2       Replicas: 2,1,0 Isr: 2,1,0

状态说明：

  • Topic: test有三个分区分别为0、1、2;

  • 分区0的leader是0（broker.id），分区0有三个副本，并且状态都为lsr（ln-sync，表示可以参加选举成为leader）

列出topic：

查看已创建的topic信息

[root@master ~]# kafka-topics.sh --list --zookeeper 192.168.137.50:2181, 192.168.137.60:2181, 192.168.137.70:2181
test

创建 producer(生产者):

# 在master节点上 测试生产消息
[root@master ~]# kafka-console-producer.sh --broker-list 192.168.137.50:9092 --topic test

image-20230620132843926

创建 consumer(消费者):

# 在node01节点上 测试消费
[root@master ~]# kafka-console-consumer.sh --bootstrap-server 192.168.137.50:9092,192.168.137.60:9092,192.168.137.70:9092 --topic test --from-beginning

#--from-beginning 表示从开始第一个消息开始接收

#在node02节点上 测试消费
[root@master ~]# kafka-console-consumer.sh --bootstrap-server 192.168.137.50:9092,192.168.137.60:9092,192.168.137.70:9092 --topic test --from-beginning

然后在 producer 里输入消息，consumer 中就会显示出同样的内容，表示消费成功!

删除 topic:

[root@master ~]# kafka-topics.sh --delete --zookeeper 192.168.137.50:2181, 192.168.137.60:2181, 192.168.137.70:2181 --topic test

启动和关闭服务:

#启动服务:
nohup kafka-server-start.sh config/server.properties &

#停止服务：
kafka-server-stop.sh

至此Zookeeper+Kafka集群配置成功

免责声明： 本文部分内容转自网络文章，转载此文章仅为个人收藏，分享知识，如有侵权，请联系博主进行删除。