1.启动Kafka
kafka-server-start.sh -daemon /kafka/config/server.properties
2.停止Kafka
kafka-server-stop.sh
3.创建topic
topic名是clotho，分区6个，副本3个
kafka-topics.sh --bootstrap-server localhost:9092 --topic clotho --create --partitions 6 --replication-factor 3
4.删除topic
topic名是clotho
kafka-topics.sh --bootstrap-server localhost:9092 --topic clotho --delete
5.列出所有topic
kafka-topics.sh --bootstrap-server localhost:9092 --list
6.显示所有topic的状态信息
kafka-topics.sh --bootstrap-server localhost:9092 --describe
7.显示指定topic的状态信息
topic名是clotho
kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic clotho
8.监听topic
只显示开始监听后新接收的数据
kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic clotho
显示历史数据（从第一条数据开始）和新接收的数据
kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic clotho --from-beginning
9.列出所有消费组
kafka-consumer-groups.sh --bootstrap-server localhost:9092 --list
10.显示指定消费组的状态信息
kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group sinensis --describe
11.压测生产数据
kafka-producer-perf-test.sh bootstrap.servers=localhost:9092 --topic clotho --num-records 100000 --record-size 1000 --throughput 2000
参数：
|
参数 |
说明 |
|
messages |
生产者发送总的消息数量 |
|
message-size |
每条消息大小 |
|
batch-size |
每次批量发送消息的数量 |
|
topics |
生产者发送的topic |
|
threads |
生产者使用几个线程同时发送 |
|
broker-list |
安装kafka服务的机器ip:port列表 |
|
producer-num-retries |
一个消息失败发送重试次数 |
|
request-timeout-ms |
一个消息请求发送超时时间 |
12.压测消费数据
kafka-consumer-perf-test.sh bootstrap.servers=localhost:9092 --topic clotho fetch-size 1048576 --messages 100000 --threads 3
参数：
|
参数 |
说明 |
|
zookeeper |
zookeeper端口配置 |
|
messages |
消费者消费消息总数量 |
|
topic |
消费者需要消费的topic |
|
threads |
消费者使用几个线程同时消费 |
|
group |
消费者组名称 |
|
socket-buffer-sizesocket |
缓冲大小 |
|
fetch-size |
每次向kafka broker请求消费大小 |
|
consumer.timeout.ms |
消费者去kafka broker拿去一条消息超时时间 |