Kafka常用命令收录

逃离我推掉我的手 2022-05-13 01:52 188阅读 0赞

# **目录** #

目录 1

1. 前言 2

2. Broker默认端口号 2

3. 安装Kafka 2

4. 启动Kafka 2

5. 创建Topic 2

6. 列出所有Topic 3

7. 删除Topic 3

8. 查看Topic 3

9. 增加topic的partition数 4

10. 生产消息 4

11. 消费消息 4

12. 查看有哪些消费者Group 4

13. 查看新消费者详情 5

14. 查看Group详情 5

15. 删除Group 5

16. 设置consumer group的offset 5

17. RdKafka自带示例 6

18. 平衡leader 6

19. 自带压测工具 6

20. 查看topic指定分区offset的最大值或最小值 6

21. 查看\_\_consumer\_offsets 6

22. 获取指定consumer group的位移信息 7

23. 20) 查看kafka的zookeeper 7

24. 如何增加\_\_consumer\_offsets的副本数？ 9

25. 问题 11

附1：进程监控工具process\_monitor.sh 12

附2：批量操作工具 12

附2.1：批量执行命令工具：mooon\_ssh 12

附2.2：批量上传文件工具：mooon\_upload 13

附2.3：使用示例 13

附3：批量设置broker.id和listeners工具 15

附4：批量设置hostname工具 16

附5：Kafka监控工具kafka-manager 16

附6：kafka的安装 16

附7:\_\_consumer\_offsets 17

# **1. 前言** #

本文内容主要来自两个方面：一是网上的分享，二是自研的随手记。日记月累，收录kafka各种命令，会持续更新。

在0.9.0.0之后的Kafka，出现了几个新变动，一个是在Server端增加了GroupCoordinator这个角色，另一个较大的变动是将topic的offset 信息由之前存储在zookeeper上改为存储到一个特殊的topic（\_\_consumer\_offsets）中。

Kafka的瓶颈容易发生在网卡，而不是CPU、内存和磁盘，所以应当考虑log的压缩。

相关网址：

