topic_name：标识topic名称

进入生产者模式

可以推送数据

bin/kafka-console-producer.sh --topic=topic_name --broker-list localhost:9092

进入消费者模式

可以查看推送的数据

bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 -from-beginning --topic topic_name

查看所有topic

./bin/kafka-topics.sh --list --bootstrap-server 127.0.0.1:9092

删除topic

通过kafka直接删除topic

./bin/kafka-topics.sh --delete --bootstrap-server 127.0.0.1:9092 --topic topic_name

查看消费者组

查看消费者组下的数据情况

./kafka-consumer-groups.sh --bootstrap-server 127.0.0.1:9092 --describe --group topic_name

删除消费者组

./kafka-consumer-groups.sh --bootstrap-server 127.0.0.1:9092 --delete --group 组名称

再次使用该消费者组，且{"auto.offset.reset", "latest"}，则客户端从当前位置消费

发送文本数据到kafka

• data：待发送数据文本，可以多条，多条换行
• /dev/null：将返回值丢弃
• sync：是否同步执行，默认异步

脚本内容

cat data | /opt/qihoo/soc/kafka/bin/kafka-console-producer.sh \
--topic=topic_test_hou \
 --sync \
--broker-list localhost:9092 | > /dev/null

data样例

多行数据会被逐条发送

1.1.1.1;2.2.2.2;11;33;http
1.1.2.1;2.2.3.2;11;33;http
1.1.3.1;2.2.4.2;11;33;http
1.1.4.1;2.2.5.2;11;33;http
1.1.5.1;2.2.6.2;11;33;http

批量发送

#!/bin/bash

#消息总数,可根据需要配置
totalNum=10

for ((i=1; i<=${totalNum}; i ++))
do
{
    cat data | /opt/qihoo/soc/kafka/bin/kafka-console-producer.sh \
    --topic=topic_test_hou \
    --sync \
    --broker-list localhost:9092 | > /dev/null
}
sleep 1
done