kafka分布式搭建

(192.168.230.129)master

(192.168.230.130)slave1

(192.168.230.131)salve2

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

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

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

[[email protected] 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文件

[[email protected] ~]# cd /opt/modules/kafka_2.10-0.8.1.1/config
[[email protected] config]# vi server.properties

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

3、将master主机上kafka配置好的安装文件夹发送到slave1、salve2主机上

[[email protected] config]# scp -r /opt/modules/kafka_2.10-0.8.1.1/ [email protected]:/opt/modules/kafka_2.10-0.8.1.1/

[[email protected] config]# scp -r /opt/modules/kafka_2.10-0.8.1.1/ [email protected]:/opt/modules/kafka_2.10-0.8.1.1/

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

[[email protected] 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配置文件

[[email protected] 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配置文件里的

[[email protected] kafka_2.10-0.8.1.1]# vi config/server.properties

#host.name=localhost
将#凝视去掉，并改成

master主机下：
host.name=master

[[email protected] kafka_2.10-0.8.1.1]# vi config/server.properties

slave1主机下：
host.name=slave1

[[email protected] kafka_2.10-0.8.1.1]# vi config/server.properties

slave2主机下：
host.name=slave2

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

在master主机上启动broker：

[[email protected] kafka_2.10-0.8.1.1]# ./bin/kafka-server-start.sh config/server.properties

在slave1主机上启动broker：

[[email protected] kafka_2.10-0.8.1.1]# ./bin/kafka-server-start.sh config/server.properties

在slave2主机上启动broker：

[[email protected] kafka_2.10-0.8.1.1]# ./bin/kafka-server-start.sh config/server.properties

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

在master主机中创建一个话题

[[email protected] 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、查看消息是否创建成功

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

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

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

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

[[email protected] 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中读取信息

[[email protected] 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

[[email protected] kafka_2.10-0.8.1.1]# zkCli.sh

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

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

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

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

14、关闭Kafka服务

[[email protected] kafka_2.10-0.8.1.1]# ./bin/kafka-server-stop.sh

[[email protected] kafka_2.10-0.8.1.1]# ./bin/kafka-server-stop.sh

[[email protected] kafka_2.10-0.8.1.1]# ./bin/kafka-server-stop.sh