Linux 部署 Kafka 集群

一、概述

1. Kafka 定义

  • Kafka 是一个分布式的基于发布/订阅模式的消息队列(Message Queue),主要应用于大数据实时处理领域
  • Kafka 是一个开源的分布式事件流平台(Event Streaming Platform),被数千家公司用于高性能数据管道、流分析、数据集成和关键任务应用

使用案例:

  • 网站在前端埋点记录用户操作行为,发送到日志服务器
  • 可通过 Flume 时刻监控日志变化,发送到 Hadoop 等大数据平台进行数据分析
  • Hadoop 正常读写速度受硬盘影响,约 100M/s。平常 Flume 采集速度小于 100M/s,但在特定场景,其速度会大于 200M/s。为避免造成阻塞,在中间引入 Kafka 集群作为缓冲

2. 消息队列

  • 目前企业中比较常见的消息队列产品主要有 Kafka、ActiveMQ、RabbitMQ、RocketMQ 等
  • 在大数据场景主要采用 Kafka 作为消息队列。在 JavaEE 开发中主要采用 ActiveMQ、RabbitMQ、RocketMQ 等

(1) 传统消息队列的应用场景

  • 缓存/消峰控制和优化数据流经过系统的速度,解决生产消息和消费消息的处理速度不一致的情况(如双十一秒杀)
  • 解耦允许独立的扩展或修改队列两端的处理过程,只要确保它们遵守同样的接口约束(如厂家、超市和顾客)
  • 异步通信允许用户把一个消息放入队列,但并不立即处理它,而是在需要的时候再去处理(如登录后发送短信通知)

(2) 消息队列的两种模式

  • 点对点模式消费者主动拉取数据,消息收到后清除消息(一对一)
  • 发布/订阅模式可以有多个 topic 主题,消费者消费数据之后,不删除数据,且每个消费者相互独立,都可以消费到数据
    • 消息的发布者不会将消息直接发送给特定的订阅者,而是将发布的消息分为不同的类别,订阅者只接收感兴趣的消息

mq-mode

3. Kafka 基础架构

kafka-overview

  • Producer:消息生产者,向 Kafka Broker 发消息的客户端
  • Consumer:消息消费者,向 Kafka Broker 取消息的客户端
  • Consumer Group(CG):消费者组,由多个 consumer 组成
  • Broker:一台 Kafka 服务器就是一个 Broker
  • Topic:可以理解为一个队列
  • Partition:为了实现扩展性,一个 topic 可以分为多个 partition,每个 partition 是一个有序的队列,可以分布在不同的 Broker 上
  • Replica:副本。一个 topic 的每个 partition 都有若干个副本,一个 Leader 和若干个 Follower
  • Leader:主副本,生产者、消费者只针对 Leader 副本操作
  • Follower:从副本,实时从 Leader 同步数据,当 Leader 发生故障时,某个 Follower 会成为新的 Leader

二、 集群部署

1. 集群规划

hadoop100 hadoop101 hadoop102
ZK、Kafka ZK、Kafka ZK、Kafka

2. ZooKeeper 集群部署

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
## 准备资源
$ cd /opt/src
$ tar -zxvf apache-zookeeper-3.7.0-bin.tar.gz -C /opt/
$ cd /opt
$ mv apache-zookeeper-3.7.0-bin/ zookeeper
$ cd zookeeper

## 配置ZooKeeper
$ mv conf/zoo_sample.cfg conf/zoo.cfg
$ vim conf/zoo.cfg
dataDir=/opt/zookeeper/data
server.0=hadoop100:2888:3888
server.1=hadoop101:2888:3888
server.2=hadoop102:2888:3888
$ mkdir data
$ echo 0 > data/myid # 0, 1, 2

## 配置ZooKeeper环境变量
$ vim /etc/profile.d/my.sh
export ZOOKEEPER_HOME=/opt/zookeeper
export PATH=$PATH:$ZOOKEEPER_HOME/bin
$ source /etc/profile

## 集群分发
$ xsync /opt/zookeeper
$ xsync /etc/profile.d/my.sh
  • 集群脚本:/usr/local/bin/zk.sh
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
#!/bin/bash

case $1 in
"start"){
for i in hadoop100 hadoop101 hadoop102
do
echo ------------- zookeeper $i 启动 ------------
ssh $i "/opt/zookeeper/bin/zkServer.sh start"
done
}
;;
"stop"){
for i in hadoop100 hadoop101 hadoop102
do
echo ------------- zookeeper $i 停止 ------------
ssh $i "/opt/zookeeper/bin/zkServer.sh stop"
done
}
;;
"status"){
for i in hadoop100 hadoop101 hadoop102
do
echo ------------- zookeeper $i 状态 ------------
ssh $i "/opt/zookeeper/bin/zkServer.sh status"
done
}
;;
esac

3. Kafka 集群部署

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
## 准备资源
$ cd /opt/src
$ tar -zxvf kafka_2.12-3.0.0.tgz -C /opt/
$ cd /opt
$ mv kafka_2.12-3.0.0/ kafka
$ cd kafka

## 配置Kafka
$ vim config/server.properties
broker.id=0 # 0, 1, 2
log.dirs=/opt/kafka/data
zookeeper.connect=hadoop100:2181,hadoop101:2181,hadoop102:2181/kafka

## 配置Kafka环境变量
$ vim /etc/profile.d/my.sh
export KAFKA_HOME=/opt/kafka
export PATH=$PATH:$KAFKA_HOME/bin
$ source /etc/profile

## 集群分发
$ xsync /opt/kafka
$ xsync /etc/profile.d/my.sh
  • 集群脚本:/usr/local/bin/kf.sh
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
#!/bin/bash

