kafka主题

2.3 主题

2.3.1 管理

使用kafka-topics.sh脚本：

主题中可以使用的参数定义：

2.3.1.1 创建主题

kafka-topics.sh --zookeeper localhost:2181/myKafka --create --topic topic_x
 --partitions 1 --replication-factor 1
kafka-topics.sh --zookeeper localhost:2181/myKafka --create --topic topic_test_02
 --partitions 3 --replication-factor 1 --config max.message.bytes=1048576
 --config segment.bytes=10485760

2.3.1.2 查看主题

kafka-topics.sh --zookeeper localhost:2181/myKafka --list
kafka-topics.sh --zookeeper localhost:2181/myKafka --describe --topic topic_x
kafka-topics.sh --zookeeper localhost:2181/myKafka --topics-with-overrides -- describe

2.3.1.3 修改主题

kafka-topics.sh --zookeeper localhost:2181/myKafka --create --topic topic_test_01 
 --partitions 2 --replication-factor 1
kafka-topics.sh --zookeeper localhost:2181/myKafka --alter --topic topic_test_01
 --config max.message.bytes=1048576
kafka-topics.sh --zookeeper localhost:2181/myKafka --describe --topic topic_test_01
kafka-topics.sh --zookeeper localhost:2181/myKafka --alter --topic topic_test_01
 --config segment.bytes=10485760 
kafka-topics.sh --zookeeper localhost:2181/myKafka --alter --delete-config
 max.message.bytes --topic topic_test_01

2.3.1.4 删除主题

kafka-topics.sh --zookeeper localhost:2181/myKafka --delete --topic topic_x

给主题添加删除的标记：

要过一段时间删除。

2.3.2 增加分区

通过命令行工具操作，主题的分区只能增加，不能减少。否则报错：

ERROR org.apache.kafka.common.errors.InvalidPartitionsException:
The number of partitions for a topic can only be increased.
Topic myTop1 currently has 2 partitions, 1 would not be an increase.

通过—alter修改主题的分区数，增加分区。

kafka-topics.sh --zookeeper localhost/myKafka --alter --topic myTop1 --partitions 2

2.3.3 分区副本的分配-了解

副本分配的三个目标：

1. 均衡地将副本分散于各个broker上
2. 对于某个broker上分配的分区，它的其他副本在其他broker上
3. 如果所有的broker都有机架信息，尽量将分区的各个副本分配到不同机架上的broker。

在不考虑机架信息的情况下：

1. 第一个副本分区通过轮询的方式挑选一个broker，进行分配。该轮询从broker列表的随机位置

进行轮询。

1. 其余副本通过增加偏移进行分配。

分配案例：
broker-0 broker-1 broker-2 broker-3 broker-4
p0 p1 p2 p3 p4 (1st replica)
p5 p6 p7 p8 p9 (1st replica)
p4 p0 p1 p2 p3 (2nd replica)
p8 p9 p5 p6 p7 (2nd replica)
p3 p4 p0 p1 p2 (3nd replica)
p7 p8 p9 p5 p6 (3nd replica)

考虑到机架信息，首先为每个机架创建一个broker列表。如：
三个机架（rack1，rack2，rack3），六个broker（0，1，2，3，4，5）
brokerID -> rack

• 0 -> “rack1”, 1 -> “rack3”, 2 -> “rack3”, 3 -> “rack2”, 4 -> “rack2”, 5 -> “rack1”

rack1：0，5
rack2：3，4
rack3：1，2
这broker列表为rack1的0，rack2的3，rack3的1，rack1的5，rack2的4，rack3的2
即：0, 3, 1, 5, 4, 2
通过简单的轮询将分区分配给不同机架上的broker：

每个分区副本在分配的时候在上一个分区第一个副本开始分配的位置右移一位。

六个broker，六个分区，正好最后一个分区的第一个副本分配的位置是该broker列表的最后一个。
如果有更多的分区需要分配，则算法开始对follower副本进行移位分配。
这主要是为了避免每次都得到相同的分配序列。
此时，如果有一个分区等待分配（分区6），这按照如下方式分配：
6 -> 0,4,2 (而不是像分区0那样重复0,3,1)
跟机架相关的副本分配中，永远在机架相关的broker列表中轮询地分配第一个副本。其余的副本，倾向于机架上没有副本的broker进行副本分配，除非每个机架有一个副本。然后其他的副本又通过轮询的方式分配给broker。
结果是，如果副本的个数大于等于机架数，保证每个机架最少有一个副本。否则每个机架最多保有一个副本。
如果副本的个数和机架的个数相同，并且每个机架包含相同个数的broker，可以保证副本在机架和broker之间均匀分布。

上图，tp_eagle_01主题的分区0分配信息：leader分区在broker1上，同步副本分区是1和2，也就是在broker1和broker2上的两个副本分区是同步副本分区，其中一个是leader分区。

2.3.4 必要参数配置

kafka-topics.sh —config xx=xx —config yy=yy
配置给主题的参数。

2.3.5 KafkaAdminClient应用

说明
除了使用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。
功能与原理介绍
Kafka官网：The AdminClient API supports managing and inspecting topics, brokers, acls, and other Kafka objects。
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)

其内部原理是使用Kafka自定义的一套二进制协议来实现，详细可以参见Kafka协议。

用到的参数：

