# Kafka_深入探秘者（4）：kafka 主题 topic

Kafka_深入探秘者（4）：kafka 主题 topic

一、kafka 主题管理

1、kafka 创建主题 topic 命令

1）命令：

2. # 切换到 kafka 安装目录
3. cd /usr/local/kafka/kafka_2.12-2.8.0/
5. # 创建一个名为 heima 的 主题
6. bin/kafka-topics.sh --zookeeper localhost:2181 --create --topic heima --partitions 2 --replication-factor 1
8. # 或者把 localhost 换成 填写你的 虚拟机 IP 地址（如：172.18.30.110）：
9. bin/kafka-topics.sh --zookeeper 172.18.30.110:2181 --create --topic heima --partitions 2 --replication-factor 1

复制代码

2）命令 参数说明：
–zookeeper ：指定了 kafka 所毗连的 zookeeper 服务地点。zookeeper必传参数，多个zookeeper用’，"分开。
–topic : 指定了所要创建主题的名称
–partitions : 指定了分区个数，每个线程处理一个分区数据。
–replication-factor : 指定了副本因子，用于设置主题副本数，每个副天职布在不通节点，不能凌驾总结点数。如你只有一个节点，但是创建时指定副本数为2，就会报错
–create : 创建主题的动作指令。
2、查看 topic 元数据信细的方法：

topic 元数据信息保存在 Zookeeper 节点中，

2. # 切换到 zookeeper 安装目录
3. cd /usr/local/zookeeper/apache-zookeeper-3.6.3-bin
5. # 连接 zookeeper 查看元数据
6. bin/zkCli.sh -server localhost:2181
8. # 查看 节点 信息：
9. get /brokers/topics/heima

复制代码

3、展示出当前全部主题

2. # 切换目录
3. cd /usr/local/kafka/kafka_2.12-2.8.0/
5. # 展示出当前所有主题
6. bin/kafka-topics.sh --list --zookeeper localhost:2181
8. # 或者把 localhost 换成 填写你的 虚拟机 IP 地址（如：172.18.30.110）：
9. bin/kafka-topics.sh --list --zookeeper 172.18.30.110:2181

复制代码

4、 查看主题详情：

2. # 切换目录
3. cd /usr/local/kafka/kafka_2.12-2.8.0/
5. # 查看主题详情
6. bin/kafka-topics.sh --zookeeper localhost:2181 --describe --topic heima
8. # 或者把 localhost 换成 填写你的 虚拟机 IP 地址（如：172.18.30.110）：
9. bin/kafka-topics.sh --zookeeper 172.18.30.110:2181 --describe --topic heima

复制代码

5、 修改 主题 topic 设置（增长设置）

2. # 切换到 kafka 安装目录
3. cd /usr/local/kafka/kafka_2.12-2.8.0/
5. # 修改 主题 topic 配置（增加配置）
6. bin/kafka-topics.sh  --alter --zookeeper localhost:2181 --topic heima --config flush.messages=1
8. # 或者把 localhost 换成 填写你的 虚拟机 IP 地址（如：172.18.30.110）：
9. bin/kafka-topics.sh  --alter --zookeeper 172.18.30.110:2181 --topic heima --config flush.messages=1

复制代码

6、 删除 主题 topic 设置 flush.messages=1

2. # 切换到 kafka 安装目录
3. cd /usr/local/kafka/kafka_2.12-2.8.0/
5. # 删除 主题 topic 配置 flush.messages=1
6. bin/kafka-topics.sh  --alter --zookeeper localhost:2181 --topic heima --delete-config flush.messages
8. # 或者把 localhost 换成 填写你的 虚拟机 IP 地址（如：172.18.30.110）：
9. bin/kafka-topics.sh  --alter --zookeeper 172.18.30.110:2181 --topic heima --delete-config flush.messages

复制代码

7、 删除 topic 主题

若 delete.topic.enable=true 直接彻底删除该 Topic 主题。
若 delete.topic.enable=false
假如当前 Topic 没有使用过，即没有传输过信息:可以彻底删除。
假如当前 Topic 有使用过，即有过传输过信息:并没有真正删除 Topic，只是把这个 Topic 标志为删除(marked fordeletion)，重启 Kafka Server 后删除。

2. # 切换到 kafka 安装目录
3. cd /usr/local/kafka/kafka_2.12-2.8.0/
5. # 删除 topic 主题名为 heima 的主题
6. bin/kafka-topics.sh  --delete --zookeeper localhost:2181 --topic heima
8. # 或者把 localhost 换成 填写你的 虚拟机 IP 地址（如：172.18.30.110）：
9. bin/kafka-topics.sh  --delete --zookeeper 172.18.30.110:2181 --topic heima

