当前位置: 首页>编程语言>正文

kafka最新版本镜像 kafka client版本

说明

除了使用Kafka的bin目录下的脚本工具来管理Kafka,还可以使用管理Kafka的API将某些管理查看 的功能集成到系统中。在Kafka0.11.0.0版本之前,可以通过kafka-core包(Kafka的服务端,采用Scala 编写)中的AdminClient和AdminUtils来实现部分的集群管理操作。Kafka0.11.0.0之后,又多了一个 AdminClient,在kafka-client包下,一个抽象类,具体的实现是 org.apache.kafka.clients.admin.KafkaAdminClient。

KafkaAdminClient包含了一下几种功能(以Kafka1.0.2版本为准):

1. 创建主题:

createTopics(final Collection newTopics, final CreateTopicsOptions options)

2. 删除主题:

deleteTopics(final Collection topicNames, DeleteTopicsOptions options)

3. 列出所有主题:

listTopics(final ListTopicsOptions options)

4. 查询主题:

describeTopics(final Collection topicNames, DescribeTopicsOptions options)

5. 查询集群信息:

describeCluster(DescribeClusterOptions options)

6. 查询配置信息:

describeConfigs(Collection configResources, final DescribeConfigsOptions options)

7. 修改配置信息:

alterConfigs(Map configs, final AlterConfigsOptions options)

8. 修改副本的日志目录:

alterReplicaLogDirs(Map replicaAssignment, final AlterReplicaLogDirsOptions options)

9. 查询节点的日志目录信息:

describeLogDirs(Collection brokers, DescribeLogDirsOptions options)

10. 查询副本的日志目录信息:

describeReplicaLogDirs(Collection replicas, DescribeReplicaLogDirsOptions options)

11. 增加分区:

createPartitions(Map newPartitions, final CreatePartitionsOptions options)

用到的参数:

属性

说明

重要性

bootstrap.servers

向Kafka集群建立初始连接用到的host/port列表。 客户端会使用这里列出的所有服务器进行集群其他服 务器的发现,而不管是否指定了哪个服务器用作引 导。 这个列表仅影响用来发现集群所有服务器的初始主 机。 字符串形式:host1:port1,host2:port2,... 由于这组服务器仅用于建立初始链接,然后发现集群 中的所有服务器,因此没有必要将集群中的所有地址 写在这里。 一般最好两台,以防其中一台宕掉。

high

client.id

生产者发送请求的时候传递给broker的id字符串。 用于在broker的请求日志中追踪什么应用发送了什 么消息。 一般该id是跟业务有关的字符串。

medium

connections.max.idle.ms

当连接空闲时间达到这个值,就关闭连接。long型 数据,默认:300000

medium

receive.buffer.bytes

TCP接收缓存(SO_RCVBUF),如果设置为-1,则 使用操作系统默认的值。int类型值,默认65536, 可选值:[-1,...]

medium

request.timeout.ms

客户端等待服务端响应的最大时间。如果该时间超 时,则客户端要么重新发起请求,要么如果重试耗 尽,请求失败。int类型值,默认:120000

medium

security.protocol

跟broker通信的协议:PLAINTEXT, SSL, SASL_PLAINTEXT, SASL_SSL. string类型值,默认:PLAINTEXT

medium

send.buffer.bytes

用于TCP发送数据时使用的缓冲大小 (SO_SNDBUF),-1表示使用OS默认的缓冲区大 小。 int类型值,默认值:131072

medium

reconnect.backoff.max.ms

对于每个连续的连接失败,每台主机的退避将成倍增 加,直至达到此最大值。在计算退避增量之后,添加 20%的随机抖动以避免连接风暴。 long型值,默认1000,可选值:[0,...]

low

reconnect.backoff.ms

重新连接主机的等待时间。避免了重连的密集循环。 该等待时间应用于该客户端到broker的所有连接。 long型值,默认:50

low

retries

The maximum number of times to retry a call before failing it.重试的次数,达到此值,失败。 int类型值,默认5。

low

主要操作步骤:

客户端根据方法的调用创建相应的协议请求,比如创建Topic的createTopics方法,其内部就是发送 CreateTopicRequest请求。

客户端发送请求至Kafka Broker。

Kafka Broker处理相应的请求并回执,比如与CreateTopicRequest对应的是 CreateTopicResponse。

客户端接收相应的回执并进行解析处理。

和协议有关的请求和回执的类基本都在org.apache.kafka.common.requests包中, AbstractRequest和AbstractResponse是这些请求和响应类的两个父类。

综上,如果要自定义实现一个功能,只需要三个步骤:

1. 自定义XXXOptions;

2. 自定义XXXResult返回值;

3. 自定义Call,然后挑选合适的XXXRequest和XXXResponse来实现Call类中的3个抽象方法。

