kafka2集群部署

jiezi

5 年前

jre 安装

假设当前目录为 /home/zk. 配置 3 节点伪集群.

下载

wget http://mirror.bit.edu.cn/apache/zookeeper/stable/apache-zookeeper-3.5.5-bin.tar.gz

zxf apache-zookeeper-3.5.5-bin.tar.gz

mv apache-zookeeper-3.5.5-bin zookeeper

mkdir /home/zk/tmp/node-0/data -p
mkdir /home/zk/tmp/node-0/datalog -p
mkdir /home/zk/tmp/node-1/data -p
mkdir /home/zk/tmp/node-1/datalog -p
mkdir /home/zk/tmp/node-2/data -p
mkdir /home/zk/tmp/node-2/datalog -p

修改配置文件

cp zookeeper/conf/zoo_sample.cfg zookeeper/conf/zoo-node-0.cfg

# 修改节点 0 配置
vi zookeeper/conf/zoo-node-0.cfg
dataDir=/home/zk/tmp/node-0/data
dataLogDir=/home/zk/tmp/node-0/datalog

# server.index = ip:port- 原子广播:port- 选举
server.0=127.0.0.1:2887:3887
server.1=127.0.0.1:2888:3888
server.2=127.0.0.1:2889:3889

# 拷贝节点 0 配置
cp zookeeper/conf/zoo-node-0.cfg zookeeper/conf/zoo-node-1.cfg
cp zookeeper/conf/zoo-node-0.cfg zookeeper/conf/zoo-node-2.cfg

# 修改节点 1 配置
vi zookeeper/conf/zoo-node-1.cfg
dataDir=/home/zk/tmp/node-1/data
dataLogDir=/home/zk/tmp/node-1/datalog
clientPort=2182

# 修改节点 2 配置
vi zookeeper/conf/zoo-node-2.cfg
dataDir=/home/zk/tmp/node-2/data
dataLogDir=/home/zk/tmp/node-2/datalog
clientPort=2183

# 指定节点编号
echo 0 >> tmp/node-0/data/myid
echo 1 >> tmp/node-1/data/myid
echo 2 >> tmp/node-2/data/myid

启动

cd zookeeper
bin/zkServer.sh start conf/zoo-node-0.cfg
bin/zkServer.sh start conf/zoo-node-1.cfg
bin/zkServer.sh start conf/zoo-node-2.cfg

# 查看状态
bin/zkServer.sh status conf/zoo-node-0.cfg
bin/zkServer.sh status conf/zoo-node-1.cfg
bin/zkServer.sh status conf/zoo-node-2.cfg

假设当前目录为 /home/kfk

下载

wget http://mirror.bit.edu.cn/apache/kafka/2.3.0/kafka_2.12-2.3.0.tgz
tar zxf kafka_2.12-2.3.0.tgz
# 或者设置软链接
mv kafka_2.12-2.3.0 kafka

cd kafka

mkdir tmp/kafka-logs-0 -p
mkdir tmp/kafka-logs-1 -p
mkdir tmp/kafka-logs-2 -p

修改配置文件

cp config/server.properties config/server-0.properties

vi config/server-0.properties
broker.id=0
listeners=PLAINTEXT://:9092
log.dirs=/home/kfk/kafka/tmp/kafka-logs-0
zookeeper.connect=127.0.0.1:2181,127.0.0.1:2182,127.0.0.1:2183
# end vi

cp config/server-0.properties config/server-1.properties
cp config/server-0.properties config/server-2.properties

vi config/server-1.properties
broker.id=1
listeners=PLAINTEXT://:9093
log.dirs=/home/kfk/kafka/tmp/kafka-logs-1
# end vi

vi config/server-2.properties
broker.id=2
listeners=PLAINTEXT://:9094
log.dirs=/home/kfk/kafka/tmp/kafka-logs-2
# end vi

启动

nohup bin/kafka-server-start.sh config/server-0.properties 1>/dev/null 2>&1 &
nohup bin/kafka-server-start.sh config/server-1.properties 1>/dev/null 2>&1 &
nohup bin/kafka-server-start.sh config/server-2.properties 1>/dev/null 2>&1 &

broker 配置

https://kafka.apache.org/docu…

https://blog.csdn.net/memoord…

必要的配置

broker.id
log.dirs
zookeeper.connect

############################# Server Basics #############################
## 服务节点 id，整数 >=-1，集群内唯一
broker.id=0

######################## Socket Server Settings #########################
## socket 服务器监听地址
listeners=PLAINTEXT://:9092

############################# Log Basics ################################
## 存储日志文件的目录列表，逗号分隔
log.dirs=/home/yukl/2_opensys/kafka/tmp/kafka-logs-0

## 每个 topic 的默认分区数
num.partitions=1

######################## Internal Topic Settings  #######################
## offset topic 的复制因子
offsets.topic.replication.factor=1
## 事务 topic 的复制因子
transaction.state.log.replication.factor=1
## 事务 topic 的 min.insync.replicas 配置
transaction.state.log.min.isr=1

