部署环境:
Server1:10.0.0.31
Server2:10.0.0.32
Server3:10.0.0.33
一、下载二进制kafka包
apt install -y apenjdk-8-jdk
cd /usr/local/src
wget https://mirrors.bfsu.edu.cn/apache/kafka/2.7.1/kafka_2.13-2.7.1.tgz
tar xvf kafka_2.13-2.7.1.tgz
ln -sv /usr/local/src/kafka_2.13-2.7.1 /usr/local/kafka
二、修改配置文件
配置文件说明
broker.id=1 #每个 broker 在集群中的唯一标识,正整数。
listeners=PLAINTEXT://10.0.0.31:9092 #监听地址
num.network.threads=8 #网络线程
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
log.dirs=/usr/local/kafka/kafka-logs #kakfa 用于保存数据的目录,所有的消息 都会存储在该目录当中
num.partitions=1 #设置创新新的 topic 默认分区数量 ,不超过服务器数量
log.retention.hours=168 #设置 kafka 中消息保留时间,默认为 168 小时即 7 天
#zookeeper.connect 指定连接的 zk 的地址,zk 中存储了 broker 的元数据信息,格式 如下:
zookeeper.connect=10.0.0.31:2181,10.0.0.32:2181,10.0.0.33:2181
zookeeper.connection.timeout.ms=6000 #设置连接 zookeeper 的超时时间,默 认 6 秒钟
kafka 节点 1
mkdir /usr/local/kafka/data #用来存放日志的目录
vim /usr/local/kafka/config/server.properties
broker.id=1
listeners=PLAINTEXT://10.0.0.31:9092
log.dirs=/usr/local/kafka/data
zookeeper.connect=10.0.0.31:2181,10.0.0.32:2181,10.0.0.33:2181
kafka 节点 2
mkdir /usr/local/kafka/data #用来存放日志的目录
vim /usr/local/kafka/config/server.properties
broker.id=2
listeners=PLAINTEXT://10.0.0.32:9092
log.dirs=/usr/local/kafka/data
zookeeper.connect=10.0.0.31:2181,10.0.0.32:2181,10.0.0.33:2181
kafka 节点 3
mkdir /usr/local/kafka/data #用来存放日志的目录
vim /usr/local/kafka/config/server.properties
broker.id=3
listeners=PLAINTEXT://10.0.0.33:9092
log.dirs=/usr/local/kafka/data
zookeeper.connect=10.0.0.31:2181,10.0.0.32:2181,10.0.0.33:2181
三、各节点启动 kafka
以守护进程启动,shell退出也仍继续执行
/usr/local/kafka/bin/kafka-server-start.sh -daemon /usr/local/kafka/config/server.properties
查看日志,查看kafka启动情况
ss -ntl #9092端口是否处在监听状态
tail -f /usr/local/kafka/logs/server.log
kafka 1 日志
四、验证 zookeeper 中 kafka 元数据
1、broker 依赖于 Zookeeper,每个 broker 的 id 和 Topic、Partition 这些元数据信 息都会写入 Zookeeper 的 ZNode 节点中;
2、Consumer 依赖于 Zookeeper,Consumer 在消费消息时,每消费完一条消息, 会将产生的 offset 保存到 Zookeeper 中,下次消费在当前 offset 往后继续消费; ps:kafka0.9 之前 Consumer 的 offset 存储在 Zookeeper 中,kafka0,9 以后 offset 存储在本地。
3、Partition 依赖于 Zookeeper,Partition 完成 Replication 备份后,选举出一个 leader, 这个是依托于 Zookeeper 的选举机制实现的;
五、测试 kafka 读写数据
5.1 创建 topic
创建名为 lck ,partitions(分区)为 3,replication(每个分区的副本数/每个 分区的分区因子)为 3 的 topic(主题)
在任意 kafaka 服务器操作
root@zk1:~# /usr/local/kafka/bin/kafka-topics.sh --create --zookeeper 10.0.0.31:2181,10.0.0.32:2181,10.0.0.33:2181 --partitions 3 --replication-factor 3 --topic lck
Created topic lck.
5.2 验证 topic
root@zk1:~# /usr/local/kafka/bin/kafka-topics.sh --describe --zookeeper 10.0.0.31:2181,10.0.0.32:2181,10.0.0.33:2181 --topic lck
Topic: lck PartitionCount: 3 ReplicationFactor: 3 Configs:
Topic: lck Partition: 0 leader: 2 Replicas: 2,3,1 Isr: 2,3,1
Topic: lck Partition: 1 leader: 3 Replicas: 3,1,2 Isr: 3,1,2
Topic: lck Partition: 2 leader: 1 Replicas: 1,2,3 Isr: 1,2,3
5.3 获取所有topic
root@zk1:~# /usr/local/kafka/bin/kafka-topics.sh --list --zookeeper 10.0.0.31:2181,10.0.0.32:2181,10.0.0.33:2181
lck
5.4 测试发送消息
root@zk1:~# /usr/local/kafka/bin/kafka-console-producer.sh --broker-list 10.0.0.31:9092,10.0.0.32:9092,10.0.0.33:9092 --topic lck
>msg1
>msg2
>msg3
5.5 测试获取消息
可以到任意一台 kafka 服务器测试消息获取,只要有相应的消息获取客户端即可
root@zk1:~# /usr/local/kafka/bin/kafka-consconsumer.sh --topic lck --bootstrap-server 10.0.0.31:9092 --from-beginning
msg2
msg1
msg3
5.6 删除 topic
root@zk1:~# /usr/local/kafka/bin/kafka-topics.sh --delete --zookeeper 10.0.0.31:2181,10.0.0.32:2181,10.0.0.33:2181 --topic lck
Topic lck is marked for deletion.
Note: This will have no impact if delete.topic.enable is not set to true.