import org.apache.kafka.clients.admin.*;
import org.apache.kafka.common.KafkaFuture;
import org.apache.kafka.common.Node;
import org.apache.kafka.common.TopicPartitionInfo;
import org.apache.kafka.common.config.ConfigResource;
import org.apache.kafka.common.requests.DescribeLogDirsResponse;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.function.BiConsumer;
import java.util.function.Consumer;

public class MyAdminClient {

    private KafkaAdminClient client;

    @Before
    public void before() {
        Map<String, Object> conf = new HashMap<>();
        conf.put("bootstrap.servers", "192.168.25.129:9092");
        conf.put("client.id", "adminclient-1");
        client = (KafkaAdminClient) KafkaAdminClient.create(conf);
    }

    @After
    public void after() {
        client.close();
    }

    @Test
    public void testListTopics1() throws ExecutionException, InterruptedException {

        ListTopicsResult listTopicsResult = client.listTopics();
        // KafkaFuture<Collection<TopicListing>> listings = listTopicsResult.listings();
        // Collection<TopicListing> topicListings = listings.get();
        //
        // topicListings.forEach(new Consumer<TopicListing>() {
        // 		@Override
        //		public void accept(TopicListing topicListing) {
        // 			boolean internal = topicListing.isInternal();
        // 			String name = topicListing.name();
        // 			String s = topicListing.toString();
        // 			System.out.println(s + "\t" + name + "\t" + internal);
        // 		}
        // });
        // KafkaFuture<Set<String>> names = listTopicsResult.names();
        // Set<String> strings = names.get();
        //
        // strings.forEach(name -> {
        // 		System.out.println(name);
        // });
        // KafkaFuture<Map<String, TopicListing>> mapKafkaFuture = listTopicsResult.namesToListings();
        // Map<String, TopicListing> stringTopicListingMap = mapKafkaFuture.get();
        //
        // stringTopicListingMap.forEach((k, v) -> {
        // 		System.out.println(k + "\t" + v);
        // });

        ListTopicsOptions options = new ListTopicsOptions();
        options.listInternal(false);
        options.timeoutMs(500);

        ListTopicsResult listTopicsResult1 = client.listTopics(options);
        Map<String, TopicListing> stringTopicListingMap = listTopicsResult1.namesToListings().get();
        stringTopicListingMap.forEach((k, v) -> {
            System.out.println(k + "\t" + v);
        });

        // 关闭管理客户端
        client.close();
    }

    @Test
    public void testCreateTopic() throws ExecutionException, InterruptedException {

        Map<String, String> configs = new HashMap<>();
        configs.put("max.message.bytes", "1048576");
        configs.put("segment.bytes", "1048576000");


        NewTopic newTopic = new NewTopic("test_topic", 2, (short) 1);
        newTopic.configs(configs);
        CreateTopicsResult topics = client.createTopics(Collections.singleton(newTopic));
        KafkaFuture<Void> all = topics.all();
        Void aVoid = all.get();
        System.out.println(aVoid);
    }

    @Test
    public void testDeleteTopic() throws ExecutionException, InterruptedException {

        DeleteTopicsOptions options = new DeleteTopicsOptions();
        options.timeoutMs(500);
        DeleteTopicsResult deleteResult = client.deleteTopics(Collections.singleton("test_topic"), options);
        deleteResult.all().get();
    }


    @Test
    public void testAlterTopic() throws ExecutionException, InterruptedException {

        NewPartitions newPartitions = NewPartitions.increaseTo(5);
        Map<String, NewPartitions> newPartitionsMap = new HashMap<>();
        newPartitionsMap.put("test_topic", newPartitions);
        CreatePartitionsOptions option = new CreatePartitionsOptions();
        // Set to true if the request should be validated without creating
        new partitions.
            // 如果只是验证,而不创建分区,则设置为true
            // option.validateOnly(true);
            CreatePartitionsResult partitionsResult = client.createPartitions(newPartitionsMap, option);
        Void aVoid = partitionsResult.all().get();
    }

    @Test
    public void testDescribeTopics() throws ExecutionException, InterruptedException {

        DescribeTopicsOptions options = new DescribeTopicsOptions();
        options.timeoutMs(3000);
        DescribeTopicsResult topicsResult = client.describeTopics(Collections.singleton("test_topic"), options);
        Map<String, TopicDescription> stringTopicDescriptionMap = topicsResult.all().get();
        stringTopicDescriptionMap.forEach((k, v) -> {

            System.out.println(k + "\t" + v);
            System.out.println("=======================================");
            System.out.println(k);
            boolean internal = v.isInternal();
            String name = v.name();
            List<TopicPartitionInfo> partitions = v.partitions();
            String partitionStr = Arrays.toString(partitions.toArray());
            System.out.println("内部的?" + internal);
            System.out.println("topic name = " + name);
            System.out.println("分区:" + partitionStr);

            partitions.forEach(partition -> {
                System.out.println(partition);
            });
        });
    }

