kafka分布式搭建

(192.168.230.129)master

(192.168.230.130)slave1

(192.168.230.131)salve2

在master、slave1、slave2三台主机上配置kafaka分布式集群

准备工作：在三台机器上配置好zookeeper

1、解压kafka压缩文件到指定目录下

[root@master software]# tar -zxf kafka_2.10-0.8.1.1.tgz -C /opt/modules

2、修改/opt/modules/kafka_2.10-0.8.1.1/config目录下的server.properties文件

[root@master ~]# cd /opt/modules/kafka_2.10-0.8.1.1/config
[root@master config]# vi server.properties

broker.id=1
zookeeper.connect=master:2181,slave1:2181,slave2:2181

3、将master主机上kafka配置好的安装目录发送到slave1、salve2主机上

[root@master config]# scp -r /opt/modules/kafka_2.10-0.8.1.1/ root@slave1:/opt/modules/kafka_2.10-0.8.1.1/

[root@master config]# scp -r /opt/modules/kafka_2.10-0.8.1.1/ root@slave2:/opt/modules/kafka_2.10-0.8.1.1/

4、在slave1主机中修改/opt/modules/kafka_2.10-0.8.1.1/config/server.properties配置文件

[root@slave1 kafka_2.10-0.8.2.1]# vi config/server.properties

broker.id=2

5、在slave1主机中修改/opt/modules/kafka_2.10-0.8.1.1/config/server.properties配置文件

[root@slave2 kafka_2.10-0.8.1.1]# vi config/server.properties

broker.id=3

6、修改三个主机的/opt/modules/kafka_2.10-0.8.1.1/config/server.properties配置文件中的

[root@master kafka_2.10-0.8.1.1]# vi config/server.properties

#host.name=localhost
将#注释去掉，并改成

master主机下：
host.name=master

[root@slave1 kafka_2.10-0.8.1.1]# vi config/server.properties

slave1主机下：
host.name=slave1

[root@slave2 kafka_2.10-0.8.1.1]# vi config/server.properties

slave2主机下：
host.name=slave2

7、在每一台节点上启动broker

在master主机上启动broker：

[root@master kafka_2.10-0.8.1.1]# ./bin/kafka-server-start.sh config/server.properties

在slave1主机上启动broker：

[root@slave1 kafka_2.10-0.8.1.1]# ./bin/kafka-server-start.sh config/server.properties

在slave2主机上启动broker：

[root@slave2 kafka_2.10-0.8.1.1]# ./bin/kafka-server-start.sh config/server.properties

8、在kafka集群中创建一个topic

在master主机中创建一个话题

[root@master kafka_2.10-0.8.1.1]# ./bin/kafka-topics.sh --create --zookeeper master:2181,slave1:2181,slave2:2181 --replication-factor 3 --partitions 3 --topic chinesescore

9、查看消息是否创建成功

[root@master kafka_2.10-0.8.1.1]# ./bin/kafka-topics.sh --list --zookeeper master:2181,slave1:2181,slave2:2181 --topic chinesescore

10、查看一个topic的分区及副本状态信息

[root@master kafka_2.10-0.8.1.1]# ./bin/kafka-topics.sh --describe --zookeeper master:2181,slave1:2181,slave2:2181 --topic chinesescore

11、用一个producer向某一个topic中写入消息

[root@master kafka_2.10-0.8.1.1]# ./bin/kafka-console-producer.sh --broker-list  master:9092,slave1:9092,slave2:9092 --topic chinesescore

12、用一个comsumer从某一个topic中读取信息

[root@slave2 kafka_2.10-0.8.1.1]# ./bin/kafka-console-consumer.sh --zookeeper master:2181,slave1:2181,slave2:2181 --from-beginning --topic chinesescore

13、从zookeeper中查看Topic

[root@slave1 kafka_2.10-0.8.1.1]# zkCli.sh

[zk: localhost:2181(CONNECTED) 4] ls /brokers/topics

13使用Java编写kafka生产者程序：

用一个comsumer从某一个topic中读取信息

[sparse@hadoop-sparse kafka_2.10-0.8.1.1]$ ./bin/kafka-console-consumer.sh --zookeeper hadoop-sparse:2181 --from-beginning --topic orders

14、关闭Kafka服务

[root@master kafka_2.10-0.8.1.1]# ./bin/kafka-server-stop.sh

[root@slave1 kafka_2.10-0.8.1.1]# ./bin/kafka-server-stop.sh

[root@slave2 kafka_2.10-0.8.1.1]# ./bin/kafka-server-stop.sh