Kafka的运维利器

您所在的位置：网站首页 › websphere删除单元 › Kafka的运维利器

Kafka的运维利器

2023-10-25 15:20| 来源: 网络整理| 查看: 265

前言

一般情况下，我们都习惯使用kafka-topics.sh脚本来管理主题，但有些时候我们希望将主题管理类的功能集成到公司内部的系统中，打造集管理、监控、运维、告警为一体的生态平台，那么就需要以程序调用API的方式去实现。

Kafka社区于0.11版本正式推出了Java客户端版的AdminClient，并不断地在后续的版本中对它进行完善。

本文主要介绍KafkaAdminClient 的基本使用方式，以及采用这种调用API方式下的创建主题时的合法性验证。

功能

鉴于社区还在不断地完善 AdminClient 的功能，AdminClient 提供的功能有以下几个大类。

主题管理：包括主题的创建、删除和查询。权限管理：包括具体权限的配置与删除。配置参数管理：包括 Kafka 各种资源的参数设置、详情查询。所谓的 Kafka 资源，主要有 Broker、主题、用户、Client-id 等。副本日志管理：包括副本底层日志路径的变更和详情查询。分区管理：即创建额外的主题分区。消息删除：即删除指定位移之前的分区消息。Delegation Token 管理：包括 Delegation Token 的创建、更新、过期和详情查询。消费者组管理：包括消费者组的查询、位移查询和删除。Preferred 领导者选举：推选指定主题分区的 Preferred Broker 为领导者。工作原理

AdminClient 是一个双线程的设计：前端主线程和后端 I/O 线程。

前端线程负责将用户要执行的操作转换成对应的请求，然后再将请求发送到后端 I/O 线程的队列中。

而后端 I/O 线程（kafka-admin-client-thread）从队列中读取相应的请求，然后发送到对应的 Broker 节点上，之后把执行结果保存起来，以便等待前端线程的获取。

使用

如果你使用的是 Maven，需要增加以下依赖项：

org.apache.kafka kafka-clients 2.6.5 构建AdminClient/** * 创建AdminClient客户端对象 */ public static AdminClient createAdminClientByProperties() { Properties prop = new Properties(); // 配置Kafka服务的访问地址及端口号 prop.setProperty(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG,"127.0.0.1:9092"); // 创建AdminClient实例 return AdminClient.create(prop); } /** * 创建AdminClient的第二种方式 */ public static AdminClient createAdminClientByMap(){ Map conf = Maps.newHashMap(); // 配置Kafka服务的访问地址及端口号 conf.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG,"127.0.0.1:9092"); // 创建AdminClient实例 return AdminClient.create(conf); } 创建Topic实例private static final String TOPIC_NAME = "test_topic"; /** * 创建Topic实例 */ public static void createTopic(){ AdminClient adminClient = AdminSample.adminClient(); //副本因子 Short re = 1; NewTopic newTopic = new NewTopic(TOPIC_NAME,1,re); CreateTopicsResult createTopicsResult = adminClient.createTopics(Arrays.asList(newTopic)); System.out.println("CreateTopicsResult : " + createTopicsResult); adminClient.close(); } 查询Topic列表private static final String TOPIC_NAME = "test_topic"; /** * 获取topic列表 */ 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 names = listTopicsResult.names().get(); //打印names names.stream().forEach(System.out::println); Collection topicListings = listTopicsResult.listings().get(); //打印TopicListing topicListings.stream().forEach((topicList) -> { System.out.println(topicList.toString()); }); adminClient.close(); } 删除topicprivate static final String TOPIC_NAME = "test_topic"; /** * 删除topic */ public static void delTopic() throws Exception { AdminClient adminClient = adminClient(); DeleteTopicsResult deleteTopicsResult = adminClient.deleteTopics(Arrays.asList(TOPIC_NAME)); deleteTopicsResult.all().get(); } 描述topic/** * 获取topic的描述信息 */ public static void describeTopics(List topics) throws Exception { // 创建AdminClient客户端对象 AdminClient adminClient = BuildAdminClient.createAdminClientByProperties(); // 获取Topic的描述信息 DescribeTopicsResult result = adminClient.describeTopics(topics); // 解析描述信息结果, Map ==> topicName:topicDescription Map topicDescriptionMap = result.all().get(); topicDescriptionMap.forEach((topicName, description) -> System.out.printf("topic name = %s, desc = %s \n", topicName, description)); // 关闭资源 adminClient.close(); } 查看 Topic 的配置信息

除了Kafka自身的配置项外，其内部的Topic也会有非常多的配置项，我们可以通过describeConfigs方法来获取某个Topic中的配置项信息。代码示例：

/** * 获取topic的配置信息 */ public static void describeConfigTopics(List topicNames) throws Exception { // 创建AdminClient客户端对象 AdminClient adminClient = BuildAdminClient.createAdminClientByMap(); List configResources = Lists.newArrayListWithCapacity(64); topicNames.forEach(topicName -> configResources.add( // 指定要获取的源 new ConfigResource(ConfigResource.Type.TOPIC, topicName))); // 获取topic的配置信息 DescribeConfigsResult result = adminClient.describeConfigs(configResources); // 解析topic的配置信息 Map resourceConfigMap = result.all().get(); resourceConfigMap.forEach((configResource, config) -> System.out.printf("topic config ConfigResource = %s, Config = %s \n", configResource, config)); // 关闭资源 adminClient.close(); } 修改 Topic 的分区数量

在创建Topic时我们需要设定Partition的数量，但如果觉得初始设置的Partition数量太少了，那么就可以使用createPartitions方法来调整Topic的Partition数量，但是需要注意在Kafka中Partition只能增加不能减少。代码示例：

/** * 修改topic的分区数量 * 只能增加不能减少 */ public static void updateTopicPartition(List topicNames, Integer partitionNum) throws Exception { // 创建AdminClient客户端对象 AdminClient adminClient = BuildAdminClient.createAdminClientByMap(); // 构建修改分区的topic请求参数 Map newPartitions = Maps.newHashMap(); topicNames.forEach(topicName -> newPartitions.put(topicName, NewPartitions.increaseTo(partitionNum))); // 执行修改 CreatePartitionsResult result = adminClient.createPartitions(newPartitions); // get方法是一个阻塞方法，一定要等到createPartitions完成之后才进行下一步操作 result.all().get(); // 关闭资源 adminClient.close(); }

社区于 0.11 版本正式推出了 Java 客户端版的 AdminClient 工具，该工具提供了几十种运维操作，而且它还在不断地演进着。如果可以的话，你最好统一使用 AdminClient 来执行各种 Kafka 集群管理操作，摒弃掉连接 ZooKeeper 的那些工具。另外，建议时刻关注该工具的功能完善情况，毕竟，目前社区对 AdminClient 的变更频率很高。

【本文地址】

Kafka的运维利器

Kafka的运维利器

今日新闻

推荐新闻