Kafka 读写数据
参考文档
Apache Kafka
常见命令
kafka-topics.sh #消息的管理命令
kafka-console-producer.sh #生产者的模拟命令
kafka-console-consumer.sh #消费者的模拟命令
创建 Topic
创建topic名为 chen,partitions(分区)为3,replication(每个分区的副本数/每个分区的分区因子)为 2
#新版命令
[root@node1 bin]#/usr/local/kafka/bin/kafka-topics.sh --create --topic chen --bootstrap-server 10.0.0.187:9092 --partitions 3 --replication-factor 2
#在各节点上观察生成的相关数据
[root@node1 ~]#ls /usr/local/kafka/data/
[root@node2 ~]#ls /usr/local/kafka/data/
[root@node3 ~]#ls /usr/local/kafka/data/
#旧版命令
[root@node1 ~]#/usr/local/kafka/bin/kafka-topics.sh --chen --zookeeper 10.0.0.187:2181,10.0.0.188:2181,10.0.0.189:2181 --partitions 3 --replication-factor 2 --topic wang
Created topic wang.
获取所有 Topic
#新版命令
[root@node1 bin]#/usr/local/kafka/bin/kafka-topics.sh --list --bootstrap-server 10.0.0.187:9092
chen
#旧版命令
[root@node1 ~]#/usr/local/kafka/bin/kafka-topics.sh --list --zookeeper 10.0.0.101:2181,10.0.0.102:2181,10.0.0.103:2181
chen
验证 Topic 详情
状态说明:wang 有三个分区分别为0、1、2,分区0的leader是3 (broker.id),分区 0 有2 个副本,并且状态都为 lsr(ln-sync,表示可 以参加选举成为 leader)。
[root@node1 bin]#/usr/local/kafka/bin/kafka-topics.sh --describe --bootstrap-server 10.0.0.187:9092 --topic chen
Topic: chen TopicId: OHzFQnjYTYS_t-PyomxkSQ PartitionCount: 3 ReplicationFactor: 2 Configs:
Topic: chen Partition: 0 Leader: 3 Replicas: 3,1 Isr: 3,1 Elr: N/A LastKnownElr: N/A
Topic: chen Partition: 1 Leader: 1 Replicas: 1,2 Isr: 1,2 Elr: N/A LastKnownElr: N/A
Topic: chen Partition: 2 Leader: 2 Replicas: 2,3 Isr: 2,3 Elr: N/A LastKnownElr: N/A
[root@node1 bin]#ls -1 /usr/local/kafka/data/
chen-0
chen-1
[root@node2 ~]#ls -1 /usr/local/kafka/data/
chen-1
chen-2
[root@node3 ~]#ls -1 /usr/local/kafka/data/
chen-0
chen-2
node3 leader | node1 leader | node2 leader |
---|---|---|
node1 follwer | node2 follwer | node3 follwer |
p0 | p1 | p2 |
生产 Topic
kafka-console-producer.sh 格式
#发送消息命令格式:
kafka-console-producer.sh --broker-list <kafkaIP1>:<端口>,<kafkaIP2>:<端口> --topic <topic名称>
#/usr/local/kafka/bin/kafka-console-producer.sh --broker-list 10.0.0.187:9092,10.0.0.102:9092,10.0.0.103:9092 --topic chen
范例:
/usr/local/kafka/bin/kafka-console-producer.sh --broker-list 10.0.0.187:9092 --topic chen
>message1
>message2
>message3
>
#或者下面方式
[root@node1 ~]#/usr/local/kafka/bin/kafka-console-producer.sh --topic wang --bootstrap-server 10.0.0.101:9092
消费 Topic
kafka-console-consumer.sh 格式
#接收消息命令格式:
kafka-console-consumer.sh --bootstrap-server <host>:<post> --topic <topic名称> --from-beginning --consumer
property group.id=<组名称>
注意:
-
消息者先生产消息,消费都后续才启动,也能收到之前生产的消息
-
同一个消息在同一个group内的消费者只有被一个消费者消费,比如:共100条消息,在一个group内有A,B两个消费者,其中A消费 50条,B消费另外的50条消息。从而实现负载均衡,不同group内的消费者则可以同时消费同一个消息--from-beginning
-
表示消费发布的消息也能收到,默认只能收到消费后发布的新消息
范例:
#交互式持续接收消息,按Ctrl+C退出
/usr/local/kafka/bin/kafka-console-consumer.sh --topic chen --bootstrap-server 10.0.0.187:9092 --from-beginning
#一个消息同时只能被同一个组内一个消费者消费(单播机制),实现负载均衡,而不能组可以同时消费同一个消息(多播机制)
[root@node2 ~]#/usr/local/kafka/bin/kafka-console-consumer.sh --topic chen --bootstrap-server 10.0.0.102:9092 --from-beginning --consumer-property group.id=group1
[root@node2 ~]#/usr/local/kafka/bin/kafka-console-consumer.sh --topic chen --bootstrap-server 10.0.0.102:9092 --from-beginning --consumer-property group.id=group1
删除 Topic
范例:
#注意:需要修改每个节点配置文件server.properties中的delete.topic.enable=true并重启
#新版本
[root@node3 ~]#/usr/local/kafka/bin/kafka-topics.sh --delete --bootstrap-server 10.0.0.101:9092,10.0.0.102:9092,10.0.0.103:9092 --topic chen
#旧版本
[root@node3 ~]#/usr/local/kafka/bin/kafka-topics.sh --delete --zookeeper
10.0.0.101:2181,10.0.0.102:2181,10.0.0.103:2181 --topic chen
Topic wang is marked for deletion.
Note: This will have no impact if delete.topic.enable is not set to true.
范例:删除zk下面 topic test
#无需修改配置文件server.properties,此方法很危险
[root@zookeeper-node1 ~]#zkCli.sh -server 10.0.0.103:2181
[zk: 10.0.0.103:2181(CONNECTED) 0] ls /brokers/topics
[zk: 10.0.0.103:2181(CONNECTED) 0] deleteall /brokers/topics/test
[zk: 10.0.0.103:2181(CONNECTED) 0] ls /brokers/topics