1) Kafka官网：[http://kafka.apache.org/][http_kafka.apache.org]

2) 下载地址：[http://kafka.apache.org/downloads][http_kafka.apache.org_downloads]

3) 客户端库：[https://cwiki.apache.org/confluence/display/KAFKA/Clients][https_cwiki.apache.org_confluence_display_KAFKA_Clients]

4) librdkafka库：[https://github.com/edenhill/librdkafka][https_github.com_edenhill_librdkafka]

5) confluent-kafka-go：[https://github.com/confluentinc/confluent-kafka-go][https_github.com_confluentinc_confluent-kafka-go]

# **2. Broker默认端口号** #

9092，建议安装时，在zookeeper中指定kafka的根目录，比如“/kafka”，而不是直接使用“/”，这样多套kafka也可共享同一个zookeeper集群。

# **3. 安装Kafka** #

相比Hadoop、HBase、Spark，甚至Redis等，Kafka的安装到跑起来比较简单，参考官方的介绍即可：[http://kafka.apache.org/quickstart][http_kafka.apache.org_quickstart]。

Kafka依赖Zookeeper，本身自带了Zookeeper，不过建议另外安装Zookeeper。

# **4. 启动Kafka** #

kafka-server-start.sh config/server.properties

后台常驻方式，请带上参数“-daemon”，如：

<table style="width:426.1pt;"> 
 <tbody> 
  <tr> 
   <td style="vertical-align:top;width:426.1pt;"> <p style="margin-left:0pt;">/usr/local/kafka/bin/kafka-server-start.sh&nbsp;-daemon&nbsp;/usr/local/kafka/config/server.properties</p> </td> 
  </tr> 
 </tbody> 
</table>

# **5. 创建Topic** #

参数--topic指定Topic名，--partitions指定分区数，--replication-factor指定备份数：

<table style="width:426.1pt;"> 
 <tbody> 
  <tr> 
   <td style="vertical-align:top;width:426.1pt;"> <p style="margin-left:0pt;">kafka-topics.sh&nbsp;--create&nbsp;--zookeeper&nbsp;localhost:2181&nbsp;--replication-factor&nbsp;1&nbsp;--partitions&nbsp;1&nbsp;--topic&nbsp;test</p> </td> 
  </tr> 
 </tbody> 
</table>

注意，如果配置文件server.properties指定了kafka在zookeeper上的目录，则参数也要指定，否则会报无可用的brokers，如：

<table style="width:426.1pt;"> 
 <tbody> 
  <tr> 
   <td style="vertical-align:top;width:426.1pt;"> <p style="margin-left:0pt;">kafka-topics.sh&nbsp;--create&nbsp;--zookeeper&nbsp;localhost:2181/kafka&nbsp;--replication-factor&nbsp;1&nbsp;--partitions&nbsp;1&nbsp;--topic&nbsp;test</p> </td> 
  </tr> 
 </tbody> 
</table>

# **6. 列出所有Topic** #

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

注意，如果配置文件server.properties指定了kafka在zookeeper上的目录，则参数也要指定，否则会报无可用的brokers，如：

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

输出示例：

<table style="width:426.1pt;"> 
 <tbody> 
  <tr> 
   <td style="vertical-align:top;width:426.1pt;"> <p style="margin-left:0pt;">__consumer_offsets</p> <p style="margin-left:0pt;">my-replicated-topic</p> <p style="margin-left:0pt;">test</p> </td> 
  </tr> 
 </tbody> 
</table>

# **7. 删除Topic** #

1) kafka-**topics**.sh --zookeeper localhost:2181 --topic test --**delete**

2) kafka-topics.sh --zookeeper localhost:2181/**kafka **\--topic test --delete

3) kafka-**run**\-class.sh kafka.admin.**DeleteTopicCommand **\--zookeeper localhost:2181 --topic test

# **8. 查看Topic** #

kafka-topics.sh --**describe **\--zookeeper localhost:2181 --topic test

注意，如果配置文件server.properties指定了kafka在zookeeper上的目录，则参数也要指定，否则会报无可用的brokers，如：

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

输出示例：

<table style="width:426.1pt;"> 
 <tbody> 
  <tr> 
   <td style="vertical-align:top;width:426.1pt;"> <p style="margin-left:0pt;">Topic:test&nbsp;PartitionCount:3&nbsp;ReplicationFactor:2&nbsp;Configs:</p> <p style="margin-left:0pt;">Topic:&nbsp;test&nbsp;Partition:&nbsp;0&nbsp;Leader:&nbsp;140&nbsp;Replicas:&nbsp;140,214&nbsp;Isr:&nbsp;140,214</p> <p style="margin-left:0pt;">Topic:&nbsp;test&nbsp;Partition:&nbsp;1&nbsp;Leader:&nbsp;214&nbsp;Replicas:&nbsp;214,215&nbsp;Isr:&nbsp;214,215</p> <p style="margin-left:0pt;">Topic:&nbsp;test&nbsp;Partition:&nbsp;2&nbsp;Leader:&nbsp;215&nbsp;Replicas:&nbsp;215,138&nbsp;Isr:&nbsp;215,138</p> </td> 
  </tr> 
 </tbody> 
</table>

# **9. 增加topic的partition数** #

kafka-topics.sh --zookeeper localhost:2181 --**alter **\--topic test --partitions 5

# **10. 生产消息** #

kafka-console-**producer**.sh --broker-**list **localhost:9092 --topic test

# **11. 消费消息** #

1) 从头开始

kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-**beginning**

2) 从尾部开始

kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --offset **latest**

3) 指定分区

kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --offset latest --**partition **1

4) 取指定个数

kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --offset latest --partition 1 --**max-messages** 1

5) 新消费者（ver>=0.9）

kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --**new-consumer** --from-beginning --consumer.config config/consumer.properties

# **12. 查看有哪些消费者Group** #

1) 分ZooKeeper方式（老）

kafka-consumer-groups.sh --zookeeper 127.0.0.1:2181/kafka --list

2) API方式（新）

kafka-consumer-groups.sh --new-consumer --bootstrap-server 127.0.0.1:9092 --list

输出示例：

<table style="width:426.1pt;"> 
 <tbody> 
  <tr> 
   <td style="vertical-align:top;width:426.1pt;"> <p style="margin-left:0pt;">test</p> <p style="margin-left:0pt;">console-consumer-37602</p> <p style="margin-left:0pt;">console-consumer-75637</p> <p style="margin-left:0pt;">console-consumer-59893</p> </td> 
  </tr> 
 </tbody> 
</table>

# **13. 查看新消费者详情** #

仅支持offset存储在zookeeper上的：

kafka-**run**\-class.sh kafka.tools.**ConsumerOffsetChecker** --zkconnect localhost:2181 --group test

# **14. 查看Group详情** #

kafka-consumer-groups.sh --new-consumer --bootstrap-server 127.0.0.1:9092 --group test --describe

输出示例：

<table style="width:426.1pt;"> 
 <tbody> 
  <tr> 
   <td style="vertical-align:top;width:426.1pt;"> <p style="margin-left:0pt;">TOPIC&nbsp;PARTITION&nbsp;CURRENT-OFFSET&nbsp;LOG-END-OFFSET&nbsp;LAG&nbsp;CONSUMER-ID&nbsp;HOST&nbsp;CLIENT-ID</p> <p style="margin-left:0pt;">test&nbsp;&nbsp;1&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;87&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;87&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;0&nbsp;&nbsp;&nbsp;-&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;-&nbsp;&nbsp;&nbsp;&nbsp;-</p> </td> 
  </tr> 
 </tbody> 
</table>

# **15. 删除Group** #

老版本的ZooKeeper方式可以删除Group，新版本则自动删除，当执行：

kafka-consumer-groups.sh --new-consumer --bootstrap-server 127.0.0.1:9092 --group test --**delete**

输出如下提示：

<table style="width:426.1pt;"> 
 <tbody> 
  <tr> 
   <td style="vertical-align:top;width:426.1pt;"> <p style="margin-left:0pt;">Option&nbsp;'[delete]'&nbsp;is&nbsp;only&nbsp;valid&nbsp;with&nbsp;'[zookeeper]'.</p> <p style="margin-left:0pt;">Note&nbsp;that&nbsp;there's&nbsp;no&nbsp;need&nbsp;to&nbsp;delete&nbsp;group&nbsp;metadata&nbsp;for&nbsp;the&nbsp;new&nbsp;consumer</p> <p style="margin-left:0pt;">as&nbsp;the&nbsp;group&nbsp;is&nbsp;deleted&nbsp;when&nbsp;the&nbsp;last&nbsp;committed&nbsp;offset&nbsp;for&nbsp;that&nbsp;group&nbsp;expires.</p> </td> 
  </tr> 
 </tbody> 
</table>

# **16. 设置consumer group的offset** #

执行zkCli.sh进入zookeeper命令行界面，假设需将group为testgroup的topic的offset设置为2018，则：set /consumers/testgroup/offsets/test/0 2018

如果kakfa在zookeeper中的根目录不是“/”，而是“/kafka”，则：

set /kafka/consumers/testgroup/offsets/test/0 2018

另外，还可以使用kafka自带工具kafka-run-class.sh kafka.tools.UpdateOffsetsInZK修改，命令用法：

kafka.tools.UpdateOffsetsInZK$ \[earliest | latest\] consumer.properties topic

从用法提示可以看出，只能修改为earliest或latest，没有直接修改zookeeper灵活。

# **17. RdKafka自带示例** #

rdkafka\_consumer\_example -b 127.0.0.1:9092 -g test test

rdkafka\_consumer\_example -e -b 127.0.0.1:9092 -g test test

# **18. 平衡leader** #

kafka-preferred-replica-election.sh --zookeeper localhost:2181/chroot

# **19. 自带压测工具** #

kafka-producer-**perf**\-test.sh --topic test --num-records 100 --record-size 1 --throughput 100 --producer-props bootstrap.servers=localhost:9092

# **20. 查看topic指定分区offset的最大值或最小值** #

time为-1时表示最大值，为-2时表示最小值：

kafka-run-class.sh kafka.tools.GetOffsetShell --topic hive-mdatabase-hostsltable --time -1 --broker-list 127.0.0.1:9092 --partitions 0

# **21. 查看\_\_consumer\_offsets** #

需consumer.properties中设置exclude.internal.topics=false：

1) 0.11.0.0之前版本

kafka-console-consumer.sh --topic \_\_consumer\_offsets --zookeeper localhost:2181 --formatter "kafka.coordinator.GroupMetadataManager\\$OffsetsMessageFormatter" --consumer.config config/consumer.properties --from-beginning

2) 0.11.0.0之后版本(含)

kafka-console-consumer.sh --topic \_\_consumer\_offsets --zookeeper localhost:2181 --formatter "kafka.coordinator.group.GroupMetadataManager\\$OffsetsMessageFormatter" --consumer.config config/consumer.properties --from-beginning

# **22. 获取指定consumer group的位移信息** #

需consumer.properties中设置exclude.internal.topics=false：

1) 0.11.0.0版本之前：

kafka-simple-consumer-shell.sh --topic \_\_consumer\_offsets --partition 11 --broker-list localhost:9091,localhost:9092,localhost:9093 --formatter "kafka.coordinator.GroupMetadataManager\\$OffsetsMessageFormatter"

2) 0.11.0.0版本以后(含)：

# **23. 20) 查看kafka的zookeeper** #

1) 查看Kakfa在zookeeper的根目录

<table style="width:426.1pt;"> 
 <tbody> 
  <tr> 
   <td style="vertical-align:top;width:426.1pt;"> <p style="margin-left:0pt;">[zk:&nbsp;localhost:2181(CONNECTED)&nbsp;0]&nbsp;ls&nbsp;/kafka</p> <p style="margin-left:0pt;">[cluster,&nbsp;controller_epoch,&nbsp;controller,&nbsp;brokers,&nbsp;admin,&nbsp;isr_change_notification,&nbsp;consumers,&nbsp;config]</p> </td> 
  </tr> 
 </tbody> 
</table>

2) 查看brokers

<table style="width:426.1pt;"> 
 <tbody> 
  <tr> 
   <td style="vertical-align:top;width:426.1pt;"> <p style="margin-left:0pt;">[zk:&nbsp;localhost:2181(CONNECTED)&nbsp;1]&nbsp;ls&nbsp;/kafka/brokers</p> <p style="margin-left:0pt;">[ids,&nbsp;topics,&nbsp;seqid]</p> </td> 
  </tr> 
 </tbody> 
</table>

3) 查看有哪些brokers（214和215等为server.properties中配置的broker.id值）：

<table style="width:426.1pt;"> 
 <tbody> 
  <tr> 
   <td style="vertical-align:top;width:426.1pt;"> <p style="margin-left:0pt;">[zk:&nbsp;localhost:2181(CONNECTED)&nbsp;2]&nbsp;ls&nbsp;/kafka/brokers/ids</p> <p style="margin-left:0pt;">[214,&nbsp;215,&nbsp;138,&nbsp;139]</p> </td> 
  </tr> 
 </tbody> 
</table>

4) 查看broker 214，下列数据显示该broker没有设置JMX\_PORT：

<table style="width:426.1pt;"> 
 <tbody> 
  <tr> 
   <td style="vertical-align:top;width:426.1pt;"> <p style="margin-left:0pt;">[zk:&nbsp;localhost:2181(CONNECTED)&nbsp;4]&nbsp;get&nbsp;/kafka/brokers/ids/214</p> <p style="margin-left:0pt;">{"listener_security_protocol_map":{"PLAINTEXT":"PLAINTEXT"},"endpoints":["PLAINTEXT://test-204:9092"],"jmx_port":-1,"host":"test-204","timestamp":"1498467464861","port":9092,"version":4}</p> <p style="margin-left:0pt;">cZxid&nbsp;=&nbsp;0x200002400</p> <p style="margin-left:0pt;">ctime&nbsp;=&nbsp;Mon&nbsp;Jun&nbsp;26&nbsp;16:57:44&nbsp;CST&nbsp;2017</p> <p style="margin-left:0pt;">mZxid&nbsp;=&nbsp;0x200002400</p> <p style="margin-left:0pt;">mtime&nbsp;=&nbsp;Mon&nbsp;Jun&nbsp;26&nbsp;16:57:44&nbsp;CST&nbsp;2017</p> <p style="margin-left:0pt;">pZxid&nbsp;=&nbsp;0x200002400</p> <p style="margin-left:0pt;">cversion&nbsp;=&nbsp;0</p> <p style="margin-left:0pt;">dataVersion&nbsp;=&nbsp;0</p> <p style="margin-left:0pt;">aclVersion&nbsp;=&nbsp;0</p> <p style="margin-left:0pt;">ephemeralOwner&nbsp;=&nbsp;0x45b9d9e841f0136</p> <p style="margin-left:0pt;">dataLength&nbsp;=&nbsp;190</p> <p style="margin-left:0pt;">numChildren&nbsp;=&nbsp;0</p> </td> 
  </tr> 
 </tbody> 
</table>

5) 查看controller，下列数据显示broker 214为controller：

<table style="width:426.1pt;"> 
 <tbody> 
  <tr> 
   <td style="vertical-align:top;width:426.1pt;"> <p style="margin-left:0pt;">[zk:&nbsp;localhost:2181(CONNECTED)&nbsp;9]&nbsp;get&nbsp;/kafka/controller</p> <p style="margin-left:0pt;">{"version":1,"brokerid":214,"timestamp":"1498467946988"}</p> <p style="margin-left:0pt;">cZxid&nbsp;=&nbsp;0x200002438</p> <p style="margin-left:0pt;">ctime&nbsp;=&nbsp;Mon&nbsp;Jun&nbsp;26&nbsp;17:05:46&nbsp;CST&nbsp;2017</p> <p style="margin-left:0pt;">mZxid&nbsp;=&nbsp;0x200002438</p> <p style="margin-left:0pt;">mtime&nbsp;=&nbsp;Mon&nbsp;Jun&nbsp;26&nbsp;17:05:46&nbsp;CST&nbsp;2017</p> <p style="margin-left:0pt;">pZxid&nbsp;=&nbsp;0x200002438</p> <p style="margin-left:0pt;">cversion&nbsp;=&nbsp;0</p> <p style="margin-left:0pt;">dataVersion&nbsp;=&nbsp;0</p> <p style="margin-left:0pt;">aclVersion&nbsp;=&nbsp;0</p> <p style="margin-left:0pt;">ephemeralOwner&nbsp;=&nbsp;0x45b9d9e841f0136</p> <p style="margin-left:0pt;">dataLength&nbsp;=&nbsp;56</p> <p style="margin-left:0pt;">numChildren&nbsp;=&nbsp;0</p> </td> 
  </tr> 
 </tbody> 
</table>

6) 查看kafka集群的id：

<table style="width:426.1pt;"> 
 <tbody> 
  <tr> 
   <td style="vertical-align:top;width:426.1pt;"> <p style="margin-left:0pt;">[zk:&nbsp;localhost:2181(CONNECTED)&nbsp;13]&nbsp;get&nbsp;/kafka/cluster/id</p> <p style="margin-left:0pt;">{"version":"1","id":"OCAEJy4qSf29bhwOfO7kNQ"}</p> <p style="margin-left:0pt;">cZxid&nbsp;=&nbsp;0x2000023e7</p> <p style="margin-left:0pt;">ctime&nbsp;=&nbsp;Mon&nbsp;Jun&nbsp;26&nbsp;16:57:28&nbsp;CST&nbsp;2017</p> <p style="margin-left:0pt;">mZxid&nbsp;=&nbsp;0x2000023e7</p> <p style="margin-left:0pt;">mtime&nbsp;=&nbsp;Mon&nbsp;Jun&nbsp;26&nbsp;16:57:28&nbsp;CST&nbsp;2017</p> <p style="margin-left:0pt;">pZxid&nbsp;=&nbsp;0x2000023e7</p> <p style="margin-left:0pt;">cversion&nbsp;=&nbsp;0</p> <p style="margin-left:0pt;">dataVersion&nbsp;=&nbsp;0</p> <p style="margin-left:0pt;">aclVersion&nbsp;=&nbsp;0</p> <p style="margin-left:0pt;">ephemeralOwner&nbsp;=&nbsp;0x0</p> <p style="margin-left:0pt;">dataLength&nbsp;=&nbsp;45</p> <p style="margin-left:0pt;">numChildren&nbsp;=&nbsp;0</p> </td> 
  </tr> 
 </tbody> 
</table>

7) 查看有哪些topics：

<table style="width:426.1pt;"> 
 <tbody> 
  <tr> 
   <td style="vertical-align:top;width:426.1pt;"> <p style="margin-left:0pt;">[zk:&nbsp;localhost:2181(CONNECTED)&nbsp;16]&nbsp;ls&nbsp;/kafka/brokers/topics</p> <p style="margin-left:0pt;">[test,&nbsp;my-replicated-topic,&nbsp;test1,&nbsp;test2,&nbsp;test3,&nbsp;test123,&nbsp;__consumer_offsets,&nbsp;info]</p> </td> 
  </tr> 
 </tbody> 
</table>

8) 查看topic下有哪些partitions：

<table style="width:426.1pt;"> 
 <tbody> 
  <tr> 
   <td style="vertical-align:top;width:426.1pt;"> <p style="margin-left:0pt;">[zk:&nbsp;localhost:2181(CONNECTED)&nbsp;19]&nbsp;ls&nbsp;/kafka/brokers/topics/__consumer_offsets/partitions</p> <p style="margin-left:0pt;">[44,&nbsp;45,&nbsp;46,&nbsp;47,&nbsp;48,&nbsp;49,&nbsp;10,&nbsp;11,&nbsp;12,&nbsp;13,&nbsp;14,&nbsp;15,&nbsp;16,&nbsp;17,&nbsp;18,&nbsp;19,&nbsp;0,&nbsp;1,&nbsp;2,&nbsp;3,&nbsp;4,&nbsp;5,&nbsp;6,&nbsp;7,&nbsp;8,&nbsp;9,&nbsp;20,&nbsp;21,&nbsp;22,&nbsp;23,&nbsp;24,&nbsp;25,&nbsp;26,&nbsp;27,&nbsp;28,&nbsp;29,&nbsp;30,&nbsp;31,&nbsp;32,&nbsp;33,&nbsp;34,&nbsp;35,&nbsp;36,&nbsp;37,&nbsp;38,&nbsp;39,&nbsp;40,&nbsp;41,&nbsp;42,&nbsp;43]</p> </td> 
  </tr> 
 </tbody> 
</table>

9) 查看“partition 0”的状态：

<table style="width:426.1pt;"> 
 <tbody> 
  <tr> 
   <td style="vertical-align:top;width:426.1pt;"> <p style="margin-left:0pt;">[zk:&nbsp;localhost:2181(CONNECTED)&nbsp;22]&nbsp;get&nbsp;/kafka/brokers/topics/__consumer_offsets/partitions/0/state</p> <p style="margin-left:0pt;">{"controller_epoch":2,"leader":215,"version":1,"leader_epoch":1,"isr":[215,214]}</p> <p style="margin-left:0pt;">cZxid&nbsp;=&nbsp;0x2000024c6</p> <p style="margin-left:0pt;">ctime&nbsp;=&nbsp;Mon&nbsp;Jun&nbsp;26&nbsp;18:02:07&nbsp;CST&nbsp;2017</p> <p style="margin-left:0pt;">mZxid&nbsp;=&nbsp;0x200bc4fc3</p> <p style="margin-left:0pt;">mtime&nbsp;=&nbsp;Mon&nbsp;Aug&nbsp;27&nbsp;18:58:10&nbsp;CST&nbsp;2018</p> <p style="margin-left:0pt;">pZxid&nbsp;=&nbsp;0x2000024c6</p> <p style="margin-left:0pt;">cversion&nbsp;=&nbsp;0</p> <p style="margin-left:0pt;">dataVersion&nbsp;=&nbsp;1</p> <p style="margin-left:0pt;">aclVersion&nbsp;=&nbsp;0</p> <p style="margin-left:0pt;">ephemeralOwner&nbsp;=&nbsp;0x0</p> <p style="margin-left:0pt;">dataLength&nbsp;=&nbsp;80</p> <p style="margin-left:0pt;">numChildren&nbsp;=&nbsp;0</p> </td> 
  </tr> 
 </tbody> 
</table>

# **24. 如何增加\_\_consumer\_offsets的副本数？** #

可使用kafka-reassign-partitions.sh来增加\_\_consumer\_offsets的副本数，方法如下，构造一JSON文件reassign.json：

<table style="width:426.1pt;"> 
 <tbody> 
  <tr> 
   <td style="vertical-align:top;width:426.1pt;"> <p style="margin-left:0pt;">{ </p> <p style="margin-left:0pt;">&nbsp;&nbsp;&nbsp;&nbsp;"version":1,</p> <p style="margin-left:0pt;">&nbsp;&nbsp;&nbsp;&nbsp;"partitions":[</p> <p style="margin-left:0pt;">&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;{"topic":"__consumer_offsets","partition":0,"replicas":[1,2,3]},&nbsp;</p> <p style="margin-left:0pt;">&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;{"topic":"__consumer_offsets","partition":1,"replicas":[2,3,1]},</p> <p style="margin-left:0pt;">&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;{"topic":"__consumer_offsets","partition":2,"replicas":[3,1,2]},</p> <p style="margin-left:0pt;">&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;{"topic":"__consumer_offsets","partition":3,"replicas":[1,2,3]},</p> <p style="margin-left:0pt;">&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;...</p> <p style="margin-left:0pt;">&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;{"topic":"__consumer_offsets","partition":100,"replicas":[2,3,1]}</p> <p style="margin-left:0pt;">&nbsp;&nbsp;&nbsp;&nbsp;]</p> <p style="margin-left:0pt;">}</p> </td> 
  </tr> 
 </tbody> 
</table>

然后执行：

<table style="width:426.1pt;"> 
 <tbody> 
  <tr> 
   <td style="vertical-align:top;width:426.1pt;"> <p style="margin-left:0pt;">kafka-reassign-partitions.sh&nbsp;--zookeeper&nbsp;localhost:2181/kafka&nbsp;--reassignment-json-file&nbsp;reassign.json&nbsp;--execute</p> </td> 
  </tr> 
 </tbody> 
</table>

“\[1,2,3\]”中的数字为broker.id值。

如果执行报错“Partitions reassignment failed due to Partition reassignment data file is empty”，可能是因为reasign.json文件格式不对，比如成下列格式了（中间的没有以逗号结尾）：

<table style="width:426.1pt;"> 
 <tbody> 
  <tr> 
   <td style="vertical-align:top;width:426.1pt;"> <p style="margin-left:0pt;">{"topic":"__consumer_offsets","partition":29,"replicas":[5,3,2]},</p> <p style="margin-left:0pt;">{"topic":"__consumer_offsets","partition":30,"replicas":[1,4,3]}</p> <p style="margin-left:0pt;">{"topic":"__consumer_offsets","partition":31,"replicas":[2,5,4]}</p> <p style="margin-left:0pt;">{"topic":"__consumer_offsets","partition":32,"replicas":[3,2,5]}</p> <p style="margin-left:0pt;">{"topic":"__consumer_offsets","partition":33,"replicas":[4,3,1]},</p> </td> 
  </tr> 
 </tbody> 
</table>

如果执行遇到下列错误：

<table style="width:426.1pt;"> 
 <tbody> 
  <tr> 
   <td style="vertical-align:top;width:426.1pt;"> <p style="margin-left:0pt;">Partitions&nbsp;reassignment&nbsp;failed&nbsp;due&nbsp;to&nbsp;Partition&nbsp;replica&nbsp;lists&nbsp;may&nbsp;not&nbsp;contain&nbsp;duplicate&nbsp;entries:&nbsp;__consumer_offsets-<strong>16&nbsp;</strong>contains&nbsp;multiple&nbsp;entries&nbsp;for&nbsp;<strong>2</strong>.&nbsp;__consumer_offsets-<strong>39&nbsp;</strong>contains&nbsp;multiple&nbsp;entries&nbsp;for&nbsp;<strong>2</strong>.&nbsp;__consumer_offsets-<strong>40&nbsp;</strong>contains&nbsp;multiple&nbsp;entries&nbsp;for&nbsp;<strong>3</strong>.&nbsp;__consumer_offsets-<strong>44&nbsp;</strong>contains&nbsp;multiple&nbsp;entries&nbsp;for&nbsp;<strong>3</strong></p> </td> 
  </tr> 
 </tbody> 
</table>

原因是一个分区的两个副本被指定在同一个broker上，以16号分区为列，有两个副本落在了broker 2上：

执行成功后的输出：

<table style="width:426.1pt;"> 
 <tbody> 
  <tr> 
   <td style="vertical-align:top;width:426.1pt;"> <p style="margin-left:0pt;">$&nbsp;../bin/<strong>kafka-reassign-partitions.sh</strong>&nbsp;--zookeeper&nbsp;192.168.1.35.31:2181/kafka&nbsp;--reassignment-json-file&nbsp;__consumer_offsets.reassign&nbsp;--execute</p> <p style="margin-left:0pt;">Current&nbsp;partition&nbsp;replica&nbsp;assignment</p> <p style="margin-left:0pt;">&nbsp;</p> <p style="margin-left:0pt;">{"version":1,"partitions":[{"topic":"__consumer_offsets","partition":22,"replicas":[2],"log_dirs":["any"]},{"topic":"__consumer_offsets","partition":30,"replicas":[5],"log_dirs":["any"]},{"topic":"__consumer_offsets","partition":8,"replicas":[3],"log_dirs":["any"]},{"topic":"__consumer_offsets","partition":21,"replicas":[1],"log_dirs":["any"]},{"topic":"__consumer_offsets","partition":4,"replicas":[4],"log_dirs":["any"]},{"topic":"__consumer_offsets","partition":27,"replicas":[2],"log_dirs":["any"]},{"topic":"__consumer_offsets","partition":7,"replicas":[2],"log_dirs":["any"]},{"topic":"__consumer_offsets","partition":9,"replicas":[4],"log_dirs":["any"]},{"topic":"__consumer_offsets","partition":46,"replicas":[1],"log_dirs":["any"]},{"topic":"__consumer_offsets","partition":25,"replicas":[5],"log_dirs":["any"]},{"topic":"__consumer_offsets","partition":35,"replicas":[5],"log_dirs":["any"]},{"topic":"__consumer_offsets","partition":41,"replicas":[1],"log_dirs":["any"]},{"topic":"__consumer_offsets","partition":33,"replicas":[3],"log_dirs":["any"]},{"topic":"__consumer_offsets","partition":23,"replicas":[3],"log_dirs":["any"]},{"topic":"__consumer_offsets","partition":49,"replicas":[4],"log_dirs":["any"]},{"topic":"__consumer_offsets","partition":47,"replicas":[2],"log_dirs":["any"]},{"topic":"__consumer_offsets","partition":16,"replicas":[1],"log_dirs":["any"]},{"topic":"__consumer_offsets","partition":28,"replicas":[3],"log_dirs":["any"]},{"topic":"__consumer_offsets","partition":31,"replicas":[1],"log_dirs":["any"]},{"topic":"__consumer_offsets","partition":36,"replicas":[1],"log_dirs":["any"]},{"topic":"__consumer_offsets","partition":42,"replicas":[2],"log_dirs":["any"]},{"topic":"__consumer_offsets","partition":3,"replicas":[3],"log_dirs":["any"]},{"topic":"__consumer_offsets","partition":18,"replicas":[3],"log_dirs":["any"]},{"topic":"__consumer_offsets","partition":37,"replicas":[2],"log_dirs":["any"]},{"topic":"__consumer_offsets","partition":15,"replicas":[5],"log_dirs":["any"]},{"topic":"__consumer_offsets","partition":24,"replicas":[4],"log_dirs":["any"]},{"topic":"__consumer_offsets","partition":38,"replicas":[3],"log_dirs":["any"]},{"topic":"__consumer_offsets","partition":17,"replicas":[2],"log_dirs":["any"]},{"topic":"__consumer_offsets","partition":48,"replicas":[3],"log_dirs":["any"]},{"topic":"__consumer_offsets","partition":19,"replicas":[4],"log_dirs":["any"]},{"topic":"__consumer_offsets","partition":11,"replicas":[1],"log_dirs":["any"]},{"topic":"__consumer_offsets","partition":13,"replicas":[3],"log_dirs":["any"]},{"topic":"__consumer_offsets","partition":2,"replicas":[2],"log_dirs":["any"]},{"topic":"__consumer_offsets","partition":43,"replicas":[3],"log_dirs":["any"]},{"topic":"__consumer_offsets","partition":6,"replicas":[1],"log_dirs":["any"]},{"topic":"__consumer_offsets","partition":14,"replicas":[4],"log_dirs":["any"]},{"topic":"__consumer_offsets","partition":20,"replicas":[5],"log_dirs":["any"]},{"topic":"__consumer_offsets","partition":0,"replicas":[5],"log_dirs":["any"]},{"topic":"__consumer_offsets","partition":44,"replicas":[4],"log_dirs":["any"]},{"topic":"__consumer_offsets","partition":39,"replicas":[4],"log_dirs":["any"]},{"topic":"__consumer_offsets","partition":12,"replicas":[2],"log_dirs":["any"]},{"topic":"__consumer_offsets","partition":45,"replicas":[5],"log_dirs":["any"]},{"topic":"__consumer_offsets","partition":1,"replicas":[1],"log_dirs":["any"]},{"topic":"__consumer_offsets","partition":5,"replicas":[5],"log_dirs":["any"]},{"topic":"__consumer_offsets","partition":26,"replicas":[1],"log_dirs":["any"]},{"topic":"__consumer_offsets","partition":29,"replicas":[4],"log_dirs":["any"]},{"topic":"__consumer_offsets","partition":34,"replicas":[4],"log_dirs":["any"]},{"topic":"__consumer_offsets","partition":10,"replicas":[5],"log_dirs":["any"]},{"topic":"__consumer_offsets","partition":32,"replicas":[2],"log_dirs":["any"]},{"topic":"__consumer_offsets","partition":40,"replicas":[5],"log_dirs":["any"]}]}</p> <p style="margin-left:0pt;">&nbsp;</p> <p style="margin-left:0pt;">Save&nbsp;this&nbsp;to&nbsp;use&nbsp;as&nbsp;the&nbsp;--reassignment-json-file&nbsp;option&nbsp;during&nbsp;rollback</p> <p style="margin-left:0pt;"><strong>Successfully&nbsp;</strong>started&nbsp;reassignment&nbsp;of&nbsp;partitions.</p> </td> 
  </tr> 
 </tbody> 
</table>

# **25. 问题** #

1) -190,Local: Unknown partition

比如单机版只有一个分区，但prodcue参数的分区值为1等。

2) Rdkafka程序日志“delivery failed. errMsg:\[Local: Message timed out\]”

同一个程序，在有些机器上会这个错误，有些机器则工作正常，相关的issues：

https://github.com/edenhill/librdkafka/issues/474

实测是因为在运行Kafka应用程序的机器上没有配置Kafka Brokers机器的hosts。

另外的解决办法是在server.properties配置listeners和advertised.listeners，并且使用IP而不是hostname作为值。

3) Name or service not known (after 9020595078ms in state INIT)

event\_callback: type(0), severity(3), (-193)kafka-204:9092/214: Failed to resolve 'kafka-204:9092': Name or service not known (after 9020595078ms in state INIT)

原因是运行kafka应用程序（非kafka本身）的机器不能识别主机名kafka-204（Kafka Brokers机器可以识别），

解决办法是在server.properties配置listeners和advertised.listeners，并且使用IP而不是hostname作为值。

# **附1：进程监控工具process\_monitor.sh** #

**process\_monitor.sh**为shell脚本，本身含详细的使用说明和帮助提示。适合放在crontab中，检测到进程不在时，3秒左右时间重拉起。支持不同用户运行相同程序，也支持同一用户带不同参数运行相同程序。

下载网址：

[https://github.com/eyjian/libmooon/blob/master/shell/process\_monitor.sh][https_github.com_eyjian_libmooon_blob_master_shell_process_monitor.sh]

使用示例：

<table style="width:426.1pt;"> 
 <tbody> 
  <tr> 
   <td style="vertical-align:top;width:426.1pt;"> <p style="margin-left:0pt;">*&nbsp;*&nbsp;*&nbsp;*&nbsp;*&nbsp;/usr/local/bin/process_monitor.sh&nbsp;"/usr/local/jdk/bin/java&nbsp;<span style="color:#0000ff;"><strong>kafkaServer</strong></span>"&nbsp;"/data/kafka/bin/kafka-server-start.sh&nbsp;-<strong>daemon&nbsp;</strong>/data/kafka/config/server.properties"</p> </td> 
  </tr> 
 </tbody> 
</table>

由于所有的java程序均运行在JVM中，所以程序名均为java，“**kafkaServer**”用于限定只监控kafka。如果同一用户运行多个kafka实例，则需加端口号区分，并且要求端口号为命令行参数，和“**kafkaServer**”共同组成匹配模式。

当检测到进程不存在时，则执行第三列的**重启**指令“/data/kafka/bin/kafka-server-**start**.sh -daemon /data/kafka/config/server.properties”。

使用示例2，监控zooekeeper：

<table style="width:426.1pt;"> 
 <tbody> 
  <tr> 
   <td style="vertical-align:top;width:426.1pt;"> <p style="margin-left:0pt;">*&nbsp;*&nbsp;*&nbsp;*&nbsp;*&nbsp;/usr/local/bin/process_monitor.sh&nbsp;"/usr/local/jdk/bin/java&nbsp;-<strong>Dzookeeper</strong>"&nbsp;"/data/zookeeper/bin/zkServer.sh&nbsp;<strong>start</strong>"</p> </td> 
  </tr> 
 </tbody> 
</table>

# **附2：批量操作工具** #

适用用来批量安装kafka和日常运维。

下载网址：

[https://github.com/eyjian/libmooon/releases][https_github.com_eyjian_libmooon_releases]

监控工具有两个版本：一是C++版本，另一是GO版本。当前C++版本比较成熟，GO版本相当简略，但C++版本依赖C++运行时库，不同环境需要特定编译，而GO版本可不依赖C和C++运行时库，所以不需编译即可应用到广泛的Linux环境。

使用简单，直接执行命令，即会提示用法。

## **附2.1：批量执行命令工具：mooon\_ssh** ##

<table style="width:426.1pt;"> 
 <tbody> 
  <tr> 
   <td style="vertical-align:top;width:44.75pt;"> <p style="margin-left:0pt;"><strong>参数名</strong></p> </td> 
   <td style="vertical-align:top;width:96.6pt;"> <p style="margin-left:0pt;"><strong>默认值</strong></p> </td> 
   <td style="vertical-align:top;width:284.75pt;"> <p style="margin-left:0pt;"><strong>说明</strong></p> </td> 
  </tr> 
  <tr> 
   <td style="vertical-align:top;width:44.75pt;"> <p style="margin-left:0pt;">-u</p> </td> 
   <td style="vertical-align:top;width:96.6pt;"> <p style="margin-left:0pt;">无</p> </td> 
   <td style="vertical-align:top;width:284.75pt;"> <p style="margin-left:0pt;">用户名参数，可用环境变量<span style="color:#0000ff;"><strong>U</strong></span>替代</p> </td> 
  </tr> 
  <tr> 
   <td style="vertical-align:top;width:44.75pt;"> <p style="margin-left:0pt;">-p</p> </td> 
   <td style="vertical-align:top;width:96.6pt;"> <p style="margin-left:0pt;">无</p> </td> 
   <td style="vertical-align:top;width:284.75pt;"> <p style="margin-left:0pt;">密码参数，可用环境变量<span style="color:#0000ff;"><strong>P</strong></span>替代</p> </td> 
  </tr> 
  <tr> 
   <td style="vertical-align:top;width:44.75pt;"> <p style="margin-left:0pt;">-h</p> </td> 
   <td style="vertical-align:top;width:96.6pt;"> <p style="margin-left:0pt;">无</p> </td> 
   <td style="vertical-align:top;width:284.75pt;"> <p style="margin-left:0pt;">IP列表参数，可用环境变量<span style="color:#0000ff;"><strong>H</strong></span>替代</p> </td> 
  </tr> 
  <tr> 
   <td style="vertical-align:top;width:44.75pt;"> <p style="margin-left:0pt;">-P</p> </td> 
   <td style="vertical-align:top;width:96.6pt;"> <p style="margin-left:0pt;">22，可修改源码，编译为常用端口号</p> </td> 
   <td style="vertical-align:top;width:284.75pt;"> <p style="margin-left:0pt;">SSH端口参数，可用环境变量<span style="color:#0000ff;"><strong>PORT</strong></span>替代</p> </td> 
  </tr> 
  <tr> 
   <td style="vertical-align:top;width:44.75pt;"> <p style="margin-left:0pt;">-c</p> </td> 
   <td style="vertical-align:top;width:96.6pt;"> <p style="margin-left:0pt;">无</p> </td> 
   <td style="vertical-align:top;width:284.75pt;"> <p style="margin-left:0pt;">在远程机器上执行的命令，建议单引号方式指定值，除非要执行的命令本身已经包含了单引号有冲突。使用双引号时，要注意转义，否则会被本地shell解释</p> </td> 
  </tr> 
  <tr> 
   <td style="vertical-align:top;width:44.75pt;"> <p style="margin-left:0pt;">-v</p> </td> 
   <td style="vertical-align:top;width:96.6pt;"> <p style="margin-left:0pt;">1</p> </td> 
   <td style="vertical-align:top;width:284.75pt;"> <p style="margin-left:0pt;">工具输出的详细度</p> </td> 
  </tr> 
 </tbody> 
</table>

## **附2.2：批量上传文件工具：mooon\_upload** ##

<table style="width:426.1pt;"> 
 <tbody> 
  <tr> 
   <td style="vertical-align:top;width:74.75pt;"> <p style="margin-left:0pt;"><strong>参数名</strong></p> </td> 
   <td style="vertical-align:top;width:73.8pt;"> <p style="margin-left:0pt;"><strong>默认值</strong></p> </td> 
   <td style="vertical-align:top;width:277.55pt;"> <p style="margin-left:0pt;"><strong>说明</strong></p> </td> 
  </tr> 
  <tr> 
   <td style="vertical-align:top;width:74.75pt;"> <p style="margin-left:0pt;">-u</p> </td> 
   <td style="vertical-align:top;width:73.8pt;"> <p style="margin-left:0pt;">无</p> </td> 
   <td style="vertical-align:top;width:277.55pt;"> <p style="margin-left:0pt;">用户名参数，可用环境变量<span style="color:#0000ff;"><strong>U</strong></span>替代</p> </td> 
  </tr> 
  <tr> 
   <td style="vertical-align:top;width:74.75pt;"> <p style="margin-left:0pt;">-p</p> </td> 
   <td style="vertical-align:top;width:73.8pt;"> <p style="margin-left:0pt;">无</p> </td> 
   <td style="vertical-align:top;width:277.55pt;"> <p style="margin-left:0pt;">密码参数，可用环境变量<span style="color:#0000ff;"><strong>P</strong></span>替代</p> </td> 
  </tr> 
  <tr> 
   <td style="vertical-align:top;width:74.75pt;"> <p style="margin-left:0pt;">-h</p> </td> 
   <td style="vertical-align:top;width:73.8pt;"> <p style="margin-left:0pt;">无</p> </td> 
   <td style="vertical-align:top;width:277.55pt;"> <p style="margin-left:0pt;">IP列表参数，可用环境变量<span style="color:#0000ff;"><strong>H</strong></span>替代</p> </td> 
  </tr> 
  <tr> 
   <td style="vertical-align:top;width:74.75pt;"> <p style="margin-left:0pt;">-P</p> </td> 
   <td style="vertical-align:top;width:73.8pt;"> <p style="margin-left:0pt;">22，可修改源码，编译为常用端口号</p> </td> 
   <td style="vertical-align:top;width:277.55pt;"> <p style="margin-left:0pt;">SSH端口参数，可用环境变量<span style="color:#0000ff;"><strong>PORT</strong></span>替代</p> </td> 
  </tr> 
  <tr> 
   <td style="vertical-align:top;width:74.75pt;"> <p style="margin-left:0pt;">-s</p> </td> 
   <td style="vertical-align:top;width:73.8pt;"> <p style="margin-left:0pt;">无</p> </td> 
   <td style="vertical-align:top;width:277.55pt;"> <p style="margin-left:0pt;">以逗号分隔的，需要上传的本地文件列表，可以带相对或绝对目录</p> </td> 
  </tr> 
  <tr> 
   <td style="vertical-align:top;width:74.75pt;"> <p style="margin-left:0pt;">-d</p> </td> 
   <td style="vertical-align:top;width:73.8pt;"> <p style="margin-left:0pt;">无</p> </td> 
   <td style="vertical-align:top;width:277.55pt;"> <p style="margin-left:0pt;">文件上传到远程机器的目录，只能为单个目录</p> </td> 
  </tr> 
 </tbody> 
</table>

## **附2.3：使用示例** ##

1) 使用**示例1**：上传/etc/hosts

<table style="width:426.1pt;"> 
 <tbody> 
  <tr> 
   <td style="vertical-align:top;width:426.1pt;"> <p style="margin-left:0pt;">mooon_upload&nbsp;-s=/etc/hosts&nbsp;-d=/etc</p> </td> 
  </tr> 
 </tbody> 
</table>

2) 使用**示例2**：检查/etc/profile文件是否一致

<table style="width:426.1pt;"> 
 <tbody> 
  <tr> 
   <td style="vertical-align:top;width:426.1pt;"> <p style="margin-left:0pt;">mooon_ssh&nbsp;-c='md5sum&nbsp;/etc/hosts'</p> </td> 
  </tr> 
 </tbody> 
</table>

3) 使用**示例3**：批量查看crontab

<table style="width:426.1pt;"> 
 <tbody> 
  <tr> 
   <td style="vertical-align:top;width:426.1pt;"> <p style="margin-left:0pt;">mooon_ssh&nbsp;-c='crontab&nbsp;-l'</p> </td> 
  </tr> 
 </tbody> 
</table>

4) 使用**示例4**：批量清空crontab

<table style="width:426.1pt;"> 
 <tbody> 
  <tr> 
   <td style="vertical-align:top;width:426.1pt;"> <p style="margin-left:0pt;">mooon_ssh&nbsp;-c='rm&nbsp;-f&nbsp;/tmp/crontab.empty;touch&nbsp;/tmp/crontab.empty'</p> <p style="margin-left:0pt;">mooon_ssh&nbsp;-c='crontab&nbsp;/tmp/crontab.emtpy'</p> </td> 
  </tr> 
 </tbody> 
</table>

5) 使用**示例5**：批量更新crontab

<table style="width:426.1pt;"> 
 <tbody> 
  <tr> 
   <td style="vertical-align:top;width:426.1pt;"> <p style="margin-left:0pt;">mooon_ssh&nbsp;-c='crontab&nbsp;/tmp/crontab.online'</p> </td> 
  </tr> 
 </tbody> 
</table>

6) 使用**示例6**：取远端机器IP

因为awk用单引号，所以参数“-c”的值不能使用单引号，所以内容需要转义，相对其它来说要复杂点：

<table style="width:426.1pt;"> 
 <tbody> 
  <tr> 
   <td style="vertical-align:top;width:426.1pt;"> <p style="margin-left:0pt;">mooon_ssh&nbsp;-c="<strong>netstat&nbsp;</strong>-ie&nbsp;|&nbsp;<strong>awk&nbsp;</strong>-F[\\&nbsp;:]+&nbsp;'BEGIN{ok=0;}{if&nbsp;(match(\$0,&nbsp;\"eth1\"))&nbsp;ok=1;&nbsp;if&nbsp;((1==ok)&nbsp;&amp;&amp;&nbsp;match(\$0,\"inet\"))&nbsp;{&nbsp;ok=0;&nbsp;if&nbsp;(<strong>7==NF</strong>)&nbsp;printf(\"%s\\n\",\$3);&nbsp;else&nbsp;printf(\"%s\\n\",\$4);}&nbsp;}'"</p> </td> 
  </tr> 
 </tbody> 
</table>

不同的环境，IP在“netstat -ie”输出中的位置稍有不同，所以awk中加了“7==NF”判断，但仍不一定适用于所有的环境。需要转义的字符包含：双引号、美元符和斜杠。

7) 使用**示例7**：批量查看kafka进程（环境变量方式）

<table style="width:426.1pt;"> 
 <tbody> 
  <tr> 
   <td style="vertical-align:top;width:426.1pt;"> <p style="margin-left:0pt;">$&nbsp;export&nbsp;H=192.168.31.9,192.168.31.10,192.168.31.11,192.168.31.12,192.168.31.13</p> <p style="margin-left:0pt;">$&nbsp;export&nbsp;U=kafka</p> <p style="margin-left:0pt;">$&nbsp;export&nbsp;P='123456'</p> <p style="margin-left:0pt;">&nbsp;</p> <p style="margin-left:0pt;">$&nbsp;<strong>mooon_ssh</strong>&nbsp;<strong>-c</strong>='<span style="color:#0000ff;">/usr/local/jdk/bin/</span><span style="color:#0000ff;"><strong>jps&nbsp;</strong></span><span style="color:#0000ff;">-m</span>'</p> <p style="margin-left:0pt;">[192.168.31.15]</p> <p style="margin-left:0pt;">50928&nbsp;Kafka&nbsp;/data/kafka/config/server.properties</p> <p style="margin-left:0pt;">125735&nbsp;Jps&nbsp;-m</p> <p style="margin-left:0pt;">[192.168.31.15]&nbsp;SUCCESS</p> <p style="margin-left:0pt;">&nbsp;</p> <p style="margin-left:0pt;">[192.168.31.16]</p> <p style="margin-left:0pt;">147842&nbsp;Jps&nbsp;-m</p> <p style="margin-left:0pt;">174902&nbsp;Kafka&nbsp;/data/kafka/config/server.properties</p> <p style="margin-left:0pt;">[192.168.31.16]&nbsp;SUCCESS</p> <p style="margin-left:0pt;">&nbsp;</p> <p style="margin-left:0pt;">[192.168.31.17]</p> <p style="margin-left:0pt;">51409&nbsp;Kafka&nbsp;/data/kafka/config/server.properties</p> <p style="margin-left:0pt;">178771&nbsp;Jps&nbsp;-m</p> <p style="margin-left:0pt;">[192.168.31.17]&nbsp;SUCCESS</p> <p style="margin-left:0pt;">&nbsp;</p> <p style="margin-left:0pt;">[192.168.31.18]</p> <p style="margin-left:0pt;">73568&nbsp;Jps&nbsp;-m</p> <p style="margin-left:0pt;">62314&nbsp;Kafka&nbsp;/data/kafka/config/server.properties</p> <p style="margin-left:0pt;">[192.168.31.18]&nbsp;SUCCESS</p> <p style="margin-left:0pt;">&nbsp;</p> <p style="margin-left:0pt;">[192.168.31.19]</p> <p style="margin-left:0pt;">123908&nbsp;Jps&nbsp;-m</p> <p style="margin-left:0pt;">182845&nbsp;Kafka&nbsp;/data/kafka/config/server.properties</p> <p style="margin-left:0pt;">[192.168.31.19]&nbsp;SUCCESS</p> <p style="margin-left:0pt;">&nbsp;</p> <p style="margin-left:0pt;">&nbsp;</p> <p style="margin-left:0pt;">================================</p> <p style="margin-left:0pt;">[192.168.31.15&nbsp;SUCCESS]&nbsp;0&nbsp;seconds</p> <p style="margin-left:0pt;">[192.168.31.16&nbsp;SUCCESS]&nbsp;0&nbsp;seconds</p> <p style="margin-left:0pt;">[192.168.31.17&nbsp;SUCCESS]&nbsp;0&nbsp;seconds</p> <p style="margin-left:0pt;">[192.168.31.18&nbsp;SUCCESS]&nbsp;0&nbsp;seconds</p> <p style="margin-left:0pt;">[192.168.31.19&nbsp;SUCCESS]&nbsp;0&nbsp;seconds</p> <p style="margin-left:0pt;">SUCCESS:&nbsp;5,&nbsp;FAILURE:&nbsp;0</p> </td> 
  </tr> 
 </tbody> 
</table>

8) 使用**示例8**：批量停止kafka进程（参数方式）

<table style="width:426.1pt;"> 
 <tbody> 
  <tr> 
   <td style="vertical-align:top;width:426.1pt;"> <p style="margin-left:0pt;">$&nbsp;mooon_ssh&nbsp;<strong>-c</strong>='<span style="color:#0000ff;">/data/kafka/bin/kafka-server-stop.sh</span>'&nbsp;<strong>-u</strong>=kafka&nbsp;<strong>-p</strong>='123456'&nbsp;<strong>-h</strong>=192.168.31.15,192.168.31.16,192.168.31.17,192.168.31.18,192.168.31.19</p> <p style="margin-left:0pt;">[192.168.31.15]</p> <p style="margin-left:0pt;">No&nbsp;kafka&nbsp;server&nbsp;to&nbsp;stop</p> <p style="margin-left:0pt;">command&nbsp;return&nbsp;1</p> <p style="margin-left:0pt;">&nbsp;</p> <p style="margin-left:0pt;">[192.168.31.16]</p> <p style="margin-left:0pt;">No&nbsp;kafka&nbsp;server&nbsp;to&nbsp;stop</p> <p style="margin-left:0pt;">command&nbsp;return&nbsp;1</p> <p style="margin-left:0pt;">&nbsp;</p> <p style="margin-left:0pt;">[192.168.31.17]</p> <p style="margin-left:0pt;">No&nbsp;kafka&nbsp;server&nbsp;to&nbsp;stop</p> <p style="margin-left:0pt;">command&nbsp;return&nbsp;1</p> <p style="margin-left:0pt;">&nbsp;</p> <p style="margin-left:0pt;">[192.168.31.18]</p> <p style="margin-left:0pt;">No&nbsp;kafka&nbsp;server&nbsp;to&nbsp;stop</p> <p style="margin-left:0pt;">command&nbsp;return&nbsp;1</p> <p style="margin-left:0pt;">&nbsp;</p> <p style="margin-left:0pt;">[192.168.31.19]</p> <p style="margin-left:0pt;">No&nbsp;kafka&nbsp;server&nbsp;to&nbsp;stop</p> <p style="margin-left:0pt;">command&nbsp;return&nbsp;1</p> <p style="margin-left:0pt;">&nbsp;</p> <p style="margin-left:0pt;">================================</p> <p style="margin-left:0pt;">[192.168.31.15&nbsp;FAILURE]&nbsp;0&nbsp;seconds</p> <p style="margin-left:0pt;">[192.168.31.16&nbsp;FAILURE]&nbsp;0&nbsp;seconds</p> <p style="margin-left:0pt;">[192.168.31.17&nbsp;FAILURE]&nbsp;0&nbsp;seconds</p> <p style="margin-left:0pt;">[192.168.31.18&nbsp;FAILURE]&nbsp;0&nbsp;seconds</p> <p style="margin-left:0pt;">[192.168.31.19&nbsp;FAILURE]&nbsp;0&nbsp;seconds</p> <p style="margin-left:0pt;">SUCCESS:&nbsp;0,&nbsp;FAILURE:&nbsp;5</p> </td> 
  </tr> 
 </tbody> 
</table>

# **附3：批量设置broker.id和listeners工具** #

为shell脚本，有详细的使用说明和帮助提示，依赖mooon\_ssh和mooon\_upload：

[https://github.com/eyjian/libmooon/blob/master/shell/set\_kafka\_id\_and\_ip.sh][https_github.com_eyjian_libmooon_blob_master_shell_set_kafka_id_and_ip.sh]

# **附4：批量设置hostname工具** #

为shell脚本，有详细的使用说明和帮助提示，依赖mooon\_ssh和mooon\_upload：

[https://github.com/eyjian/libmooon/blob/master/shell/set\_hostname.sh][https_github.com_eyjian_libmooon_blob_master_shell_set_hostname.sh]

# **附5：Kafka监控工具kafka-manager** #

官网：[https://github.com/yahoo/kafka-manager][https_github.com_yahoo_kafka-manager]

kafka-manager的数据主要来源两个方便：一是kafka的zookeeper数据，二是kafka的JMX数据。

kafka-manager要求JDK版本不低于1.8，从源码编译kafka-manager相对复杂，但编译拿到二进制包后，只需修改application.conf中的“kafka-manager.zkhosts”值，即可开始启动kafka-manager。“kafka-manager.zkhosts”值，不是kafka的zookeeper配置值，而是kafka-manager自己用的zookeeper配置，所以两者可以为不同的zookeeper，注意值用双引号引起来。

crontab启动示例：

<table style="width:426.1pt;"> 
 <tbody> 
  <tr> 
   <td style="vertical-align:top;width:426.1pt;"> <p style="margin-left:0pt;">JMX_PORT=9999</p> <p style="margin-left:0pt;">*&nbsp;*&nbsp;*&nbsp;*&nbsp;*&nbsp;/usr/local/bin/process_monitor.sh&nbsp;"/usr/local/jdk/bin/java&nbsp;kafkaServer"&nbsp;"/data/kafka/bin/kafka-server-<strong>start</strong>.sh&nbsp;-daemon&nbsp;/data/kafka/config/<strong>server.properties</strong>"</p> </td> 
  </tr> 
 </tbody> 
</table>

指定**JMX\_POR**T不是必须的，但建议设置，这样kafka-manager可以更详细的查看brokers。

crontab中启动kafka-manager示例（指定服务端口为8080，不指定的默认值为9000）：

<table style="width:426.1pt;"> 
 <tbody> 
  <tr> 
   <td style="vertical-align:top;width:426.1pt;"> <p style="margin-left:0pt;">*&nbsp;*&nbsp;*&nbsp;*&nbsp;*&nbsp;/usr/local/bin/process_monitor.sh&nbsp;"/usr/local/jdk/bin/java&nbsp;kafka-manager"&nbsp;"/data/kafka/kafka-manager/bin/kafka-manager&nbsp;-Dconfig.file=/data/kafka/kafka-manager/conf/application.conf&nbsp;-Dhttp.port=8080&nbsp;&gt;&nbsp;/dev/null&nbsp;2&gt;&amp;1"</p> </td> 
  </tr> 
 </tbody> 
</table>

process\_monitor.sh下载：

[https://github.com/eyjian/libmooon/blob/master/shell/process\_monitor.sh][https_github.com_eyjian_libmooon_blob_master_shell_process_monitor.sh]

注意crontab的用户密码有效，crontab才能正常执行。

# **附6：kafka的安装** #

最基本的两个配置项为server.properties文件中的：

1) Broker.id

2) zookeeper.connect

其中broker.id每个节点要求不同，zookeeper.connect值建议指定目录，不要直接放在zookeeper根目录下。另外也建议设置listeners值，不然需要客户端配置hostname和IP的映射关系。

因broker.id和listeners的原因，每个节点的server.properties不一致，可利用工具set\_kafka\_id\_and\_ip.sh实现批量的替换，以简化kafka集群的部署。set\_kafka\_id\_and\_ip.sh下载地址：https://github.com/eyjian/libmooon/blob/master/shell/set\_kafka\_id\_and\_ip.sh。

crontab中启动kafka示例：

<table style="width:426.1pt;"> 
 <tbody> 
  <tr> 
   <td style="vertical-align:top;width:426.1pt;"> <p style="margin-left:0pt;">JMX_PORT=9999</p> <p style="margin-left:0pt;">*&nbsp;*&nbsp;*&nbsp;*&nbsp;*&nbsp;/usr/local/bin/process_monitor.sh&nbsp;"/usr/local/jdk/bin/java&nbsp;kafkaServer"&nbsp;"/data/kafka/bin/kafka-server-start.sh&nbsp;-daemon&nbsp;/data/kafka/config/server.properties"</p> </td> 
  </tr> 
 </tbody> 
</table>

设置JMX\_PORT是为方便kafka-manager管理kafka。

# **附7:\_\_consumer\_offsets** #

\_\_consumer\_offsets是kafka内置的Topic，在0.9.0.0之后的Kafka，将topic的offset 信息由之前存储在zookeeper上改为存储到内置的\_\_consumer\_offsets中。

server.properties中的配置项num.partitions和default.replication.factor对\_\_consumer\_offsets无效，而是受offsets.topic.num.partitions和offsets.topic.replication.factor两个控制。

[http_kafka.apache.org]: http://kafka.apache.org/
[http_kafka.apache.org_downloads]: http://kafka.apache.org/downloads
[https_cwiki.apache.org_confluence_display_KAFKA_Clients]: https://cwiki.apache.org/confluence/display/KAFKA/Clients
[https_github.com_edenhill_librdkafka]: https://github.com/edenhill/librdkafka
[https_github.com_confluentinc_confluent-kafka-go]: https://github.com/confluentinc/confluent-kafka-go
[http_kafka.apache.org_quickstart]: http://kafka.apache.org/quickstart
[https_github.com_eyjian_libmooon_blob_master_shell_process_monitor.sh]: https://github.com/eyjian/libmooon/blob/master/shell/process_monitor.sh
[https_github.com_eyjian_libmooon_releases]: https://github.com/eyjian/libmooon/releases
[https_github.com_eyjian_libmooon_blob_master_shell_set_kafka_id_and_ip.sh]: https://github.com/eyjian/libmooon/blob/master/shell/set_kafka_id_and_ip.sh
[https_github.com_eyjian_libmooon_blob_master_shell_set_hostname.sh]: https://github.com/eyjian/libmooon/blob/master/shell/set_hostname.sh
[https_github.com_yahoo_kafka-manager]: https://github.com/yahoo/kafka-manager