Windows环境中Kafka的安装及使用_kafka-console-producer.bat

1. 简介

Kafka是一种高吞吐量的分布式发布——订阅消息系统。

2.环境搭建

2.1安装JDK（例如JDK1.8等）

2.2 安装Zookeeper

Zookeeper下载地址
下载后，解压放在目录D:\SoftWare\Kafka\zookeeper-3.4.14（本文所用的目录）下，关于zookeeper以及kafka的目录，路径中最好不要出现空格，比如D:\Program Files，尽量别用，运行脚本时会有问题。

（1）进入zookeeper的相关设置所在的文件目录，例如本文的：D:\SoftWare\Kafka\zookeeper-3.4.14\zookeeper-3.4.14\conf
（2）将"zoo_sample.cfg"重命名为"zoo.cfg"
（3）打开zoo.cfg(至于使用什么编辑器，根据自己喜好选即可)，找到并编辑
dataDir=/tmp/zookeeper ——> D:/SoftWare/Kafka/zookeeper-3.4.14/Data 或 D:\SoftWare\Kafka\zookeeper-3.4.14\Data
（注：路径要么是"/“分割，要么是转义字符”\"，这样会生成正确的路径(层级，子目录)）
（4）添加环境变量“D:\SoftWare\Kafka\zookeeper-3.4.14\zookeeper-3.4.14\bin”
（5）在zoo.cfg文件中修改默认的Zookeeper端口(默认端口2181)
（6）打开cmd窗口，输入zkserver，运行Zookeeper，运行结果如下：
恭喜，Zookeeper已经安装完成，已在2181端口运行。

2.3 安装Kafka

Kafka下载地址
要下载Binary downloads这个类型，不要下载源文件，这种方便使用。下载后，解压放在D:\SoftWare\Kafka\kafka_2.11-2.4.0目录下。

（1）进入kafka配置文件所在目录，D:\SoftWare\Kafka\kafka_2.11-2.4.0\config
（2）编辑文件"server.properties"，找到并编辑：
log.dirs=/tmp/kafka-logs ——> log.dirs=D:/SoftWare/Kafka/kafka_2.11-2.4.0/kafka-logs 或
log.dirs=D:\SoftWare\Kafka\kafka_2.11-2.4.0\kafka-logs
（注：路径要么是"/“分割，要么是转义字符”\"，这样会生成正确的路径(层级，子目录)）
（3）在server.properties文件中，zookeeper.connect=localhost:2181代表kafka所连接的zookeeper所在的服务器IP以及端口，可根据需要更改。本文在同一台机器上使用，故不用修改。
（4）kafka会按照默认配置，在9092端口上运行，并连接zookeeper的默认端口2181。

2.4 运行Kafka

提示：请确保启动kafka服务器前，Zookeeper实例已经在运行，因为kafka的运行是需要zookeeper这种分布式应用程序协调服务。

（1）进入kafka安装目录： D:\SoftWare\Kafka\kafka_2.11-2.4.0
（2）在地址栏中输入“cmd”然后回车，进入命令行 或者 按下shift+鼠标右键，选择"在此处打开命令窗口"，打开命令行。
（3）在命令行中输入：.\bin\windows\kafka-server-start.bat .\config\server.properties，然后回车

到目前为止，zookeeper以及kafka都已正确运行。保持运行状态，不要关闭。

重要（操作日志的处理）：
kafka启动后，如果你去查看kafka所在的根目录，或者是kafka本身的目录，会发现已经默认生成一堆操作日志(这样看起来真心很乱)：
而且会不断生成不同时间戳的操作日志。刚开始不知所措，一番研究后，看了启动的脚本内容，发现启动的时候是会默认使用到这个log4j.properties文件中的配置，而在zoo.cfg是不会看到本身的启动会调用到这个，还以为只有那一个日志路径：
在这里配置一下就可以了，找到config下的log4j.properties，将路径更改下即可，这样就可以归档在一个文件夹下边了，路径根据自己喜好定义：

