1.主题管理

kafka-topics.sh工具脚本用于对主题操作，如创建、删除、修改、分区数、副本数及主题级别的配置。

1.1创建名为kafka-test主题，有2个副本，3个分区

[hadoop@h201 kafka_2.12-0.10.2.1]$ bin/kafka-topics.sh --create --zookeeper h201:2181,h202:2181,h203:2181 --replication-factor 2 --partitions 3 --topic kafka-test

输出：

Created topic "kafka-test".

####

--zookeeper 参数必传，用于与zookeeper连接

--partitions 参数必传，用于设置分区数

--replication-factor 参数必传，用于设置副本数

####

1.2进入log目录

[hadoop@h201 kafka_2.12-0.10.2.1]$ cd kafkalogs/

[hadoop@h201 kafkalogs]$ ls

kafka-test-1kafka-test-0

##副本分布到不同的节点上###

h202上

[hadoop@h202 kafkalogs]$ ls

kafka-test-2kafka-test-1

[hadoop@h201 zookeeper-3.4.5-cdh5.5.2]$ bin/zkCli.sh

[zk: localhost:2181(CONNECTED) 1] ls /brokers/topics/kafka-test/partitions

结果：

[0, 1, 2]

[zk: localhost:2181(CONNECTED) 2] get /brokers/topics/kafka-test

1.3 创建主题时，可以通过config参数设置主题级别以覆盖默认配置

[hadoop@h201 kafka_2.12-0.10.2.1]$ bin/kafka-topics.sh --create --zookeeper h201:2181,h202:2181,h203:2181 --replication-factor 2 --partitions 3 --topic kafka-test1 --config max.message.bytes=404800

[hadoop@h201 zookeeper-3.4.5-cdh5.5.2]$ bin/zkCli.sh

[zk: localhost:2181(CONNECTED) 1] get /config/topics/kafka-test1

结果：

{"version":1,"config":{"max.message.bytes":"404800"}}

2.删除主题

若想彻底删除，启动时加载server.properties文件中配置delete.topic.enable=true,默认为false

如果没有设置为true，只是标记kafka-test1为删除状态，当设置为true时删除文件目录及元数据。

[hadoop@h201 kafka_2.12-0.10.2.1]$ bin/kafka-topics.sh --delete --zookeeper h201:2181,h202:2181,h203:2181 --topic kafka-test1

3.查看主题

3.1

[hadoop@h201 kafka_2.12-0.10.2.1]$ bin/kafka-topics.sh --list --zookeeper h201:2181,h202:2181,h203:2181

3.2查看详细描述信息

[hadoop@h201 kafka_2.12-0.10.2.1]$ bin/kafka-topics.sh --describe --zookeeper h201:2181,h202:2181,h203:2181

3.3 查看处于正在同步主题

可以看到正在处于“under replicated”状态分区，该状态可能为正在同步或同步异常。

[hadoop@h201 kafka_2.12-0.10.2.1]$ bin/kafka-topics.sh --describe --zookeeper h201:2181,h202:2181,h203:2181 --under-replicated-partitions

3.4 查看没有leader的分区

[hadoop@h201 kafka_2.12-0.10.2.1]$ bin/kafka-topics.sh --describe --zookeeper h201:2181,h202:2181,h203:2181 --unavailable-partitions

3.5 查看主题覆盖的配置

[hadoop@h201 kafka_2.12-0.10.2.1]$ bin/kafka-topics.sh --describe --zookeeper h201:2181,h202:2181,h203:2181 --topics-with-overrides

4.修改主题

4.1

[hadoop@h201 kafka_2.12-0.10.2.1]$ bin/kafka-topics.sh --alter --zookeeper h201:2181,h202:2181,h203:2181 --topic kafka-test --config segment.bytes=209715200

##段大小 修改为200m##

4.2 删除修改

[hadoop@h201 kafka_2.12-0.10.2.1]$ bin/kafka-topics.sh --alter --zookeeper h201:2181,h202:2181,h203:2181 --topic kafka-test --delete-config segment.bytes

4.3 增加分区

Kafka不支持减少分区操作，只能增加分区操作

主题kafka-test当前分区数为3个，调整为5个

[hadoop@h201 kafka_2.12-0.10.2.1]$ bin/kafka-topics.sh --alter --zookeeper h201:2181,h202:2181,h203:2181 --topic kafka-test --partitions 5

[zk: localhost:2181(CONNECTED) 1] ls /brokers/topics/kafka-test/partitions

结果：

[0, 1, 2, 3, 4]

5.生产者操作

使用kafka-console-producer.sh脚本

参数producer.config：用于加载一个生产者配置文件

