凡科网做网站花多少钱,正规的报价单格式,做篮球视频网站,泉州服装电商网站建设kafka在新版本中已经可以不使用zookeeper进行服务部署#xff0c;排除zookeeper的部署方案可以节省一些服务资源#xff0c;这里使用 kafka_2.13-3.6.1.tgz 版本进行服务部署。
测试部署分为三个服务器#xff1a;
服务器名称服务器IP地址test01192.168.56.101test02192.1…kafka在新版本中已经可以不使用zookeeper进行服务部署排除zookeeper的部署方案可以节省一些服务资源这里使用 kafka_2.13-3.6.1.tgz 版本进行服务部署。
测试部署分为三个服务器
服务器名称服务器IP地址test01192.168.56.101test02192.168.56.102test03192.168.56.103
将下载的安装包分别上传到三个服务器并解压安装包
[rootlocalhost ~]# tar -zvxf kafka_2.13-3.6.1.tgz
[rootlocalhost ~]# cd kafka_2.13-3.6.1
[rootlocalhost kafka_2.13-3.6.1]# pwd
/root/kafka_2.13-3.6.1
[rootlocalhost kafka_2.13-3.6.1]# ls
bin config libs LICENSE licenses NOTICE site-docs
[rootlocalhost kafka_2.13-3.6.1]# 修改配置文件这里需要在三个服务器上面分别进行修改需要调整修改几个地方如下
[rootlocalhost kafka_2.13-3.6.1]# vim ./config/kraft/server.properties# 设置角色对应的节点ID节点ID在集群内不能重复
node.id1
# kraft中集群节点是有角色划分的分为broker和controller测试环境就不去做区分所有节点都有两种角色
process.rolesbroker,controller
# 指定controller仲裁节点列表
controller.quorum.voters1192.168.56.101:9093,2192.168.56.102:9093,3192.168.56.103:9093
# 指定数据存储目录
log.dirs/root/kafka_2.13-3.6.1/datas
# 服务监听地址
listenersPLAINTEXT://192.168.56.101:9092,CONTROLLER://192.168.56.101:9093
# 对外提供服务的地址和端口号如果不设置将使用listeners配置的地址
advertised.listenersPLAINTEXT://192.168.56.101:9092完成上面的配置后就可以进行服务器的启动了首次启动服务器步骤如下
生成集群ID集群id生成使用下面的命令这个命令只需要在一台服务器上面执行即可记录生成的这个ID
[rootlocalhost kafka_2.13-3.6.1]# ./bin/kafka-storage.sh random-uuid
nJID7Q7dT62E_ehtbO3RaA分别在几台服务器上面使用集群ID格式化目录
[rootlocalhost kafka_2.13-3.6.1]# ./bin/kafka-storage.sh format -t nJID7Q7dT62E_ehtbO3RaA -c ./config/kraft/server.properties
Formatting /root/kafka_2.13-3.6.1/datas with metadata.version 3.6-IV2.上面命令执行完成后配置的数据目录将会生成两个文件其中 bootstrap.checkpoint 是一个二进制文件meta.properties 是元数据文件
[rootlocalhost kafka_2.13-3.6.1]# ls datas/
bootstrap.checkpoint meta.properties
[rootlocalhost kafka_2.13-3.6.1]#
[rootlocalhost kafka_2.13-3.6.1]# cat datas/meta.properties
#
#Sat Feb 03 17:26:43 CST 2024
cluster.idnJID7Q7dT62E_ehtbO3RaA
node.id1
version1数据目录格式化完成后就可以启动服务在启动服务时需要指定配置文件同时可以通过参数来决定是前台启动服务还是后台启动
# 后台启动
$ ./bin/kafka-server-start.sh -daemon ./config/kraft/server.properties
# 前台启动
$ ./bin/kafka-server-start.sh ./config/kraft/server.properties完成上面这些一个kafka集群就已经搭建成功了不使用zookeeper使得服务搭建非常简单目前这个版本还支持使用zookeeper方式的安装。
# 启动zookeeper服务
$ bin/zookeeper-server-start.sh config/zookeeper.properties# 启动kafka服务
$ bin/kafka-server-start.sh config/server.properties但还是建议使用kraft方式启动毕竟可以省去zookeeper服务的维护节省一部分资源。
# 以下所有命令都是以test-topic主题为示例test-group消费者组为示例# 1. 创建主题(test-topic 主题名replication-factor 副本数量副本是包含leader的如果某个topic有副本该值至少要配置为2)
$ ./bin/kafka-topics.sh --bootstrap-server 192.168.56.101:9092 --create --topic test-topic --partitions 3 --replication-factor 2# 2. 查看主题
$ ./bin/kafka-topics.sh --bootstrap-server 192.168.56.101:9092 --describe --topic test-topic# 3. 删除主题
$ ./bin/kafka-topics.sh --bootstrap-server 192.168.56.101:9092 --delete --topic test-topic# 4. 列出主题列表
$ ./bin/kafka-topics.sh --bootstrap-server 192.168.56.101:9092 --list# 5. 调整分区数量
$ ./bin/kafka-topics.sh --bootstrap-server 192.168.56.101:9092 -alter --partitions 4 --topic test-topic# 6. 查看消费者组信息
$ ./bin/kafka-consumer-groups.sh --bootstrap-server 192.168.56.101:9092 --list# 7. 查看某个消费者组消费情况
$ ./bin/kafka-consumer-groups.sh --bootstrap-server 192.168.56.101:9092 --group test-group --describe# 8. 在控制台向某个主题写入数据
$ ./bin/kafka-console-producer.sh --broker-list 192.168.56.101:9092,192.168.56.102:9092,192.168.56.103:9092 --topic test-topic# 9. 在控制台消费某个主题数据
$ ./bin/kafka-console-consumer.sh --bootstrap-server 192.168.56.101:9092,192.168.56.102:9092,192.168.56.103:9092 --topic test-topic# 10. 指定消费10条数据
$ ./bin/kafka-console-consumer.sh --bootstrap-server 192.168.56.101:9092 --topic test-topic --max-messages 10# 在控制台消费数据时还可以添加参数
# 最早数据开始消费--from beginning
# 删除offsets并重新开始消费--delete-consumer-offsets --from beginning
# 指定消费者组相关信息--consumer.config ./config/consumer.properties------
# 下面这些是不怎么常用的命令没有验证过作为记录供参考# 更改主题配置信息
./bin/kafka-configs.sh --bootstrap-server 192.168.56.101:9092 --entity-type topics --entity-name test-topic --alter --add-config max.message.bytes128000# 查看主题配置信息
./bin/kafka-configs.sh --bootstrap-server 192.168.56.101:9092 --entity-type topics --entity-name test-topic --describe# 删除配置
./bin/kafka-configs.sh --bootstrap-server 192.168.56.101:9092 --entity-type topics --entity-name test-topic --alter --delete-config max.message.bytes# 将test topic的消费组的0分区的偏移量设置为最新
./bin/kafka-consumer-groups.sh --bootstrap-server 192.168.56.101:9092,192.168.56.102:9092,192.168.56.103:9092 --group test-group --topic test-topic:0 --reset-offsets --to-earliest –execute# 将test topic的消费组的0和1分区的偏移量设置为最旧
./bin/kafka-consumer-groups.sh --bootstrap-server 192.168.56.101:9092,192.168.56.102:9092,192.168.56.103:9092 --group test-group --topic test-topic:0,1 --reset-offsets --to-latest –execute# 将test topic的消费组的所有分区的偏移量设置为1000
./bin/kafka-consumer-groups.sh --bootstrap-server 192.168.56.101:9092,192.168.56.102:9092,192.168.56.103:9092 --group test-group --topic test-topic --reset-offsets --to-offset 1 –execute# --reset-offsets后可以跟的其他用法--to-current把位移调整到分区当前位移
# --reset-offsets后可以跟的其他用法--shift-by N 把位移调整到当前位移 N处注意N可以是负数表示向前移动
# --reset-offsets后可以跟的其他用法--to-datetime datetime把位移调整到大于给定时间的最早位移处datetime格式是yyyy-MM-ddTHH:mm:ss.xxx比如2017-08-04T00:00:00.000
