关于springboot创建kafkaTopic

工具类提供，方法名见名知意。使用kafka admin

import org.apache.kafka.clients.admin.*;
import org.apache.kafka.common.KafkaFuture;

import java.util.*;
import java.util.concurrent.ExecutionException;



import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.admin.ListTopicsOptions;
import org.apache.kafka.clients.admin.TopicDescription;
import org.apache.kafka.common.KafkaFuture;
import org.apache.kafka.common.TopicPartitionInfo;

import java.util.Properties;
import java.util.Set;
import java.util.concurrent.ExecutionException;
/**
 * @author: zhoumo
 * @data: 2024/6/24 16:37
 * @descriptions:
 */
public class KafkaTopicInfo {

    final static String ip="127.0.0.1:9090";


    public static void main(String[] args) {
        getListDetail();
    }


    
    public static void createTopic(String topicName) throws ExecutionException, InterruptedException {
        // Kafka 配置
        Properties props = new Properties();
        // Kafka 服务器地址和端口
        props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, ip);

        // 创建 AdminClient 实例
        try (AdminClient adminClient = AdminClient.create(props)) {
            // 创建一个新的主题
            // 指定分区数量
            // 指定复制因子
            int numPartitions = 2;
            short replicationFactor = 1;

            NewTopic newTopic = new NewTopic(topicName, numPartitions, replicationFactor);

            // 创建主题
            adminClient.createTopics(Collections.singletonList(newTopic)).all().get();
            System.out.println("Topic created successfully: " + topicName);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public static void deleteTopic(String topicName) {
        // Kafka 配置
        Properties props = new Properties();
        props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, ip);

        try (AdminClient adminClient = AdminClient.create(props)) {
            // 要删除的主题名称
         //   String topicName = "myTopic";

            // 删除主题
            DeleteTopicsResult deleteResult = adminClient.deleteTopics(Collections.singletonList(topicName));
            deleteResult.all().get();

            System.out.println("Topic deleted successfully: " + topicName);

        } catch (ExecutionException | InterruptedException e) {
            e.printStackTrace();
        }
    }
    public static void getList() {
        // Kafka 配置
        Properties props = new Properties();
        props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, ip);

        try (AdminClient adminClient = AdminClient.create(props)) {
            // 列出所有主题
            ListTopicsOptions options = new ListTopicsOptions();
            // 是否包括内部主题，默认为 false
            options.listInternal(true);
            ListTopicsResult topicsResult = adminClient.listTopics(options);
            Set<String> topics = topicsResult.names().get();

            System.out.println("Existing topics:");
            for (String topic : topics) {
                System.out.println(topic);
            }

        } catch (ExecutionException | InterruptedException e) {
            e.printStackTrace();
        }
    }



    public static void getListDetail() {
        // Kafka 配置
        Properties props = new Properties();
        props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, ip);

        try (AdminClient adminClient = AdminClient.create(props)) {
            // 列出所有主题
            ListTopicsOptions options = new ListTopicsOptions();
            // 是否包括内部主题，默认为 false
            options.listInternal(true);
            KafkaFuture<Set<String>> topics = adminClient.listTopics(options).names();

            System.out.println("Existing topics:");
            for (String topic : topics.get()) {
                System.out.println(topic);
                // 获取主题的详细信息（包括分区情况）
                /*
                回退jdk1.8 版本
                KafkaFuture<TopicDescription> topicDescription = adminClient.describeTopics(Set.of(topic)).values().get(topic);
                printTopicDetails(topicDescription.get());
                */
                Set<String> topicSet = new HashSet<>();
                topicSet.add(topic);
                KafkaFuture<TopicDescription> topicDescriptionFuture = adminClient.describeTopics(topicSet).values().get(topic);
                TopicDescription topicDescription = topicDescriptionFuture.get();
                printTopicDetails(topicDescription);
            }

        } catch (ExecutionException | InterruptedException e) {
            e.printStackTrace();
        }
    }
    private static void printTopicDetails(TopicDescription topicDescription) {
        System.out.println("Topic: " + topicDescription.name());
        System.out.println("Partitions:");

        for (TopicPartitionInfo partition : topicDescription.partitions()) {
            System.out.printf("  Partition %d, Leader: %d, Replicas: %s, Isrs: %s%n",
                    partition.partition(),
                    partition.leader().id(),
                    partition.replicas(),
                    partition.isr());
        }
        System.out.println();
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150