目录
- 主题的管理
- 创建主题
- 查看主题
- 修改主题
- 删除主题
- 增加分区
- 分区副本的分配
- 必要参数配置
- KafkaAdminClient应用
- 功能
- 操作示例
主题的管理
使用kafka-topics.sh脚本。
下面是使用脚本的一些选项
| 选项 | 说明 |
|---|---|
| –config <String: name=value> | 为创建的或修改的主题指定配置信息。 |
| –create | 创建一个新主题 |
| –delete | 删除一个主题 |
| –delete-config <String: name> | 删除现有主题的一个主题配置条目。这些条目就 是在–config中给出的配置条目。 |
| –alter | 更改主题的分区数量，副本分配和/或配置条目。 |
| –describe | 列出给定主题的细节。 |
| –disable-rack-aware | 禁用副本分配的机架感知。 |
| –force | 抑制控制台提示信息 |
| –help | 打印帮助信息 |
| –replica-assignment <String:broker_id_for_part1_replica1 :broker_id_for_part1_replica2 ,broker_id_for_part2_replica1 :broker_id_for_part2_replica2 , …> | 当创建或修改主题的时候手动指定partition-to-broker的分配关系。 |
| –replication-factor <Integer:replication factor> | 要创建的主题分区副本数。1表示只有一个副本， 也就是Leader副本。 |
| –topic <String: topic> | 要创建、修改或描述的主题名称。除了创建，修改和描述在这里还可以使用正则表达式。 |
| –topics-with-overrides | if set when describing topics, only show topics that have overridden configs |
| –unavailable-partitions | if set when describing topics, only show partitions whose leader is not available |
| –under-replicated-partitions | if set when describing topics, only show under replicated partitions |
| –zookeeper <String: urls> | 必需的参数：连接zookeeper的字符串，逗号分隔的多个host:port列表。多个URL可以故障转移。 |
创建主题
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
查看主题
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
修改主题
kafka-topics.sh --zookeeper localhost:2181/myKafka --create --topic topic_test_01 --partitions 2 --replication-factor 1kafka-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_01kafka-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
删除主题
kafka-topics.sh --zookeeper localhost:2181/myKafka --delete --topic topic_x
删除主题时，只是给主题添加删除的标记，要过一段时间删除。
增加分区
通过命令行工具操作，主题的分区只能增加，不能减少。否则报错
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
分区副本的分配
副本分配的三个目标：
- 均衡地将副本分散于各个broker上
- 对于某个broker上分配的分区，它的其他副本在其他broker上
- 如果所有的broker都有机架信息，尽量将分区的各个副本分配到不同机架上的broker。
在不考虑机架信息的情况下：
- 第一个副本分区通过轮询的方式挑选一个broker，进行分配。该轮询从broker列表的随机位置进行轮询。
- 其余副本通过增加偏移进行分配。
必要参数配置
使用kafka-topics.sh --config xx=xx --config yy=yy
配置给主题的参数如下：
| 属性 | 默认值 | 服务器默认属性 | 说明 |
|---|---|---|---|
| cleanup.policy | delete | log.cleanup.policy | 要么是”delete“要么是”compact“； 这个字符串指明了当他们的回收时间或者尺寸限制到达时，针对旧日志部分的利用方式。默认 方式（“delete”）将会丢弃旧的部分；”compact“将会进行日志压缩。 |
| compression.type | none | producer用于压缩数据的压缩类 型。默认是无压缩。正确的选项 值是none、gzip、snappy、 lz4。压缩最好用于批量处理，批 量处理消息越多，压缩性能越 好。 | |
| max.message.bytes | 1000000 | max.message.bytes | kafka追加消息的最大字节数。注 意如果你增大这个字节数，也必 须增大consumer的fetch字节 数，这样consumer才能fetch到 这些最大字节数的消息。 |
| min.cleanable.dirty.ratio | 0.5 | min.cleanable.dirty.ratio | 此项配置控制log压缩器试图进行 清除日志的频率。默认情况下， 将避免清除压缩率超过50%的日 志。这个比率避免了最大的空间 浪费 |
| min.insync.replicas | 1 | min.insync.replicas | 当producer设置 request.required.acks为-1时， min.insync.replicas指定replicas 的最小数目（必须确认每一个 repica的写数据都是成功的）， 如果这个数目没有达到， producer会产生异常。 |
| retention.bytes | None | log.retention.bytes | 如果使用“delete”的retention 策 略，这项配置就是指在删除日志之前，日志所能达到的最大尺寸。默认情况下，没有尺寸限制而只有时间限制 |
| retention.ms | 7 days | log.retention.minutes | 如果使用“delete”的retention策 略，这项配置就是指删除日志前 日志保存的时间。 |
| segment.bytes | 1GB | log.segment.bytes | kafka中log日志是分成一块块存储的，此配置是指log |
| segment.index.bytes | 10MB | log.index.size.max.bytes | 此配置是有关offsets和文件位置 之间映射的索引文件的大小；一 般不需要修改这个配置 |
| segment.jitter.ms | 0 | log.roll.jitter.{ms,hours} | The maximum jitter to subtract from logRollTimeMillis. |
| segment.ms | 7 days | log.roll.hours | 即使log的分块文件没有达到需要删除、压缩的大小，一旦log 的时间达到这个上限，就会强制新建一个log分块文件 |
| unclean.leader.election.enable | true | 指明了是否能够使不在ISR中的replicas设置用来作为leader |
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。
功能
KafkaAdminClient包含了一下几种功能（以Kafka1.0.2版本为准）：
-
创建主题：
createTopics(final Collection<NewTopic> newTopics, final CreateTopicsOptions options) -
删除主题：
deleteTopics(final Collection<String> topicNames, DeleteTopicsOptions options) -
列出所有主题：
listTopics(final ListTopicsOptions options) -
查询主题：
describeTopics(final Collection<String> topicNames, DescribeTopicsOptions options) -
查询集群信息：
describeCluster(DescribeClusterOptions options) -
查询配置信息：
describeConfigs(Collection<ConfigResource> configResources, final DescribeConfigsOptions options) -
修改配置信息：
alterConfigs(Map<ConfigResource, Config> configs, final AlterConfigsOptions options) -
修改副本的日志目录：
alterReplicaLogDirs(Map<TopicPartitionReplica, String> replicaAssignment, final AlterReplicaLogDirsOptions options) -
查询节点的日志目录信息：
describeLogDirs(Collection<Integer> brokers, DescribeLogDirsOptions options) -
查询副本的日志目录信息：
describeReplicaLogDirs(Collection<TopicPartitionReplica> replicas,DescribeReplicaLogDirsOptions options) -
增加分区：
createPartitions(Map<String, NewPartitions> newPartitions, final CreatePartitionsOptions options)
其内部原理是使用Kafka自定义的一套二进制协议来实现，详细可以参见Kafka协议。
用到的参数：
| 属性 | 说明 | 重要性 |
|---|---|---|
| bootstrap.servers | 向Kafka集群建立初始连接用到的host/port列表。客户端会使用这里列出的所有服务器进行集群其他服务器的发现，而不管是否指定了哪个服务器用作引导。这个列表仅影响用来发现集群所有服务器的初始主机。 字符串形式：host1:port1,host2:port2,… 由于这组服务器仅用于建立初始链接，然后发现集群中的所有服务器，因此没有必要将集群中的所有地址写在这里。 一般最好两台，以防其中一台宕掉。 | high |
| client.id | 生产者发送请求的时候传递给broker的id字符串。用于在broker的请求日志中追踪什么应用发送了什么消息。一般该id是跟业务有关的字符串。 | medium |
| connections.max.idle.ms | 当连接空闲时间达到这个值，就关闭连接。long型数据，默认：300000 | medium |
| request.timeout.ms | 客户端等待服务端响应的最大时间。如果该时间超时，则客户端要么重新发起请求，要么如果重试耗尽，请求失败。int类型值，默认：120000 | medium |
| security.protocol | 跟broker通信的协议：PLAINTEXT, SSL,SASL_PLAINTEXT, SASL_SSL.string类型值，默认：PLAINTEXT | medium |
| reconnect.backoff.ms | 重新连接主机的等待时间。避免了重连的密集循环。该等待时间应用于该客户端到broker的所有连接。long型值，默认：50 | medium |
| retries | 重试的次数，达到此值，失败。int类型值，默认5。 | low |
| retry.backoff.ms | 在发生失败的时候如果需要重试，则该配置表示客户端等待多长时间再发起重试。该时间的存在避免了密集循环。long型值，默认值：100。 | low |
| receive.buffer.bytes | TCP接收缓存（SO_RCVBUF），如果设置为-1，则使用操作系统默认的值。int类型值，默认65536 | medium |
| send.buffer.bytes | 用于TCP发送数据时使用的缓冲大小（SO_SNDBUF），-1表示使用OS默认的缓冲区大小。int类型值，默认值：131072 | medium |
| reconnect.backoff.max.ms | 对于每个连续的连接失败，每台主机的退避将成倍增加，直至达到此最大值。在计算退避增量之后，添加20％的随机抖动以避免连接风暴。 long型值，默认1000 | low |
操作示例
主要操作步骤：
- 客户端根据方法的调用创建相应的协议请求，比如创建Topic的createTopics方法，其内部就是发送CreateTopicRequest请求。
- 客户端发送请求至Kafka Broker。
- Kafka Broker处理相应的请求并回执，比如与CreateTopicRequest对应的是CreateTopicResponse。
- 客户端接收相应的回执并进行解析处理。
和协议有关的请求和回执的类基本都在org.apache.kafka.common.requests包中，AbstractRequest和AbstractResponse是这些请求和响应类的两个父类。
综上，如果要自定义实现一个功能，只需要三个步骤：
- 自定义XXXOptions;
- 自定义XXXResult返回值；
- 自定义Call，然后挑选合适的XXXRequest和XXXResponse来实现Call类中的3个抽象方法。
示例：
import org.apache.kafka.clients.admin.*;
import org.apache.kafka.common.TopicPartition;
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.function.BiConsumer;
import java.util.function.Consumer;public class MyAdminClient {private KafkaAdminClient client;@Beforepublic void before() {Map<String, Object> configs = new HashMap<>();configs.put("bootstrap.servers", "node1:9092");configs.put("client.id", "admin_001");client = (KafkaAdminClient) KafkaAdminClient.create(configs);}@Afterpublic void after() {// 关闭admin客户端client.close();}@Testpublic void testListTopics() throws ExecutionException, InterruptedException {// 列出主题
// final ListTopicsResult listTopicsResult = client.listTopics();ListTopicsOptions options = new ListTopicsOptions();// 列出内部主题options.listInternal(true);// 设置请求超时时间，单位是毫秒options.timeoutMs(500);final ListTopicsResult listTopicsResult = client.listTopics(options);// final Set<String> strings = listTopicsResult.names().get();
//
// strings.forEach(name -> {
// System.out.println(name);
// });// 将请求变成同步的请求，直接获取结果final Collection<TopicListing> topicListings = listTopicsResult.listings().get();topicListings.forEach(new Consumer<TopicListing>() {@Overridepublic void accept(TopicListing topicListing) {// 该主题是否是内部主题final boolean internal = topicListing.isInternal();// 该主题的名字final String name = topicListing.name();System.out.println("主题是否是内部主题：" + internal);System.out.println("主题的名字：" + name);System.out.println(topicListing);System.out.println("=====================================");}});}@Testpublic void testDescribeLogDirs() throws ExecutionException, InterruptedException {final DescribeLogDirsResult describeLogDirsResult = client.describeLogDirs(Collections.singleton(0));final Map<Integer, Map<String, DescribeLogDirsResponse.LogDirInfo>> integerMapMap= describeLogDirsResult.all().get();integerMapMap.forEach(new BiConsumer<Integer, Map<String, DescribeLogDirsResponse.LogDirInfo>>() {@Overridepublic void accept(Integer integer, Map<String, DescribeLogDirsResponse.LogDirInfo> stringLogDirInfoMap) {System.out.println("broker.id = " + integer);
// log.dirs可以设置多个目录stringLogDirInfoMap.forEach(new BiConsumer<String, DescribeLogDirsResponse.LogDirInfo>() {@Overridepublic void accept(String s, DescribeLogDirsResponse.LogDirInfo logDirInfo) {System.out.println("logdir = " + s);final Map<TopicPartition, DescribeLogDirsResponse.ReplicaInfo> replicaInfos = logDirInfo.replicaInfos;replicaInfos.forEach(new BiConsumer<TopicPartition, DescribeLogDirsResponse.ReplicaInfo>() {@Overridepublic void accept(TopicPartition topicPartition, DescribeLogDirsResponse.ReplicaInfo replicaInfo) {System.out.println("主题分区：" + topicPartition.partition());System.out.println("主题：" + topicPartition.topic());
// final boolean isFuture = replicaInfo.isFuture;
// final long offsetLag = replicaInfo.offsetLag;
// final long size = replicaInfo.size;}});}});}});}
}
