写在前面
众所周知，kafka是现代流行的消息队列，它使用经典的消息订阅发布模式实现消息的流转，大部分代码结合kafka使用都是使用它的生产者和消费者来实现消息的传递，那么对于kafka的主题的管理怎么使用代码实现呢，这是今天要讲的主题
命令行模式
kafka要结合zookeeper使用，因为它把元数据信息交给了zookeeper管理，其实使用命令行命令很容易就能对topic进行管理，主要使用的命令是kafka-topics.sh
创建主题 kafka-topics.sh --zookeeper localhost:2181 --create --topic my-topic --replication-factor 3 --partitions 3 查看主题列表 kafka-topics.sh --zookeeper localhost:2181 --list 查看主题状态 kafka-topics.sh --describe --zookeeper 127.0.0.1:2181 --topic TestTopic
代码模式
那么话说回来如何使用代码实现topic的管理呢，那么现在就来看一下代码的实现方式，此处使用springboot2框架实现。
首先引进依赖kafka的相关
<dependency> <groupId>org.springframework.kafka</groupId> <artifactId>spring-kafka</artifactId> </dependency>
创建一个测试类进行测试
public static void main(String[] args) {
Properties properties = new Properties();
properties.put("bootstrap.servers", "10.0.59.11:9093");
properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
AdminClient adminClient = AdminClient.create(properties);
ListTopicsResult result = adminClient.listTopics();
KafkaFuture<Set<String>> names = result.names();
try {
names.get().forEach((k)->{
System.out.println(k);
});
} catch (InterruptedException | ExecutionException e) {
e.printStackTrace();
}
adminClient.close();
}这里面最主要的就是AdminClient这个类，AdminClient实现了Admin接口，Admin里面定义了许多和kafka配置相关的东西
让我们依次来看一下
public abstract class AdminClient implements Admin {
public AdminClient() {
}
public static AdminClient create(Properties props) {
return (AdminClient)Admin.create(props);
}
public static AdminClient create(Map<String, Object> conf) {
return (AdminClient)Admin.create(conf);
}
}而Admin接口里有以下方法
static Admin create(Properties props) {
return KafkaAdminClient.createInternal(new AdminClientConfig(props, true), (TimeoutProcessorFactory)null);
}
static Admin create(Map<String, Object> conf) {
return KafkaAdminClient.createInternal(new AdminClientConfig(conf, true), (TimeoutProcessorFactory)null);
}
default void close() {
this.close(9223372036854775807L, TimeUnit.MILLISECONDS);
}
/** @deprecated */
@Deprecated
default void close(long duration, TimeUnit unit) {
this.close(Duration.ofMillis(unit.toMillis(duration)));
}
void close(Duration var1);
default CreateTopicsResult createTopics(Collection<NewTopic> newTopics) {
return this.createTopics(newTopics, new CreateTopicsOptions());
}
CreateTopicsResult createTopics(Collection<NewTopic> var1, CreateTopicsOptions var2);
default DeleteTopicsResult deleteTopics(Collection<String> topics) {
return this.deleteTopics(topics, new DeleteTopicsOptions());
}
DeleteTopicsResult deleteTopics(Collection<String> var1, DeleteTopicsOptions var2);
default ListTopicsResult listTopics() {
return this.listTopics(new ListTopicsOptions());
}
ListTopicsResult listTopics(ListTopicsOptions var1);
default DescribeTopicsResult describeTopics(Collection<String> topicNames) {
return this.describeTopics(topicNames, new DescribeTopicsOptions());
}
DescribeTopicsResult describeTopics(Collection<String> var1, DescribeTopicsOptions var2);
default DescribeClusterResult describeCluster() {
return this.describeCluster(new DescribeClusterOptions());
}
DescribeClusterResult describeCluster(DescribeClusterOptions var1);
default DescribeAclsResult describeAcls(AclBindingFilter filter) {
return this.describeAcls(filter, new DescribeAclsOptions());
}
DescribeAclsResult describeAcls(AclBindingFilter var1, DescribeAclsOptions var2);通过名称我们可以看出，里面有创建Topic，有删除Topic，有列出所有Topic,有描述Topic
我们通过这些方法可以管理Kafka的Topic
最后我们来看一下实现效果
控制台打印里面有3个Topic
去服务器命令行验证一下
也是3个说明代码没问题
总结
网上大多数关于kafka的代码实现都是关于生产者和消费者的实现，今天主要是使用一下kakfa的配置管理类，实现对topic的管理，以此记录作为以后工作中的参考。希望能给大家一个参考，也希望大家多多支持Devmax。