另外如何消除不断生成日志的问题，就是同一天的不同时间会不停生成。
修改这里，还是在log4j.properties中：
本身都为trace，字面理解为会生成一堆跟踪日志，将其改为INFO即可。

2.5 创建主题

（1）创建主题，命名为"test0924"，replicationfactor=1(因为只有一个kafka服务器在运行)。可根据集群中kafka服务器个数来修改replicationfactor的数量，以便提高系统容错性等。
（2）在D:\SoftWare\Kafka\kafka_2.11-2.4.0\bin\windows目录下打开新的命令行；
（3）输入命令：kafka-topics.bat --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test0924，然后回车

该窗口可以关闭。

（4）查看创建的主题列表
kafka-topics.bat --list --zookeeper localhost:2181

2.6 创建生产者(producer)和消费者(consumer)

（1）在D:\SoftWare\Kafka\kafka_2.11-2.4.0\bin\windows目录下打开新的命令行；
（2）输入命令，启动producer：
kafka-console-producer.bat --broker-list localhost:9092 --topic test0924
（3）同样在该目录下打开新的命令行。输入命令，启动consumer：
kafka-console-consumer.bat --bootstrap-server localhost:9092 --topic test0924 --from-beginning
（4）现在生产者、消费者均已创建完成。在producer命令行窗口中任意输入内容，然后回车。在consumer命令行窗口中即可看到相应的内容。

2.8 关闭Kafka服务

（1）进入kafka安装目录： D:\SoftWare\Kafka\kafka_2.11-2.4.0
（2）在地址栏中输入“cmd”然后回车，进入命令行 或者 按下shift+鼠标右键，选择"在此处打开命令窗口"，打开命令行。
（3）在命令行中输入：.\bin\windows\kafka-server-stop.bat .\config\server.properties，然后回车

2.9 关闭Zookeeper服务

（1）进入kafka安装目录： D:\SoftWare\Kafka\kafka_2.11-2.4.0
（2）在地址栏中输入“cmd”然后回车，进入命令行 或者 按下shift+鼠标右键，选择"在此处打开命令窗口"，打开命令行。
（3）在命令行中输入：.\bin\windows\zookeeper-server-stop.bat .\config\server.properties，然后回车

3.常用的Kafka命令

#列出主题
kafka-topic.bat -list -zookeeper localhost:2181  

#描述主题  
kafka-topics.bat -describe -zookeeper localhost:2181 -topic [topic name]   

#从头读取消息
kafka-console-consumer.bat -zookeeper localhost:2181 -topic [topic name] -from-beginning 

#删除主题
kafka-run-class.bat kafka.admin.TopicCommand -delete -topic [topic_to_delete] -zookeeper localhost:2181  
#查看topic的详细信息 
./kafka-topic.sh -zookeeper localhost:2181 -describe -topic  [topic name]    

#为topic增加副本
./kafka-reassign-partitions.sh -zookeeper localhost:2181 -reassignment-json-file json/partitions-to-move.json -execute  

#为topic增加partition
./bin/kafka-topics.sh -zookeeper localhost:2181 -alter -partitions 20 -topic [topic name]   

#下线broker
./kafka-run-class.sh kafka.admin.ShutdownBroker --zookeeper localhost:2181 broker [brokerId] --num.retries 3 --retry.interval.ms 60  shutdown broker 

#luanch console producer, and specifiy the parse key, the key and value is splited by blank symbol
kafka-console-producer.sh --broker-list server-1:9092,server-2:9092,server-3:9092 --topic kafka-action --property parse.key=true --property key.separator=' '

#alter topic    config-test ,  sets the property segment.bytes to 200MB
kafka-topics.sh --alter --zookeeper server-1:2181,server-2:2181,server-3:2181  --topic config-test --config segment.bytes=209715200

#alter topic    config-test ,  delete the property config segment.bytes
kafka-topics.sh --alter --zookeeper server-1:2181,server-2:2181,server-3:2181  --topic config-test --delete-config segment.bytes=209715200

＃查看主题当前己覆盖的配置
kafka-topics.sh --alter --zookeeper server-1:2181,server-2:2181,server-3:2181  --topics-with-overrides --topic config-test 

