说明

通过 AdminClient 实例，可以对主题进行创建、删除，对主题的分区进行创建、删除等操作

依赖

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>2.4.1</version>
</dependency>
<dependency>
    <groupId>com.fasterxml.jackson.core</groupId>
    <artifactId>jackson-databind</artifactId>
    <version>2.10.1</version>
</dependency>
<dependency>
    <groupId>org.slf4j</groupId>
    <artifactId>slf4j-api</artifactId>
    <version>1.7.30</version>
</dependency>
<dependency>
    <groupId>org.slf4j</groupId>
    <artifactId>slf4j-log4j12</artifactId>
    <version>1.7.30</version>
</dependency>

创建admin对象

Properties properties = new Properties();
properties.setProperty(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092"); // kafka服务器端口号
AdminClient adminClient = AdminClient.create(properties);

关闭

adminClient.close();

创建主题

adminClient.createTopics()

查看主题

ListTopicsResult result = adminClient.listTopics();

工具类

package kafka.util;

import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.admin.ListTopicsResult;

import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.ExecutionException;

public class TopicUtil {

    private String server;
    private AdminClient adminClient;

    public TopicUtil(String server) {
        this.server = server;

        Properties properties = new Properties();
        properties.setProperty(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, server);
        adminClient = AdminClient.create(properties);
    }

    public Set<String> set() throws ExecutionException, InterruptedException {
        ListTopicsResult result = adminClient.listTopics();
        // 打印Topic的名称Set<String>
//        System.out.println(result.names().get());
        return result.names().get();
        // 打印Topic列表的信息Map<String, TopicListing>>
//        System.out.println(result1.namesToListings().get());
        // 打印Topic列表的信息Collection<TopicListing>>
//        System.out.println(result1.listings().get());
    }

    /**
     * 获取主题，按照字典排序
     * @return
     * @throws ExecutionException
     * @throws InterruptedException
     */
    public List<String> list() throws ExecutionException, InterruptedException {
        ListTopicsResult result = adminClient.listTopics();
        Set<String> set = result.names().get();

        // 对主题进行排序
        List<String> retList = new ArrayList<>();
        for(String item:set){
            retList.add(item);
        }
        retList.sort(null);
        return retList;
    }

    public void close(){
        adminClient.close();
    }

    public static void main(String[] args) {
        TopicUtil tu = new TopicUtil("hadoop1:9092");
        try {
            List<String> set = tu.list();
            for(String item : set){
                System.out.println(item);
            }
        } catch (ExecutionException e) {
            throw new RuntimeException(e);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }
}

原文出处：/show_1IX4bY75gsmI.html