kafka java api客户端编程

内容导读

互联网集市收集整理的这篇技术教程文章主要介绍了kafka java api客户端编程，小编现在分享给大家，供广大互联网技能从业者学习和参考。文章包含5382字，纯文字阅读大概需要8分钟。

内容图文

环境

Ubuntu18.04
zookeeper3.4.13
kafka2.1.1

说明

因为是使用了最新的kafka，所以很多方法都过时了。研究了很久新api的用法，然后在此记录。
zkUtil已经不用了，改用AdminClient。
主要功能包括：

创建Topic：createTopics(Collection newTopics)
删除Topic：deleteTopics(Collection topics)
显示所有Topic：listTopics()
查询Topic：describeTopics(Collection topicNames)
查询集群信息：describeCluster()
查询ACL信息：describeAcls(AclBindingFilter filter)
创建ACL信息：createAcls(Collection acls)
删除ACL信息：deleteAcls(Collection filters)
查询配置信息：describeConfigs(Collection resources)
修改配置信息：alterConfigs(Map<ConfigResource, Config> configs)
修改副本的日志目录：alterReplicaLogDirs(Map<TopicPartitionReplica, String> replicaAssignment)
查询节点的日志目录信息：describeLogDirs(Collection brokers)
查询副本的日志目录信息：describeReplicaLogDirs(Collection replicas)
增加分区：createPartitions(Map<String, NewPartitions> newPartitions)
官方文档
我将前5个功能实现了一下。接下来是代码。

代码

package datacool.hadoop.kafka;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Properties;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;

import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.DescribeClusterResult;
import org.apache.kafka.clients.admin.DescribeTopicsResult;
import org.apache.kafka.clients.admin.ListTopicsResult;
import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.clients.admin.TopicDescription;
import org.apache.kafka.clients.admin.TopicListing;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.KafkaFuture;
import org.apache.kafka.common.Node;
import org.apache.kafka.common.TopicPartitionInfo;

import datacool.hadoop.common.IPUtil;
public class KafkaUtil {

	private static Properties props;//配置信息
	private static AdminClient adminClient;//kafka的admin
	
	/*
	 * 加载配置项
	 * */
	static {
		props = new Properties();
		props.put("bootstrap.servers", "localhost:9092");//运行kafka客户端的主机
        props.put("acks", "all");
        props.put("retries", 0);
        props.put("batch.size", 16384);
        props.put("linger.ms", 1);
        props.put("buffer.memory", 33554432);
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");//序列化方式
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        System.out.println("配置文件准备成功");
	}
	/*
	 * 创建话题
	 * */
    public static void createTopic(String topic) {
    	adminClient= AdminClient.create(props);
		NewTopic newTopic = new NewTopic(topic,1,(short)1);
		Collection<NewTopic> newTopicList = new ArrayList<>();
		newTopicList.add(newTopic);
		adminClient.createTopics(newTopicList);
		adminClient.close();
		System.out.println("创建主题成功："+topic);
    }
    /*
	 * 删除话题
	 * */
    public static void deleteTopic(String topic) {
    	adminClient= AdminClient.create(props);
		Collection<String> TopicList = new ArrayList<>();
		TopicList.add(topic);
		adminClient.deleteTopics(TopicList);
		adminClient.close();
		System.out.println("删除主题成功："+topic);
    }
    /*
	 * 列出所有话题
	 * */
    public static void listTopics() {
    	adminClient= AdminClient.create(props);
    	ListTopicsResult topicList = adminClient.listTopics();
    	try {
			Collection<TopicListing> collection = topicList.listings().get();
			Iterator<TopicListing> it = collection.iterator();
			while(it.hasNext()) {
				System.out.println(it.next().name());
			}
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (ExecutionException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
    	adminClient.close();
    	System.out.println("成功列出所有主题");
    }
    
    
    public static void describeTopics(String topic) {
    	adminClient= AdminClient.create(props);
		Collection<String> topicList = new ArrayList<>();
		topicList.add(topic);
    	DescribeTopicsResult descTopic = adminClient.describeTopics(topicList);
    	
    	try {
			Map<String, TopicDescription> map = descTopic.all().get();
			TopicDescription topicDescription = map.get("hope");
			System.out.println("hope的描述："+topicDescription);
			//后面都是如何具体的取出细节。我没有写输出了
			topicDescription.name();
			topicDescription.isInternal();
			List<TopicPartitionInfo> partitions = topicDescription.partitions();
			Iterator<TopicPartitionInfo> iterator = partitions.iterator();
			while(iterator.hasNext()) {
				TopicPartitionInfo next = iterator.next();
				next.partition();
				Node leader = next.leader();
				leader.host();
				leader.port();
				leader.id();
				leader.rack();
				List<Node> replicas = next.replicas();
				for(Node it:replicas) {
					it.host();
					it.port();
					it.id();
					it.rack();
				}
				List<Node> isr = next.isr();
				for(Node it:isr) {
					it.host();
					it.port();
					it.id();
					it.rack();
				}
			}
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (ExecutionException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
    	
    	adminClient.close();
    	System.out.println("成功描述主题："+topic);
    }
    
    public static void describeCluster() {
    	adminClient= AdminClient.create(props);
    	DescribeClusterResult describeCluster = adminClient.describeCluster();
    	
    	describeCluster.clusterId();
    	describeCluster.controller();
    	
    	adminClient.close();
    	System.out.println("成功描述集群");
    }
    
    
    public static void main(String[] args) {
    	//测试用
	}
}

内容总结

以上是互联网集市为您收集整理的kafka java api客户端编程全部内容，希望文章能够帮你解决kafka java api客户端编程所遇到的程序开发问题。 如果觉得互联网集市技术教程内容还不错，欢迎将互联网集市网站推荐给程序员好友。

内容备注

版权声明：本文内容由互联网用户自发贡献，该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 gblab@vip.qq.com 举报，一经查实，本站将立刻删除。

内容手机端