kafka实体演示IDEA

package com.xinchen.kafka.utils; import com.xinchen.kafka.demo.MyKafkaProperties; import kafka.admin.AdminUtils; import kafka.admin.RackAwareMode; import kafka.utils.ZkUtils; import lombok.extern.slf4j.Slf4j; import org.I0Itec.zkclient.ZkClient; import org.apache.kafka.clients.admin.*; import org.apache.kafka.common.Node; import org.apache.kafka.common.config.ConfigResource; import scala.collection.Seq; import java.util.*; import java.util.concurrent.ExecutionException; /** * @Author: ASUS * @Data:Created in 2017/12/11 17:07 * @Version: * @Acton: */ @Slf4j public class TopicCRUD{ private final static String ZOOKEEPER_CONNECT = MyKafkaProperties.zkConnect; private static ZkClient zkClient = new ZkClient(ZOOKEEPER_CONNECT); /** 创建topic */ public static String createTopic(String topicName, int partition, int replication, Properties properties,ZkUtilsValue value){ ZkUtils zkUtils = ZkUtils.apply(value.getZkConnection(), value.getSessionTimeout(),value.getConnectionTimout(), value.getIsTrue()); AdminUtils.createTopic(zkUtils,topicName,partition,replication,properties, RackAwareMode.Enforced$.MODULE$); log.info("创建成功！"); zkUtils.close(); return topicName; } /** 删除topic */ public static void deleteTopic(String topicName,ZkUtilsValue value){ ZkUtils zkUtils = ZkUtils.apply(value.getZkConnection(), value.getSessionTimeout(),value.getConnectionTimout(), value.getIsTrue()); AdminUtils.deleteTopic(zkUtils,topicName); log.info("删除成功！"); zkUtils.close(); } /** 描述主题 */ public static void describeCluster(AdminClient client) throws ExecutionException, InterruptedException { DescribeClusterResult ret = client.describeCluster(); System.out.println(String.format("Cluster id: %s, controller: %s", ret.clusterId().get(), ret.controller().get())); System.out.println("Current cluster nodes info: "); for (Node node : ret.nodes().get()) { System.out.println(node); } } /** 描述主题的配置 */ public static void describeConfig(AdminClient client,String topicName) throws ExecutionException, InterruptedException { DescribeConfigsResult ret = client.describeConfigs(Collections.singleton(new ConfigResource(ConfigResource.Type.TOPIC,topicName))); Map<ConfigResource, Config> configs = ret.all().get(); for (Map.Entry<ConfigResource, Config> entry : configs.entrySet()) { ConfigResource key = entry.getKey(); Config value = entry.getValue(); System.out.println(String.format("Resource type:%s,resource name:%s",key.type(),key.name())); Collection<ConfigEntry> entries = value.entries(); for (ConfigEntry configEntry : entries) { System.out.println(configEntry.name() + "=" +configEntry.value()); } } } /** 更改主题的配置 */ public static void alterConfigs(AdminClient client,String topicName,ConfigEntry configEntry) throws ExecutionException, InterruptedException { Config topicConfig = new Config(Arrays.asList(configEntry)); client.alterConfigs(Collections.singletonMap(new ConfigResource(ConfigResource.Type.TOPIC,topicName),topicConfig)).all().get(); } /** 描述指定的主题 */ public static void describeTopics(AdminClient client, String topicName) throws ExecutionException, InterruptedException { DescribeTopicsResult ret = client.describeTopics(Arrays.asList(topicName, "__consumer_offsets")); Map<String, TopicDescription> map = ret.all().get(); for (Map.Entry<String, TopicDescription> entry : map.entrySet()) { System.out.println(entry.getKey()+ "===>" +entry.getValue()); } } /** 打印集群中所有的主题 */ public static void listAllTopic(AdminClient client) throws ExecutionException, InterruptedException { ListTopicsOptions topicsOptions = new ListTopicsOptions(); topicsOptions.listInternal(true); ListTopicsResult listTopics = client.listTopics(topicsOptions); Set<String> topicNames = listTopics.names().get(); System.out.println("Current topics in this cluster:" + topicNames); } }