集群管理

前台启动broker

1
bin/kafka-server-start.sh <path>/server.properties

Ctrl + C 关闭

后台启动broker

1
bin/kafka-server-start.sh -daemon <path>/server.properties

后台启动broker并开放JMX端口

1
JMX_PORT=9999 bin/kafka-server-start.sh -daemon <path>/server.properties

关闭broker

1
bin/kafka-server-stop.sh

Topic管理

创建topic

1
bin/kafka-topics.sh --create --zookeeper localhost:2181 --partitions 3 --replication-factor 3 --topic topicname

删除topic

(需要server.properties中设置delete.topic.enable=true否则只是标记删除或者直接重启)

1
bin/kafka-topics.sh --delete --zookeeper localhost:2181 --topic topicname

查询topic列表

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

查询topic详情

1
bin/kafka-topics.sh --zookeeper localhost:2181 --describe --topic topicname

修改topic

1
bin/kafka-topics.sh --alter --zookeeper localhost:2181 --partitions 6 --topic topicname

Consumer-Groups管理

查询消费者组

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

查询消费者组详情

1
bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group groupname

重设消费者组位移

1
2
3
4
5
6
7
8
最早处
bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group groupname --reset-offsets --all-topics --to-earliest --execute
最新处
bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group groupname --reset-offsets --all-topics --to-latest --execute
某个位置
bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group groupname --reset-offsets --all-topics --to-offset 2000 --execute
调整到某个时间之后得最早位移
bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group groupname --reset-offsets --all-topics --to-datetime 2019-09-15T00:00:00.000

删除消费者组

1
bin/kafka-consumer-groups.sh --zookeeper localhost:2181 --delete --group groupname

脚本工具

producer脚本

1
2
3
4
5
6
7
bin/kafka-console-producer.sh --broker-list localhost:9092 --topic topicname 
参数含义：
--compression-codec lz4  压缩类型
--request-required-acks all acks的值
--timeout 3000  linger.ms的值
--message-send-max-retries 10   retries的值
--max-partition-memory-bytes batch.size值

consumer脚本

1
2
3
4
5
6
7
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic topicname --from-beginning
指定groupid
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic topicname --from-beginning
--consumer-property group.id=old-consumer-group
指定分区
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic topicname --from-beginning
--partition 0

kafka-run-class脚本

1
2
kafka-run-class.sh kafka.tools.ConsoleConsumer   就是 kafka-console-consumer.sh
kafka-run-class.sh kafka.tools.ConsoleProducer   就是 kafka-console-producer.sh

获取topic当前消息数

1
kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list localhost:9092 --topic topicname --time -1

–time -1表示最大位移 –time -2表示最早位移

查询_consumer_offsets

1
bin/kafka-simple-consumer-shell.sh --topic _consumer_offsets --partition 12 --broker-list localhost:9092 --formatter "kafka.coorfinator.GroupMetadataManager\$OffsetsMessageFormatter"

MirrorMaker

跨机房灾备工具

1
bin/kafka-mirror-maker.sh --consumer.config consumer.properties --producer.config producer.properties --whitelist topicA|topicB

Zookeeper常用命令

查看znode中的内容

1
ls  /

创建普通的节点 

1
create

获得节点的信息

1
get

创建临时节点 

1
create -e

编号节点： 

1
create -s

删除一个节点 

1
delete

递归删除节点

1
rmr

修改节点内容

1
set

监听节点 

1
get /test watch

在其他节点进行修改 

1
set /test 555

监听节点上收到WatchedEvent state:SyncConnected type:NodeDataChanged path:/test