参数producer-property：用于设置参数将会覆盖所加载的配置文件中的参数

参数property：用于设置消息消费者相关配置

[hadoop@h201 kafka_2.12-0.10.2.1]$ bin/kafka-console-producer.sh --broker-list h201:9092,h202:9092,h203:9092 --topic kafka-test

必传参数--broker-list，--topic

测试：（显示主题下各个分区的偏移量）

[hadoop@h201 kafka_2.12-0.10.2.1]$ bin/kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list h201:9092,h202:9092,h203:9092 --topic kafka-test --time -1

##kafka-run-class.sh为kafka自带测试工具###

6.消费者操作

[hadoop@h201 kafka_2.12-0.10.2.1]$ bin/kafka-console-consumer.sh --zookeeper h201:2181,h202:2181,h203:2181 --topic kafka-test --consumer-property group.id=consumer-g1  --from-beginning

消费者创建时会向zookeeper中注册元数据信息 /consumers/${group.id}

---------------------测试-------------------------

[hadoop@h201 kafka_2.12-0.10.2.1]$ bin/kafka-console-producer.sh --broker-list h201:9092,h202:9092,h203:9092 --topic kafka-test

添加测试消息内容

hello kafka

[hadoop@h201 kafka_2.12-0.10.2.1]$ bin/kafka-console-consumer.sh --zookeeper h201:2181,h202:2181,h203:2181 --topic kafka-test --consumer-property group.id=consumer-g1  --from-beginning

能够接收到 “hello kafka”消息内容

1. 主题管理

kafka-topics.sh工具脚本用于对主题操作，如创建、删除、修改、分区数、副本数及主题级别的配置。

1.1 创建名为kafka-test主题，有2个副本，3个分区

[hadoop@h201 kafka_2.12-0.10.2.1]$ bin/kafka-topics.sh --create --zookeeper h201:2181,h202:2181,h203:2181 --replication-factor 2 --partitions 3 --topic kafka-test

输出：

Created topic "kafka-test".

####

--zookeeper 参数必传，用于与zookeeper连接

--partitions 参数必传，用于设置分区数

--replication-factor 参数必传，用于设置副本数

####

1.2进入log目录

[hadoop@h201 kafka_2.12-0.10.2.1]$ cd kafkalogs/

[hadoop@h201 kafkalogs]$ ls

kafka-test-1kafka-test-0

##副本分布到不同的节点上###

h202上

[hadoop@h202 kafkalogs]$ ls

kafka-test-2kafka-test-1

[hadoop@h201 zookeeper-3.4.5-cdh5.5.2]$ bin/zkCli.sh

[zk: localhost:2181(CONNECTED) 1] ls /brokers/topics/kafka-test/partitions

结果：

[0, 1, 2]

[zk: localhost:2181(CONNECTED) 2] get /brokers/topics/kafka-test

1.3 创建主题时，可以通过config参数设置主题级别以覆盖默认配置

[hadoop@h201 kafka_2.12-0.10.2.1]$ bin/kafka-topics.sh --create --zookeeper h201:2181,h202:2181,h203:2181 --replication-factor 2 --partitions 3 --topic kafka-test1 --config max.message.bytes=404800

[hadoop@h201 zookeeper-3.4.5-cdh5.5.2]$ bin/zkCli.sh

[zk: localhost:2181(CONNECTED) 1] get /config/topics/kafka-test1

结果：

{"version":1,"config":{"max.message.bytes":"404800"}}

2. 删除主题

若想彻底删除，启动时加载server.properties文件中配置delete.topic.enable=true,默认为false

如果没有设置为true，只是标记kafka-test1为删除状态，当设置为true时删除文件目录及元数据。

[hadoop@h201 kafka_2.12-0.10.2.1]$ bin/kafka-topics.sh --delete --zookeeper h201:2181,h202:2181,h203:2181 --topic kafka-test1

3. 查看主题

3.1

[hadoop@h201 kafka_2.12-0.10.2.1]$ bin/kafka-topics.sh --list --zookeeper h201:2181,h202:2181,h203:2181

3.2查看详细描述信息

[hadoop@h201 kafka_2.12-0.10.2.1]$ bin/kafka-topics.sh --describe --zookeeper h201:2181,h202:2181,h203:2181

3.3 查看处于正在同步主题

可以看到正在处于“under replicated”状态分区，该状态可能为正在同步或同步异常。

[hadoop@h201 kafka_2.12-0.10.2.1]$ bin/kafka-topics.sh --describe --zookeeper h201:2181,h202:2181,h203:2181 --under-replicated-partitions

3.4 查看没有leader的分区

