Kafka（六）Kafka基本客户端命令操作

主题管理

创建主题

如果配置了auto.create.topics.enable=true（这也是默认值）这样当生产者向一个没有创建的主题发送消息就会自动创建，其分区数量和副本数量也是有默认配置来控制的。

# 我们这里创建一个3个分区每个分区有2个副本的主题
kafka-topics.sh --create --zookeeper 172.16.48.171:2181/kafka --replication-factor 2 --partitions 3 --topic KafkaTest

在Zookeeper中可以看到如下信息

删除主题

删除有两种方式手动和自动

• 手动方式需要删除各个节点日志路径下的该主题所有分区，并且删除zookeeper上/brokers/topics和/config/topics下的对应主题节点
• 自动删除就是通过脚本来完成，同时需要配置服务器配置文件中的delete.topic.enable=true，默认为false也就是说通过命令删除主题只会删除ZK中的节点，日志文件不会删除需要手动清理，如果配置为true，则会自动删除日志文件。

kafka-topics.sh --delete --zookeeper 172.16.48.171:2181/kafka --topic KafkaTest

下面的两句话就是说该主题标记为删除/admin/delete_topics节点下。实际数据没有影响因为该参数没有设置为true。

查看主题

# 列出所有主题
kafka-topics.sh --list --zookeeper 172.16.48.171:2181/kafka

下面是从ZK中看到的所有主题

# 查看所有主题信息
kafka-topics.sh --describe --zookeeper 172.16.48.171:2181/kafka

# 查看特定主题信息
kafka-topics.sh --describe --zookeeper 172.16.48.171:2181/kafka --topic BBB

Replicas：是AR列表，表示副本分布在哪些代理上，且该列表第一个元素就是Leader副本所在代理

ISR：该列表是显示已经同步的副本集合，这个列表的副本都是存活的

# 通过--describe 和 --under-replicated-partitions 可以查看正在同步的主题或者同步可能发生异常，
# 也就是ISR列表长度小于AR列表，如果一切正常则不会返回任何东西，也可以通过 --tipic 指定具体主题
kafka-topics.sh --describe --zookeeper 172.16.48.171:2181/kafka --under-replicated-partitions

# 查看哪些主题建立时使用了单独的配置 
kafka-topics.sh --describe --zookeeper 172.16.48.171:2181/kafka --topics-with-overrides

这里只有一个内部主题__comsumer_offsets使用了非配置文件中的设置

配置管理

所谓配置就是参数，比如修改主题的默认参数。

主题级别的

# 查看配置
kafka-configs.sh --describe --zookeeper 172.16.48.171:2181/kafka --entity-type topics --entity-name BB

这里显示 Configs for topic 'BBB' are 表示它的配置有哪些，这里没有表示没有为该主题单独设置配置，都是使用的默认配置。

# 增加一个配置
kafka-configs.sh --zookeeper 172.16.48.171:2181/kafka --entity-type topics --entity-name BBB --alter --add-config flush.messages=2

如果修改的话还是相同的命令，只是把值修改一下

# 删除配置
kafka-configs.sh --zookeeper 172.16.48.171:2181/kafka --entity-type topics --entity-name BBB --alter --delete-config flush.messages

客户端级别

这个主要是设置流控

# 设置指定消费者的流控 --entity-name 是客户端在创建生产者或者消费者时是指定的client.id名称
kafka-configs.sh --zookeeper 172.16.48.171:2181/kafka --alter --add-config 'producer_byte_rate=1024,consumer_byte_rate=1024' --entity-type clients --entity-name COMSUMER_NAME

下图为ZK中对应的信息

分区管理

分区平衡

Leader副本在集群中应该是均衡分布，因为Leader副本对外提供读写服务，尽可能不让同一个主题的多个Leader副本在同一个代理上，但是随着时间推移比如故障转移等情况发送，Leader副本可能不均衡。有两种方式设置自动平衡，自动和手动。

自动就是在配置文件中增加 auto.leader.rebalance.enable = true 如果该项为false，当某个节点故障恢复并重新上线后，它原来的Leader副本也不会转移回来，只是一个Follower副本。

手动就是通过命令来执行

kafka-preferred-replica-election.sh --zookeeper 172.16.48.171:2181/kafka

分区迁移

当下线一个节点需要将该节点上的分区副本迁移到其他可用节点上，Kafka并不会自动进行分区迁移，如果不迁移就会导致某些主题数据丢失和不可用的情况。当增加新节点时，只有新创建的主题才会分配到新节点上，之前的主题分区不会自动分配到新节点上，因为老的分区在创建时AR列表中没有这个新节点。

上面2个主题，每个主题3个分区，每个分区3个副本，我们假设现在代理2要下线，所以我们要把代理2上的这两个主题的分区数据迁移出来。

# 1. 在KAFKA目录的config目录中建立topics-to-move.json文件
{
    "topics":[
        {
            "topic":"AAA"
        },
        {
            "topic":"BBB"
        }
    ],
    "version":1
}

# 2. 生成分区分配方案，只是生成一个方案信息然后输出
kafka-reassign-partitions.sh --zookeeper 172.16.48.171:2181/kafka --topics-to-move-json-file ./topics-to-move.json --broker-list "1,2" --generate

这个命令的原理是从zookeeper中读取主题元数据信息及制定的有效代理，根据分区副本分配算法重新计算指定主题的分区副本分配方案。把【Proposed partition reassignment configuration】下面的分区方案保存到一个JSON文件中，partitions-reassignment.json 文件名无所谓。

# 3. 执行方案
kafka-reassign-partitions.sh --zookeeper 172.16.48.171:2181/kafka --reassignment-json-file ./partitions-reassignment.json --execute

# 4. 查看进度
kafka-reassign-partitions.sh --zookeeper 172.16.48.171:2181/kafka --reassignment-json-file ./partitions-reassignment.json --verify

查看结果，这里已经没有代理0了。

集群扩容

上面演示了节点下线的数据迁移，这里演示一下集群扩容的数据迁移。我们还是用上面两个主题，假设代理0又重新上线了。其实扩容就是上面的反向操作

# 1. 建立JSON文件
# 该文件和之前的相同

# 2. 生成方案并保存到一个JSON文件中
kafka-reassign-partitions.sh --zookeeper 172.16.48.171:2181/kafka --topics-to-move-json-file ./topics-to-move.json --broker-list "0,1,2" --generate

# 3. 数据迁移，这里通过--throttle做一个限流操作，如果数据过大会把网络堵塞。
kafka-reassign-partitions.sh --zookeeper 172.16.48.171:2181/kafka --reassignment-json-file ./partitions-reassignment.json --execute --throttle 1024

查看进度和结果

增加副本

原文地址：http://blog.51cto.com/littledevil/2147297