五类API
Kafka客户端API类型
- AdminClient API:允许管理和检测Topic、broker以及其它Kafka对象(类似于命令行的create topic)
- Producer API:发送消息到1个或多个Topic
- Consumer API:订阅一个或多个Topic,并处理产生的消息
- Streams API:高效地将输入流转换到输出流
- Connector API:从一些源系统或应用程序中拉取数据到kafka
初始化工程
创建一个SpringBoot项目
引入依赖
<!-- https://mvnrepository.com/artifact/org.apache.kafka/kafka-clients --> <dependency> <groupId>org.apache.kafka</groupId> <artifactId>kafka-clients</artifactId> <version>2.4.0</version> </dependency>
AdminClient客户端API的使用
客户端建立
public class AdminSample { public static void main(String[] args) { AdminClient adminClient = AdminSample.adminClient(); System.out.println("adminClient: " + adminClient); } /** * 设置AdminClient */ public static AdminClient adminClient(){ Properties properties = new Properties(); properties.setProperty(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.75.136:9092"); AdminClient adminClient = AdminClient.create(properties); return adminClient; } }
连接测试成功
创建Topic
public final static String TOPIC_NAME = "topic_test"; /** * 创建Topic实例 */ public static void createTopic(){ AdminClient adminClient = adminClient(); // 副本因子 Short rs = 1; NewTopic newTopic = new NewTopic(TOPIC_NAME, 1, rs); CreateTopicsResult topics = adminClient.createTopics(Arrays.asList(newTopic)); System.out.println("CreateTopicsResult: " + topics); } public static void main(String[] args) { // AdminClient adminClient = AdminSample.adminClient(); // System.out.println("adminClient: " + adminClient); createTopic(); }
创建成功
查看Topic列表及Internal
查看是否成功创建topic
/** * 获取Topic列表 */ public static void topicList() throws Exception { AdminClient adminClient = adminClient(); ListTopicsResult listTopicsResult = adminClient.listTopics(); Set<String> names = listTopicsResult.names().get(); //打印 names.stream().forEach(System.out::println); }
public static void topicList() throws Exception { AdminClient adminClient = adminClient(); //是否查看Internal选项 ListTopicsOptions options = new ListTopicsOptions(); options.listInternal(true); //ListTopicsResult listTopicsResult = adminClient.listTopics(); ListTopicsResult listTopicsResult = adminClient.listTopics(options); Set<String> names = listTopicsResult.names().get(); Collection<TopicListing> topicListings = listTopicsResult.listings().get(); //打印 names.stream().forEach(System.out::println); //打印topicListing topicListings.stream().forEach((topicList)->{ System.out.println(topicList); }); }
删除Topic
/** * 删除Topic */ public static void delTopics() throws Exception { AdminClient adminClient = adminClient(); DeleteTopicsResult deleteTopicsResult = adminClient.deleteTopics(Arrays.asList(TOPIC_NAME)); deleteTopicsResult.all().get(); }
成功删除
Topic描述信息查看
/** * 描述Topic */ public static void describeTopics() throws Exception { AdminClient adminClient = adminClient(); DescribeTopicsResult describeTopicsResult = adminClient.describeTopics(Arrays.asList(TOPIC_NAME)); Map<String, TopicDescription> stringTopicDescriptionMap = describeTopicsResult.all().get(); Set<Map.Entry<String, TopicDescription>> entries = stringTopicDescriptionMap.entrySet(); entries.stream().forEach((entry)->{ System.out.println("name: " + entry.getKey() + ", desc: " + entry.getValue()); }); }
输出以下内容
Topic配置信息查看
/** * 查看配置 */ public static void describeConfig() throws Exception { AdminClient adminClient = adminClient(); //TODO 预留,用来做集群 ConfigResource configResource = new ConfigResource(ConfigResource.Type.TOPIC, TOPIC_NAME); DescribeConfigsResult describeConfigsResult = adminClient.describeConfigs(Arrays.asList(configResource)); Map<ConfigResource, Config> configResourceConfigMap = describeConfigsResult.all().get(); configResourceConfigMap.entrySet().stream().forEach((entry)->{ System.out.println("configResource: " + entry.getKey() + ", Config: " + entry.getValue()); }); }
输出信息
des:configResource: ConfigResource(type=TOPIC, name='topic_test'), Config: Config(entries=[ConfigEntry(name=compression.type, value=producer, source=DEFAULT_CONFIG, isSensitive=false, isReadOnly=false, synonyms=[]), ConfigEntry(name=leader.replication.throttled.replicas, value=, source=DEFAULT_CONFIG, isSensitive=false, isReadOnly=false, synonyms=[]), 。。。。。。
Topic修改描述信息
/** * 修改配置 */ public static void alterConfig() throws Exception { AdminClient adminClient = adminClient(); Map<ConfigResource, Config> configMaps = new HashMap<>(); ConfigResource configResource = new ConfigResource(ConfigResource.Type.TOPIC, TOPIC_NAME); Config config = new Config(Arrays.asList(new ConfigEntry("preallocate", "true"))); configMaps.put(configResource, config); AlterConfigsResult alterConfigsResult = adminClient.alterConfigs(configMaps); alterConfigsResult.all().get(); }
Patition增加
/** * 增加Partition数量 */ public static void incrPartitions(int partitions) throws Exception { AdminClient adminClient = adminClient(); Map<String, NewPartitions> partitionsMap = new HashMap<>(); NewPartitions newPartitions = NewPartitions.increaseTo(partitions); partitionsMap.put(TOPIC_NAME, newPartitions); CreatePartitionsResult createPartitionsResult = adminClient.createPartitions(partitionsMap); createPartitionsResult.all().get(); }
标签:adminClient,void,Kafka,Topic,AdminClient,static,操作,public,客户端 From: https://www.cnblogs.com/szhNJUPT/p/17574421.html