[hadoop@h201 kafka_2.12-0.10.2.1]$ bin/kafka-topics.sh --describe --zookeeper h201:2181,h202:2181,h203:2181 --unavailable-partitions

3.5 查看主题覆盖的配置

[hadoop@h201 kafka_2.12-0.10.2.1]$ bin/kafka-topics.sh --describe --zookeeper h201:2181,h202:2181,h203:2181 --topics-with-overrides

4. 修改主题

4.1

[hadoop@h201 kafka_2.12-0.10.2.1]$ bin/kafka-topics.sh --alter --zookeeper h201:2181,h202:2181,h203:2181 --topic kafka-test --config segment.bytes=209715200

##段大小 修改为200m##

4.2 删除修改

[hadoop@h201 kafka_2.12-0.10.2.1]$ bin/kafka-topics.sh --alter --zookeeper h201:2181,h202:2181,h203:2181 --topic kafka-test --delete-config segment.bytes

4.3 增加分区

Kafka不支持减少分区操作，只能增加分区操作

主题kafka-test当前分区数为3个，调整为5个

[hadoop@h201 kafka_2.12-0.10.2.1]$ bin/kafka-topics.sh --alter --zookeeper h201:2181,h202:2181,h203:2181 --topic kafka-test --partitions 5

[zk: localhost:2181(CONNECTED) 1] ls /brokers/topics/kafka-test/partitions

结果：

[0, 1, 2, 3, 4]

5. 生产者操作

使用kafka-console-producer.sh脚本

参数producer.config：用于加载一个生产者配置文件

参数producer-property：用于设置参数将会覆盖所加载的配置文件中的参数

参数property：用于设置消息消费者相关配置

[hadoop@h201 kafka_2.12-0.10.2.1]$ bin/kafka-console-producer.sh --broker-list h201:9092,h202:9092,h203:9092 --topic kafka-test

必传参数--broker-list，--topic

测试：（显示主题下各个分区的偏移量）

[hadoop@h201 kafka_2.12-0.10.2.1]$ bin/kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list h201:9092,h202:9092,h203:9092 --topic kafka-test --time -1

##kafka-run-class.sh为kafka自带测试工具###

6. 消费者操作

[hadoop@h201 kafka_2.12-0.10.2.1]$ bin/kafka-console-consumer.sh --zookeeper h201:2181,h202:2181,h203:2181 --topic kafka-test --consumer-property group.id=consumer-g1  --from-beginning

消费者创建时会向zookeeper中注册元数据信息 /consumers/${group.id}

---------------------测试-------------------------

[hadoop@h201 kafka_2.12-0.10.2.1]$ bin/kafka-console-producer.sh --broker-list h201:9092,h202:9092,h203:9092 --topic kafka-test

添加测试消息内容

hello kafka

[hadoop@h201 kafka_2.12-0.10.2.1]$ bin/kafka-console-consumer.sh --zookeeper h201:2181,h202:2181,h203:2181 --topic kafka-test --consumer-property group.id=consumer-g1  --from-beginning

能够接收到 “hello kafka”消息内容

最新文章

1. Win8/Win10无法打开这个应用 内置管理员账户
2. js 更改head的title
3. {C#}{GDI+}各种C#，GDI+的资料
4. ZOJ2532_Internship
5. 由多次使用Statement实例引起的Result set already closed异常的解决方案
6. Codeforces Round #188 (Div. 1) B. Ants 暴力
7. HDU5715 XOR 游戏 二分+字典树+dp
8. mmc生产运输问题
9. 几种任务调度的 Java 实现方法与比较Timer,ScheduledExecutor,Quartz,JCronTab
10. python之列表作为函数的参数
11. Python3 标准库概览
12. Android中的Message机制
13. cannot import name '_imaging' from 'PIL'
14. springBoot(6)---过滤器，监听器，拦截器
15. Windows 不能在本地计算机启动 OracleDBConsoleorcl的问题解决方法
16. Android、iOS、和Web如何做灰度发布？
17. ABAQUS粘弹性边界及地震荷载施加的简单实现(Matlab生成input文件)
18. 2017/05/08 java 基础 随笔
19. 版本控制-GitHub
20. linux下怎么删除名称带空格的文件

热门文章

1. CentOS 7镜像下载
2. Python3基础 yield send 变量名= yield i
3. Spring cloud微服务安全实战-7-9自定义日志采集的格式和内容
4. [译]使用to_dict将pandas.DataFrame转换为Python中的字典列表
5. tensorflow2.0手写数字识别
6. Oracle表存在则删除后再重建
7. 玩转CONSUL(3)–大规模部署的性能开销定量分析
8. Apollo环境配置
9. linux系统状态脚本
10. 装饰器login_required