复制代码

二、kafka 分区

1、kafka 增长分区

2. # 切换到 kafka 安装目录
3. cd /usr/local/kafka/kafka_2.12-2.8.0/
5. # 增加 topic 主题名为 heima 的主题的 分区为 3 个
6. bin/kafka-topics.sh  --alter --zookeeper localhost:2181 --topic heima --partitions 3
8. # 或者把 localhost 换成 填写你的 虚拟机 IP 地址（如：172.18.30.110）：
9. bin/kafka-topics.sh  --alter --zookeeper 172.18.30.110:2181 --topic heima --partitions 3
11. # 查看 heima 主题详情：
12. bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic heima
14. # 或者把 localhost 换成 填写你的 虚拟机 IP 地址（如：172.18.30.110）：
15. bin/kafka-topics.sh --describe --zookeeper 172.18.30.110:2181 --topic heima

复制代码

2、其他主题参数设置

见官方文档: http://kafka.apache.org/documentation/#topicconfigs

2. bin/kafka-topics.sh --bootstrap-server localhost:9092 --create --topic my-topic --partitions 1 \
3.   --replication-factor 1 --config max.message.bytes=64000 --config flush.messages=1
4.   
5.   
6. bin/kafka-configs.sh --bootstrap-server localhost:9092 --entity-type topics --entity-name my-topic
7.   --alter --add-config max.message.bytes=128000
8.   
9.   
10. bin/kafka-configs.sh --bootstrap-server localhost:9092 --entity-type topics --entity-name my-topic --describe  
13. bin/kafka-configs.sh --bootstrap-server localhost:9092  --entity-type topics --entity-name my-topic
14.   --alter --delete-config max.message.bytes

复制代码

三、kafka 主题 topic 总结

1、KafkaAdminClient 应用

我们都习惯使用 Kafka 中 bin 目次下的脚本工具来管理査看 Kafka，但是有些时候需要将某些管理查看的功能集成到系统 (比如Kafka Manager)中，那么就需要调用一些 API 来直接利用 Kafka 了。
2、在 kafka_learn 工程中，创建 KafkaAdminConfigOperation.java 类, KafkaAdminClient 应用

2. /**
3. *  D:\java-test\idea2019\kafka_learn\src\main\java\djh\it\kafka\learn\chapter3\KafkaAdminConfigOperation.java
4. *
5. *  2024-6-22 创建 KafkaAdminConfigOperation.java 类, KafkaAdminClient 应用
6. */
7. package djh.it.kafka.learn.chapter3;
9. import kafka.controller.NewPartition;
10. import org.apache.kafka.clients.admin.*;
11. import org.apache.kafka.common.config.ConfigResource;
13. import java.util.Collections;
14. import java.util.HashMap;
15. import java.util.Map;
16. import java.util.Properties;
17. import java.util.concurrent.ExecutionException;
19. public class KafkaAdminConfigOperation {
21.     public static void main( String[] args ) throws ExecutionException, InterruptedException {
22.         describeTopicConfig();   //获取主题详细信息
23. //        alterTopicConfig();    //添加主题
24. //        addTopicPartitions();  //添加分区
25.     }
26.     private static void addTopicPartitions() throws ExecutionException, InterruptedException {
27.         String brokerList = "172.18.30.110:9092";
28.         String topic = "heima";
30.         Properties props = new Properties();
31.         props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, brokerList);
32.         props.put(AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG, 30000);
33.         AdminClient client = AdminClient.create(props);
35.         NewPartitions newPartitions = NewPartitions.increaseTo(4);
37.         Map<String, NewPartitions> newPartitionsMap = new HashMap<>();
38.         newPartitionsMap.put(topic,newPartitions);
39.         CreatePartitionsResult result = client.createPartitions(newPartitionsMap);
40.         result.all().get();
42.         client.close();
43.     }
45.     private static void describeTopicConfig() throws ExecutionException, InterruptedException {
46.         String brokerList = "172.18.30.110:9092";
47.         String topic = "heima";
49.         Properties props = new Properties();
50.         props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, brokerList);
51.         props.put(AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG, 30000);
52.         AdminClient client = AdminClient.create(props);
54.         ConfigResource resource = new ConfigResource(ConfigResource.Type.TOPIC, topic);
55.         DescribeConfigsResult result = client.describeConfigs(Collections.singleton(resource));
56.         Config config = result.all().get().get(resource);
57.         System.out.println(config);
58.         client.close();
59.     }
60. }

复制代码

上一节关联链接请点击
# Kafka_深入探秘者（3）：kafka 消费者

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。