启动kafka集群

首先要启动好kafka集群
1、集群时间同步
2、启动zookeeper集群
3、启动kafka集群
启动kafka集群的方式就是在集群中每台机器kafka目录下运行
nohup bin/kafka-server-start.sh config/server.properties >/dev/null 2>&1 &

创建topic

kafka-topics.sh

kafka发出消息和接收消息都是基于topic，所以要先创建一个topic，才能向里面发消息。创建topic的脚本：

直接运行kafka-topics.sh或者kafka-topics.sh --help可以看到帮助。

./kafka-topics.sh --create --partitions 3 --replication-factor 2 --topic test --zookeeper node-1:2181,node-2:2181,node-3:2181

--create 表示创建topic
--partitions 表示指定我们topic的分区数
--replication-factor 指定副本数
--topic 指定topic的名字
--zookeeper 指定zookeeper的地址
topic的信息保存在zookeeper中

topic创建好了，就可以向里边发送消息了。

发送消息

kafka-console-producer.sh

通过命令行实现数据的发送 producer 生产者
kafka-console-producer.sh就是用来测试用的脚本，可以模拟kafka消息的发送端。
直接运行kafka-console-producer.sh查看帮助

REQUIRED的选项是必须要有的。

 ./kafka-console-producer.sh --broker-list node-1:9092,node-2:9092,node-3:9092 --topic test

在server.properties文件中的配置，默认监听端口是9092

--broker-list 指定我们kafka集群的地址
--topic 指定我们的消息发送到哪个topic里面去

输入命令后回车，会出现一个尖括号。在后边输入信息回车，就会发送出去输入的信息。

接受消息

kafka-console-consumer.sh

通过命令行实现数据的接收 consumer 消费者

./kafka-console-consumer.sh --bootstrap-server node-1:9092,node-2:9092,node-3:9092 --from-beginning --topic test

--bootstrap-server 表示我们的kafak集群的地址，在旧版本中使用的是--zookeeper参数，两者至少使用一个
--from-beginning 表示我们从最开始的数据进行消费
--topic指定我们topic的名字

在producer端发送数据，在consumer端可以收到数据

node-1

node-2
后边两个consumer里有些信息是我之前发送的，所以在node-1的producer中没有显示。这里，在node-1和node-2上都启动了consumer，由于设置了参数--from-beginning，所以之前发送过的所有的数据都接受到了，而且顺序不一样，注意吃饭了吗那一行消息。但是所有的消息都接受到了，说明这些消息都保存在kafka集群中，而且支持中文。