    @Test
    public void testDescribeCluster() throws ExecutionException, InterruptedException {

        DescribeClusterResult describeClusterResult = client.describeCluster();
        KafkaFuture<String> stringKafkaFuture = describeClusterResult.clusterId();
        String s = stringKafkaFuture.get();
        System.out.println("cluster name = " + s);
        KafkaFuture<Node> controller = describeClusterResult.controller();
        Node node = controller.get();
        System.out.println("集群控制器:" + node);
        Collection<Node> nodes = describeClusterResult.nodes().get();
        nodes.forEach(node1 -> {
            System.out.println(node1);
        });
    }

    @Test
    public void testDescribeConfigs() throws ExecutionException, InterruptedException, TimeoutException {

        ConfigResource configResource = new ConfigResource(ConfigResource.Type.BROKER, "0");
        DescribeConfigsResult describeConfigsResult = client.describeConfigs(Collections.singleton(configResource));
        Map<ConfigResource, Config> configMap = describeConfigsResult.all().get(15, TimeUnit.SECONDS);

        configMap.forEach(new BiConsumer<ConfigResource, Config>() {

            @Override
            public void accept(ConfigResource configResource, Config config) {
                ConfigResource.Type type = configResource.type();
                String name = configResource.name();
                System.out.println("资源名称:" + name);
                Collection<ConfigEntry> entries = config.entries();

                entries.forEach(new Consumer<ConfigEntry>() {

                    @Override
                    public void accept(ConfigEntry configEntry) {
                        boolean aDefault = configEntry.isDefault();
                        boolean readOnly = configEntry.isReadOnly();
                        boolean sensitive = configEntry.isSensitive();
                        String name1 = configEntry.name();
                        String value = configEntry.value();
                        System.out.println("是否默认:" + aDefault + "\t是否只读?" + readOnly + "\t是否敏感?" + sensitive + "\t" + name1 + " --> " + value);
                    }
                });

                ConfigEntry retries = config.get("retries");
                if (retries != null) {
                    System.out.println(retries.name() + " -->" + retries.value());
                } else {
                    System.out.println("没有这个属性");
                }
            }
        });
    }

    @Test
    public void testAlterConfig() throws ExecutionException, InterruptedException {

        // 这里设置后,原来资源中不冲突的属性也会丢失,直接按照这里的配置设置
        Map<ConfigResource, Config> configMap = new HashMap<>();
        ConfigResource resource = new ConfigResource(ConfigResource.Type.TOPIC, "test_topic");
        Config config = new Config(Collections.singleton(new ConfigEntry("segment.bytes", "1048576000")));
        configMap.put(resource, config);
        AlterConfigsResult alterConfigsResult = client.alterConfigs(configMap);
        Void aVoid = alterConfigsResult.all().get();
    }

    @Test
    public void testDescribeLogDirs() throws ExecutionException, InterruptedException {

        DescribeLogDirsOptions option = new DescribeLogDirsOptions();
        option.timeoutMs(1000);
        DescribeLogDirsResult describeLogDirsResult = client.describeLogDirs(Collections.singleton(0), option);
        Map<Integer, Map<String, DescribeLogDirsResponse.LogDirInfo>>integerMapMap = describeLogDirsResult.all().get();
        integerMapMap.forEach(new BiConsumer<Integer, Map<String,DescribeLogDirsResponse.LogDirInfo>>() {

            @Override
            public void accept(Integer integer, Map<String, DescribeLogDirsResponse.LogDirInfo> stringLogDirInfoMap) {

                System.out.println("broker.id = " + integer);
                stringLogDirInfoMap.forEach(new BiConsumer<String,DescribeLogDirsResponse.LogDirInfo>() {

                    @Override
                    public void accept(String s, DescribeLogDirsResponse.LogDirInfo logDirInfo) {

                        System.out.println("log.dirs:" + s);

                        // 查看该broker上的主题/分区/偏移量等信息
                        // logDirInfo.replicaInfos.forEach(new BiConsumer<TopicPartition, DescribeLogDirsResponse.ReplicaInfo>() {
                        // 
                        // 		@Override
                        //		public void accept(TopicPartition topicPartition, DescribeLogDirsResponse.ReplicaInfo replicaInfo) {
                        // 			int partition = topicPartition.partition();
                        // 			String topic = topicPartition.topic();
                        // 			boolean isFuture = replicaInfo.isFuture;
                        // 			long offsetLag = replicaInfo.offsetLag;
                        // 			long size = replicaInfo.size;
                        // 			System.out.println("partition:" + partition + "\ttopic:" + topic
                        // 			+ "\tisFuture:" + isFuture
                        // 			+ "\toffsetLag:" + offsetLag
                        // 			+ "\tsize:" + size);
                        // 		}
                        // });
                    }
                });
            }
        });
    }
}

https://www.xamrdz.com/lan/5qx1931489.html

相关文章: