Kafka 主题解析

一、主题管理

使?kafka-topics.sh脚本时可用的配置：

主题中可以使?的参数定义（也就是上面--config 的参数）：

1.1 创建主题

kafka-topics.sh --zookeeper localhost:2181/myKafka --create --topic tp_demo_02 --partitions 2 --replication-factor 1

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

1.2 查看主题

kafka-topics.sh --zookeeper localhost:2181/myKafka --list

kafka-topics.sh --zookeeper localhost:2181/myKafka --describe --topic tp_demo_02

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

1.3 修改主题

kafka-topics.sh --zookeeper localhost:2181/myKafka --create --topic tp_demo_04 --partitions 4 --replication-factor 1

kafka-topics.sh --zookeeper localhost:2181/myKafka --alter --topic tp_demo_04 --config max.message.bytes=1048576

kafka-topics.sh --zookeeper localhost:2181/myKafka --describe --topic tp_demo_04

kafka-topics.sh --zookeeper localhost:2181/myKafka --alter --topic tp_demo_04 --config segment.bytes=10485760

kafka-topics.sh --zookeeper localhost:2181/myKafka --alter --delete-config max.message.bytes --topic tp_demo_04

1.4 删除主题

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

二、增加分区

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

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

kafka-topics.sh --zookeeper localhost:2181/myKafka --alter --topic tp_demo_02 --partitions 3

三、分区副本的分配

副本分配的三个?标：

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

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

1. 第?个副本分区通过轮询的?式挑选?个broker，进?分配。该轮询从broker列表的随机位置进?轮询。
2. 其余副本通过增加偏移进?分配。

分配案例：

考虑到机架信息，?先为每个机架创建?个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之间均匀分布。

四、常用参数配置

上面已经介绍过：kafka-topics.sh --config xx=xx --config yy=yy

配置给主题的参数。

五、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版本为准）：

• 创建主题

  createTopics(final Collection newTopics, final CreateTopicsOptions options)

• 删除主题

  deleteTopics(final Collection topicNames, DeleteTopicsOptions options)

• 列出所有主题

  listTopics(final ListTopicsOptions options)

• 查询主题

  describeTopics(final Collection topicNames, DescribeTopicsOptions options)

• 查询集群信息

  describeCluster(DescribeClusterOptions options)

• 查询配置信息

  describeConfigs(Collection configResources, final DescribeConfigsOptions options)

• 修改配置信息

  alterConfigs(Map configs, final AlterConfigsOptions options)

• 修改副本的日志记录

  alterReplicaLogDirs(Map replicaAssignment, final AlterReplicaLogDirsOptions options)

• 查询节点的?志?录信息

  describeLogDirs(Collection brokers, DescribeLogDirsOptions options)

• 查询副本的?志?录信息

  describeReplicaLogDirs(Collection replicas, DescribeReplicaLogDirsOptions options)

• 增加分区

  createPartitions(Map newPartitions, final CreatePartitionsOptions options)

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

KafkaAdminClient?到的参数：

主要操作步骤：

客户端根据?法的调?创建相应的协议请求，?如创建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个抽象?法。

六、偏移量管理

Kafka 1.0.2，__consumer_offsets主题中保存各个消费组的偏移量。

早期由zookeeper管理消费组的偏移量。

查询?法：

通过原? kafka 提供的?具脚本进?查询。

?具脚本的位置与名称为bin/kafka-consumer-groups.sh

?先运?脚本，查看帮助：

由于kafka 消费者记录group的消费偏移量有两种?式 ：

1. kafka ?维护 （新）
2. zookpeer 维护 (旧) ，已经逐渐被废弃

所以 ，脚本只查看由broker维护的，由zookeeper维护的可以将--bootstrap-server换成--zookeeper即可。

1. 查看有那些 group ID 正在进?消费

   kafka-consumer-groups.sh --bootstrap-server localhost:9092 --list
   

   注意：

   • 这??是没有指定 topic，查看的是所有topic消费者的 group.id 的列表。

   • 注意： 重名的 group.id 只会显示?次

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

   kafka-consumer-groups.sh --bootstrap-server node1:9092 --describe --group group
   

   如果消费者停?，查看偏移量信息：
   
   将偏移量设置为最早的：
   
   将偏移量设置为最新的：
   
   分别将指定主题的指定分区的偏移量向前移动10个消息：