#view the partitions with status under replicated
kafka-topics.sh --describe --zookeeper server-1:2181,server-2:2181,server-3:2181  --under-replicated  partitions

#查看(某个特定)主题的哪些分区的Leader 己不可用
kafka-topics . sh --describe --zookeeper server-1:2181,server-2:2181,server-3:2181  --unavailablepartitions   [--topic  {topic_name}]

#查看某个主题各分区对应消息偏移量,time 参数表示查看在指定时间之前的数据，支持-1(latest),-2 (earliest) 两个时间选项，默认取值为-l 。
kafka-run-class.sh kafka . tools.GetOffsetShell --broker - list server-1:9092,server-2:9092,server-3:9092 --topic kafka- action --time -1

#查看kafka日志消息,
#files 是必传参数，用于指定要转储（ dump ）文件的路径， 可同时指定多个文件，多个文件路径之间以逗号分隔。
kafka-run- class.sh kafka.tools.DumpLogSegments --files /opt/data/kafka-logs/producer-create-topic-0/00000000000000000000.log  --print-data-log

#性能测试工具，向一个名为producer-perιtest的主题发送100 万条消息，每条消息大小为1000字节，
kafka-producer-perf-test.sh --num-records 1000000 --record-size 1000  --topic producer-perf-test --throughput 1000000 \
--producer-props  bootstrap.servers=server-1:9092,server-2:9092,server-3:9092 acks=all

#老版本consumer消费kafka-action数据,offset保存在zookeeper中，并在结束时删除offsets
kafka-console-consurner.sh --zookeeper server-1:2181,server-2:2181,server-3:2181 --topic kafka-action --consumer-property group.id=old-consumer-test  --consumer-property consumer.id=old-consumer-cl  \--from-beginning  --delete-consumer-offsets

#新版本consumer消费kafka-action数据，默认offset保存在kafka的__consumer-offsets内部主题中
#若以参数bootstrap-server方式启动，则默认调用的是新版消费者，此时可以不用设置new-consumer 参数
kafka-console consumer.sh -bootstrap-server server-1:9092,server-2:9092,server-3:9092 --new-consumer --consumer-property group.id=new-consumer-test \
--consumer-property client.id=new-consumer-cl --topic kafka-action

#查看主题ka:fka-action 各分区的偏移量信息
#time 参数表示查看在指定时间之前的数据，支持-1(latest),-2 (earliest) 两个时间选项，默认取值为-l 。
kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list server-1:9092,server-2:9092,server-3:9092 --topic kafka-action -time -1

#查看__consumer-offsets主题编号为6的分区的信息
kafka-simple-consumer-shell.sh --topic __consumer_offsets --partition 6 --broker-list server-1:9092,server-2:9092,server-3:9092 --formatter
"kafka.coordinator.GroupMetadataManager\$0ffsetsMessageFormatter"

#同时消费多个主题
kafka-console-consumer.sh --bootstrap-server server-1:9092,server-2:9092,server-3:9092 --new-consumer --consumer-property group.id=consume-multi-topic 
--whitelist  "kafka-action|producer-perf-test"

#查看消费组的消费偏移量，如果待查询消费组的状态维Dead,则无法查看到
kafka-consumer-groups.sh --bootstrap-server  server-1:9092,server-2:9092,server-3:9092 --describe --group consumer-offset-test  --new-consumer

#删除消费组
kafka-consumer-groups.sh --zookeeper server-1:2181,server-2:2181,server-3:2181  --delete--group old-consumer-test

#消费者性能测试工具, 还可以通过--consumer.config加载property文件来设置属性
kafka-consumer-perf-test.sh --broker-list server-1:9092,server-2:9092,server-3:9092 --threads 5 --messages 1000000 --message-size 1000 \
--num-fetch-threads 2  --group consumer-perf-test --topic producer-perf-test --new-consumer

#查看指定主题所有的覆盖配置
kafka-configs.sh --zookeeper server-1:2181,server-2:2181,server-3:2181   --describe-entity-type topics --entity-name config-test 