case $1 in
"start"){
for i in hadoop100 hadoop101 hadoop102
do
echo ------------- kafka $i 启动 ------------
ssh $i "/opt/kafka/bin/kafka-server-start.sh -daemon /opt/kafka/config/server.properties"
done
}
;;
"stop"){
for i in hadoop100 hadoop101 hadoop102
do
echo ------------- kafka $i 停止 ------------
ssh $i "/opt/kafka/bin/kafka-server-stop.sh"
done
}
;;
esac

4. 启动

1
2
3
4
# 启动ZooKeeper集群
$ zk.sh start
# 启动Kafka集群
$ kf.sh start

先停止 Kafka,等待其所有节点进程停止后,再停止 ZooKeeper:Kafka 将一些信息存储到了 ZooKeeper,停止前需要先和 ZooKeeper 通信

三、Kafka 命令行操作

1. 主题命令行操作

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
# 查看操作主题命令参数
$ bin/kafka-topics.sh
# 查看当前服务器中的所有topic
$ bin/kafka-topics.sh --bootstrap-server hadoop100:9092,hadoop101:9092 --list
# 创建topic(1个分区,3个副本,名为test)
$ bin/kafka-topics.sh --bootstrap-server hadoop100:9092 --create --partitions 1 --replication-factor 3 --topic test
# 查看topic详情
$ bin/kafka-topics.sh --bootstrap-server hadoop100:9092 --describe --topic test
Topic: test TopicId: D7jWqskZSf-F-MC39chP5A PartitionCount: 1 ReplicationFactor: 3 Configs: segment.bytes=1073741824
Topic: test Partition: 0 Leader: 2 Replicas: 2,1,0 Isr: 2,1,0
# 修改分区数(只能增加,不能减少)
$ bin/kafka-topics.sh --bootstrap-server hadoop100:9092 --alter --topic test --partitions 3
# 修改副本数(不能通过命令行方式修改)
# 再次查看详情
$ bin/kafka-topics.sh --bootstrap-server hadoop100:9092 --describe --topic test
Topic: test TopicId: D7jWqskZSf-F-MC39chP5A PartitionCount: 3 ReplicationFactor: 3 Configs: segment.bytes=1073741824
Topic: test Partition: 0 Leader: 2 Replicas: 2,1,0 Isr: 2,1,0
Topic: test Partition: 1 Leader: 0 Replicas: 0,1,2 Isr: 0,1,2
Topic: test Partition: 2 Leader: 1 Replicas: 1,2,0 Isr: 1,2,0
# 删除topic
$ bin/kafka-topics.sh --bootstrap-server hadoop100:9092 --delete --topic test
  • 核心参数
参数 描述
--bootstrap-server <String: server to connect to> 连接的 Kafka Broker 的主机名称和端口号
--topic <String: topic> 操作的 topic 名称
--create 创建主题
--delete 删除主题
--alter 修改主题
--list 查看所有主题
--describe 查看主题详细描述
--partitions <Integer: # of partitions> 设置分区数
--replication-factor <Integer: replication factor> 设置分区副本
--config <String: name=value> 更新系统默认的配置

2. 生产者命令行操作

1
2
3
4
5
# 查看操作生产者命令参数
$ bin/kafka-console-producer.sh
# 发送消息
$ bin/kafka-console-producer.sh --bootstrap-server hadoop100:9092 --topic test
>hello
  • 核心参数
参数 描述
--bootstrap-server <String: server to connect to> 连接的 Kafka Broker 的主机名称和端口号
--topic <String: topic> 操作的 topic 名称

3. 消费者命令行操作

1
2
3
4
5
6
# 查看操作消费者命令参数
$ bin/kafka-console-consumer.sh
# 消费topic中的数据
$ bin/kafka-console-consumer.sh --bootstrap-server hadoop100:9092 --topic test
# 把主题中所有的数据都读取出来(包括历史数据)
$ bin/kafka-console-consumer.sh --bootstrap-server hadoop100:9092 --from-beginning --topic test
  • 核心参数
参数 描述
--bootstrap-server <String: server to connect to> 连接的 Kafka Broker 的主机名称和端口号
--topic <String: topic> 操作的 topic 名称
--from-beginning 从头开始消费
--group <String: consumer group id> 指定消费者组名称

四、其他

1. Kafka 配置文件

server.properties

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
# broker的全局唯一编号,不能重复,只能是数字
broker.id=0
# 处理网络请求的线程数量
num.network.threads=3
# 处理磁盘IO的线程数量
num.io.threads=8
# 发送套接字的缓冲区大小
socket.send.buffer.bytes=102400
# 接收套接字的缓冲区大小
socket.receive.buffer.bytes=102400
# 请求套接字的缓冲区大小
socket.request.max.bytes=104857600
# kafka运行日志/数据存放的路径,无需提前创建,可以配置多个磁盘路径,用“,”分割
log.dirs=/opt/kafka/data
# topic在当前broker上的分区个数
num.partitions=1
# 恢复和清理data下数据的线程数量
num.recovery.threads.per.data.dir=1
# 每个topic创建时的副本数
offsets.topic.replication.factor=1
# segment文件保留的最长时间,超时将被删除
log.retention.hours=168
# 每个segment文件的大小,默认最大1G
log.segment.bytes=1073741824
# 检查过期数据的时间,默认5分钟检查一次是否数据过期
log.retention.check.interval.ms=300000
# 配置连接ZooKeeper集群地址(在ZK根目录下创建“/kafka”,方便管理)
zookeeper.connect=hadoop100:2181,hadoop101:2181,hadoop102:2181/kafka