kafka的使用

1.为了方便启动kafka，我们在kafka的bin目录下写2个脚本。一个是启动zookeeper集群的脚本，一个是启动kafka集群的脚本

1.1 vim start-zks.sh  启动zookeeper集群的脚本

1.1.1给该文件赋权限

        chmod 777 start-zks.sh

1.1.2 编写内容

#!/bin/bash
echo "zhangqi 上的zookeeper 正在启动"
ssh root@zhangqi "source /etc/profile;/root/Downloads/zookeeper-3.4.5/bin/zkServer.sh start"
echo "zhangqi 上的zookeeper 启动完成"

echo "hadoop02 上的zookeeper 正在启动"
ssh root@hadoop02 "source /etc/profile;/root/Downloads/zookeeper-3.4.5/bin/zkServer.sh start"
echo "hadoop02 上的zookeeper 启动完成"

echo "hadoop03 上的zookeeper 正在启动"
ssh root@hadoop03 "source /etc/profile;/root/Downloads/zookeeper-3.4.5/bin/zkServer.sh start"
echo "hadoop03 上的zookeeper 启动完成"

1.2 vim start-brokers.sh 启动kafka集群的脚本

1.2.1 给该文件赋权限

        chmod 777 start-brokers.sh

1.2.2 编写内容

#!/bin/bash
echo "zhangqi 上的kafka 正在启动..."
ssh root@zhangqi "source /etc/profile;/root/Downloads/kafka_2.11-2.0.0/bin/kafka-server-start.sh -daemon  /root/Downloads/kafka_2.11-2.0.0/config/server.properties"
echo "zhangqi 上的kafka 启动完成"

echo "hadoop02 上的kafka 正在启动..."
ssh root@hadoop02 "source /etc/profile;/root/Downloads/kafka_2.11-2.0.0/bin/kafka-server-start.sh -daemon  /root/Downloads/kafka_2.11-2.0.0/config/server.properties"
echo "hadoop02 上的kafka 启动完成"

echo "hadoop03 上的kafka 正在启动..."
ssh root@hadoop03 "source /etc/profile;/root/Downloads/kafka_2.11-2.0.0/bin/kafka-server-start.sh -daemon  /root/Downloads/kafka_2.11-2.0.0/config/server.properties"
echo "hadoop03 上的kafka 启动完成"

2.使用

2.1 在bin目录下启动kafka，在后台启动

./kafka-server-start.sh -daemon  ../config/server.properties 

2.2 创建topic，在bin目录下

./kafka-topics.sh --create --topic t1807a1 --partitions 2 --replication-factor 2 --zookeeper zhangqi:2181

2.3 查看分区、副本在哪

./kafka-topics.sh --describe --topic t1807a1 --zookeeper zhangqi:2181

2.4 生产者

./kafka-console-producer.sh --broker-list zhangqi:9092 --topic test

2.5 消费者

./kafka-console-consumer.sh --bootstrap-server zhangqi:9092 --topic test

      消费者可以接收，生成者生成的数据
2.6 消费者可以接收，生成者生成的数据，如果不指定 --from-beginning  那么消费者消费的是最新的数据。指定了之后也可以消费以前的数据

      消费者，可以消费之前产生的数据

./kafka-console-consumer.sh --bootstrap-server zhangqi:9092 --topic test --from-beginning