主要操作步骤：
客户端根据方法的调用创建相应的协议请求，比如创建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个抽象方法。 ```java package com.lagou.kafka.demo;

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 org.junit.After; import org.junit.Before; import org.junit.Test;

import java.util.*; 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", "node1: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("adm_tp_01", 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("adm_tp_01"), options);
    deleteResult.all().get();
}

@Test
public void testAlterTopic() throws ExecutionException, InterruptedException {
    NewPartitions newPartitions = NewPartitions.increaseTo(5);
    Map<String, NewPartitions> newPartitionsMap = new HashMap<>();
    newPartitionsMap.put("adm_tp_01", 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("adm_tp_01"), 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, "adm_tp_01");
    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); 
                    // } 
                    // });
                }
            });
        }
    });
}

}

<a name="9lXTm"></a>
### 2.3.6 偏移量管理
Kafka 1.0.2，__consumer_offsets主题中保存各个消费组的偏移量。<br />早期由zookeeper管理消费组的偏移量。<br />**查询方法：**<br />通过原生 kafka 提供的工具脚本进行查询。<br />工具脚本的位置与名称为 bin/kafka-consumer-groups.sh<br />首先运行脚本，查看帮助：

| **参数 ** | **说明** |
| --- | --- |
| **--all-topics** | 将所有关联到指定消费组的主题都划归到 reset-offsets 操作范围。 |
| **--bootstrap-server**<br /> <String: server to connect to> | **必须**：(基于消费组的新的消费者): 要连接的服务器地址。 |
| **--by-duration** <String: duration> | 距离当前时间戳的一个时间段。格式：'PnDTnHnMnS' |
| **--command-config** <br /><String:command config property file> | 指定配置文件，该文件内容传递给Admin Client和消费者。 |
| **--delete** | 传值消费组名称，删除整个消费组与所有主题的各个分区偏移量和所有者关系。<br />如： --group g1 --group g2 。<br />传值消费组名称和单个主题，仅删除该消费组到指定主题的分区偏移量和所属关系。<br />如： --group g1 --group g2 --topic t1 。<br />传值一个主题名称，仅删除指定主题与所有消费组分区偏移量以及所属关系。<br />如： --topic t1<br />注意：消费组的删除仅对基于ZK保存偏移量的消费组有效，并且要小心使用，仅删除不活跃的消费组。 |
| **--describe ** | 描述给定消费组的偏移量差距（有多少消息还没有消费） |
| **--execute ** | 执行操作。支持的操作： reset-offsets 。  |
| **--export** | 导出操作的结果到CSV文件。支持的操作： reset-offsets 。  |
| **--from-file**<br /> <String: path to CSV file> | 重置偏移量到CSV文件中定义的值。 |
| **--group** <br /><String: consumer group> | 目标消费组。 |
| **--list**  | 列出所有消费组。 |
| **--new-consumer** | 使用新的消费者实现。这是默认值。随后的发行版中会删除这一操作。 |
| **--reset-offsets** | 重置消费组的偏移量。当前一次操作只支持一个消费组，并且该消费组应该是不活跃的。<br />有三个操作选项<br />1. （默认）plan：要重置哪个偏移量。<br />2. execute：执行 reset-offsets 操作。<br />3. process：配合 --export 将操作结果导出到CSV格式。<br />可以使用如下选项：<br />--to-datetime <br />--by-period <br />--to-earliest <br />--to-latest <br />--shift-by <br />--from-file <br />--to-current 。<br />必须选择一个选项使用。<br />要定义操作的范围，使用：<br />--all-topics <br />--topic 。<br />必须选择一个，除非使用 --from-file 选项。 |
| **--shift-by**<br /> <Long: number-of-offsets> | 重置偏移量n个。n可以是正值，也可以是负值。 |
| **--timeout** <br /><Long: timeout (ms)> | 对某些操作设置超时时间。<br />如：对于描述指定消费组信息，指定毫秒值的最大等待时间，以获取正常数据（如刚创建的消费组，或者消费组做了一些更改操作）。<br />默认时间： 5000 。 |
| **--to-current ** | 重置到当前的偏移量。 |
| **--to-datetime**<br /><String:datetime> | 重置偏移量到指定的时间戳。格式：'YYYY-MM<br />DDTHH:mm:SS.sss |
| **--to-earliest** | 重置为最早的偏移量 |
| **--to-latest ** | 重置为最新的偏移量 |
| **--to-offset** <br /><Long:offset> | 重置到指定的偏移量。 |
| **--topic **<String: topic> | 指定哪个主题的消费组需要删除，或者指定哪个主题的消费组需要包含到 reset-offsets 操作中。对于 reset-offsets 操作，还可以指定分区： topic1:0,1,2 。其中0，1，2表示要包含到操作中的分区号。重置偏移量的操作支持多个主题一起操作。 |
| **--zookeeper** <String: urls>  | **必须**，它的值，你懂的。 --zookeeper node1:2181/myKafka 。 |

这里我们先编写一个生产者，消费者的例子：<br />我们先启动消费者，再启动生产者， 再通过 bin/kafka-consumer-groups.sh 进行消费偏移量查询，

由于kafka 消费者记录group的消费偏移量有两种方式 ： <br />1）kafka 自维护 （新）<br />2）zookpeer 维护 (旧) ，已经逐渐被废弃<br />所以 ，脚本只查看由broker维护的，由zookeeper维护的可以将** --bootstrap-server **换成 **-- zookeeper **即可。<br /> <br />1**. ****查看有那些**** group ID ****正在进行消费：**
```java
[root@node11 ~]# kafka-consumer-groups.sh --bootstrap-server node1:9092 --list
Note: This will not show information about old Zookeeper-based consumers. 

group

注意：

1. 这里面是没有指定** topic，查看的是所有topic消费者的 group.id **的列表。
2. 注意： 重名的** group.id **只会显示一次

2.**查看指定group.id 的消费者消费情况**

[root@node11 ~]# kafka-consumer-groups.sh --bootstrap-server node1:9092 --describe
 --group group