#增加特定主题的配置，如果有多个配置，以逗号分隔如--alter --add-config flush.messages=2,max.message.bytes=102400
kafka-configs.sh --zookeeper server-1:2181,server-2:2181,server-3:2181   --describe-entity-type topics --entity-name config-test  --alter \
--add-config flush.messages=2


#删除配置
kafka-configs.sh --zookeeper server-1:2181,server-2:2181,server-3:2181   --describe-entity-type topics --entity-name config-test   --alter --delete-config flush.messages,max.message.bytes

#broker限流:对server-1对应的代理(broker.id=1)上分布的Leader副本和Follower副本的复制速率控制为10MB/s
kafka-configs.sh --zookeeper server-1:2181,server-2:2181,server-3:2181  --entity-type brokers --entity-name 1 --alter --add-config follower.replication.throttled.rate=l0485760,leader.replication.throttled.rate=l0485760

#boker限流:查看broker-1的限制配置信息
kafka-configs.sh --zookeeper server-1:2181,server-2:2181,server-3:2181  --entity-type brokers  --entity-name 1  --describe

#broker限流:删除配置
kafka-configs.sh --zookeeper server-1:2181,server-2:2181,server-3:2181  --entity-type brokers --entity-name 1 --alter --delete-config follower.replication.throttled.rate,leader.replication.throttled.rate

#主题级别限流
kafka-configs.sh --zookeeper server-1:2181,server-2:2181,server-3:2181 --entity-type topics --entity-name kafka-action --alter \--add-config  leader.replication.throttled.replicas=[O:l,1:2,2:3],follower.replication.throttled.replicas=[0:2,1:3,2:1]

#客户端流控:为用户lenmom添加限流,前提:kafka集群添加了认证机制，否则不可用此法
kafka-configs.sh --zookeeper  server-1:2181,server-2:2181,server-3:2181 --alter --add-config  'producer_byte_rate=l024,consumer_byte_ rate=2048'  \
--entity-type users  --entity-name lenmom

#客户端流控:查看用户流控信息
kafka-configs.sh --zookeeper server-1:2181,server-2:2181,server-3:2181    describe --entity-type users

#客户端流控:在producer/consumer的clientid等于acl-client时,该配置将会生效
kafka-configs.sh --zookeeper localhost:2181 --alter --add-config 'producer_byte_rate=l024 , consumer_byte_rate=2048'   --entity type clients \
--entity-name acl-client

#客户端流控:为用户lenmom的客户端user-client-config添加流控
kafka-configs.sh --zookeeper  server-1:2181,server-2:2181,server-3:2181  --alter --add-config  'producer byte_rate=1024,consumer_byte_rate=2048' \
 --entity-type users --entity-name lenmom --entity-type clients --entity-name user-client-config

#leader平衡,对topic:kafka-action的分区1进行再平衡 
#echo '{"partitions": [{"topic":"kafka-action","partition": 1}]}' >partitions-leader-election.json  
kafka-preferred-replica-election.sh --zookeeper server-1:2181,server-2:2181,server-3:2181 --path-to-json-file   partitions-leader-election.json

#节点下线&分区迁移, 下线broker2
#topics-to-move.json:  其中,version为固定值
#{"topics":[{"topic":"reassign-partitions"}],"version":1} 
kafka-reassign-partitions.sh --zookeeper server-1:2181,server-2:2181,server-3:2181  --topics-to-move-json-file  ../config/topics-to-move.json \
--broker-list " 1,3"   --generate


#增加指定主题的分区数
kafka-topics.sh   --alter  --zookeeper server-1:2181,server-2:2181,server-3:2181 --partitions 6 --topic kafka-action

#增加指定主题的副本数，需要事先准备副本分配json文件
kafka-reassign-partitions.sh --zookeeper server-1:2181,server-2:2181,server-3:2181  --reassignment-json-file .. /config/replica-extends.json --execute

#查看分区副本重分配执行状态
kafka-reassign-partitions.sh --zookeeper server-1:2181,server-2:2181,server-3:2181  --reassignment-json-file .. /config/replica-extends.json --verify
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139