######################### Log Flush Policy ############################
## 消息数达到目标刷新到磁盘
#log.flush.interval.messages=10000

## 时间间隔达到目标刷新到磁盘
#log.flush.interval.ms=1000

######################### Log Retention Policy #########################
## 日志文件被删除前的保存时间（单位小时）log.retention.hours=168

## 优先使用 minutes 配置，否则使用 ms，如果这两项都没设置，则使用 hours
#log.retention.minutes
#log.retention.ms

## 日志达到删除大小的阈值。每个 topic 下每个分区保存数据的最大文件大小；注意，这是每个分区的上限，因此这个数值乘以分区的个数就是每个 topic 保存的数据总量。同时注意：如果 log.retention.hours 和 log.retention.bytes 都设置了，则超过了任何一个限制都会造成删除一个段文件。注意，这项设置可以由每个 topic 设置时进行覆盖。- 1 为不限制。#log.retention.bytes=1073741824

## topic 分区的日志存放在某个目录下诸多文件中，这些文件将 partition 的日志切分成一段一段的，这就是段文件（segment file）；一个 topic 的一个分区对应的所有 segment 文件称为 log。这个设置控制着一个 segment 文件的最大的大小，如果超过了此大小，就会生成一个新的 segment 文件。此配置可以被覆盖。int
log.segment.bytes=1073741824

## 检查日志段文件的间隔时间，以确定是否文件属性是否到达删除要求。log.retention.check.interval.ms=300000

############################# Zookeeper #############################
## zookeeper 主机地址
zookeeper.connect=127.0.0.1:2181,127.0.0.1:2182,127.0.0.1:2183

## 与 zookeeper 建立连接的超时时间(ms)
zookeeper.connection.timeout.ms=6000


###################### Group Coordinator Settings########################
# The following configuration specifies the time, in milliseconds, that the GroupCoordinator will delay the initial consumer rebalance.
# The rebalance will be further delayed by the value of group.initial.rebalance.delay.ms as new members join the group, up to a maximum of max.poll.interval.ms.
# The default value for this is 3 seconds.
# We override this to 0 here as it makes for a better out-of-the-box experience for development and testing.
# However, in production environments the default value of 3 seconds is more suitable as this will help to avoid unnecessary, and potentially expensive, rebalances during application startup.
group.initial.rebalance.delay.ms=0

## 消息中的时间戳取值为创建时间或者日志追加时间
## 取值 CreateTime 或者 LogAppendTime
## 默认 CreateTime
log.message.timestamp.type=CreateTime

producer 配置

https://kafka.apache.org/docu…

name	desc	type	default	valid values
acks	生产者的请求反馈方式：0- 不等待，仅仅将消息发送至缓冲区即返回，此模式下 retries 配置无效；1- 等待 leader 写成功即返回；all 或者 -1 表示等待所有的 follower 写成功才返回。	string	1	[all, -1, 0, 1]
bootstrap.servers	初始连接服务列表 (host:port,host:port)，是 kafka 集群的一个子集，用于发现所有的 kafak 集群	list	“”	non-null string
compression.type	消息压缩模式，支持 none, gzip, snappy, lz4, or zstd.	string	none
retries	设置一个比零大的值，客户端如果发送失败则会重新发送。注意，这个重试功能和客户端在接到错误之后重新发送没什么不同。如果 max.in.flight.requests.per.connection 没有设置为 1，有可能改变消息发送的顺序，因为如果 2 个批次发送到一个分区中，并第一个失败了并重试，但是第二个成功了，那么第二个批次将超过第一个	int	2147483647	[0,…,2147483647]
client.id	一个字符串，在请求时传递给服务器。用于追踪请求源	string	“”
connections.max.idle.ms	连接闲置时间	long	540000
delivery.timeout.ms	发送消息上报成功或失败的最大时间. 注意，如果此时间在 retries 耗尽之前达到，则判定请求失败	int	“”

consumer 配置

https://kafka.apache.org/docu…

name	desc	type	default	valid values
bootstrap.servers	同 producer
group.id	consumer group 唯一标识	string
auto.offset.reset	earliest: 当各分区下有已提交的 offset 时，从提交的 offset 开始消费；无提交的 offset 时，从头开始消费 <br/>latest : 当各分区下有已提交的 offset 时，从提交的 offset 开始消费；无提交的 offset 时，消费新产生的该分区下的数据 <br/>none :topic 各分区都存在已提交的 offset 时，从 offset 后开始消费；只要有一个分区不存在已提交的 offset，则抛出异常	string	latest	[earliest,<br/>latest,<br/>none]
connections.max.idle.ms	同上
client.id	同上
enable.auto.commit	如果为 true，消费者的 offset 后台会定期自动提交	boolean	true
auto.commit.interval.ms	消费者自动提交 offset 的周期（ms）	int	5000

kafka2集群部署

kakfa 集群部署

准备工作

zookeeper 单机集群搭建

kafka 单机集群搭建

常用配置说明