一、Kafka的下载与解压
http://kafka.apache.org/downloads.html下载kafka_2.11-1.1.1.tgz.gz并解压到/home/jun下
[jun@master ~]$ cd kafka_2.11-1.1.1/ [jun@master kafka_2.11-1.1.1]$ ls -l total 48 drwxr-xr-x. 3 jun jun 4096 Jul 7 12:15 bin drwxr-xr-x. 2 jun jun 4096 Jul 7 12:15 config drwxr-xr-x. 2 jun jun 4096 Jul 23 16:32 libs -rw-r--r--. 1 jun jun 28824 Jul 7 12:12 LICENSE -rw-r--r--. 1 jun jun 336 Jul 7 12:12 NOTICE drwxr-xr-x. 2 jun jun 44 Jul 7 12:15 site-docs
二、配置Kafka集群
1.在Master节点上的配置
配置server.properties
[jun@master kafka_2.11-1.1.1]$ gedit /home/jun/kafka_2.11-1.1.1/config/server.properties
（1）在Server Basics部分增加下面的配置：将master作为broker，其id采用了默认的0
############################# Server Basics ############################# # The id of the broker. This must be set to a unique integer for each broker. broker.id=0 host.name=master
listeners=PLAINTEXT://master:9092
（2）在Zookeeper部分增加下面的配置，Zookeeper作为协调器，它连接的节点包括了集群的所有计算机。
############################# Zookeeper ############################# # Zookeeper connection string (see zookeeper docs for details). # This is a comma separated host:port pairs, each corresponding to a zk # server. e.g. "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002". # You can also append an optional chroot string to the urls to specify the # root directory for all kafka znodes. zookeeper.connect=master:2181,slave0:2181,slave1:2181 # Timeout in ms for connecting to zookeeper zookeeper.connection.timeout.ms=6000
将Kafka文件复制Slave节点
[jun@master kafka_2.11-1.1.1]$ scp -r /home/jun/kafka_2.11-1.1.1/ slave0:~/ [jun@master kafka_2.11-1.1.1]$ scp -r /home/jun/kafka_2.11-1.1.1/ slave1:~/
2.在Slave节点上的配置
同理，将slave0的broker.id设置为1，host.name=slave0，listeners=PLAINTEXT://slave0:9092；将slave1的broker.id设置为2，host.name=slave1，listeners=PLAINTEXT://slave1:9092即可。
三、Kafka的应用
1.启动Kafka自带的Zookeeper服务
首先查看当前系统进程列表
[jun@master ~]$ jps 3070 Jps [jun@slave0 ~]$ jps 3141 Jps [jun@slave1 ~]$ jps 3114 Jps
启动Kafka自带的Zookeeper服务
[jun@master kafka_2.11-1.1.1]$ bin/zookeeper-server-start.sh config/zookeeper.properties
[jun@slave0 kafka_2.11-1.1.1]$ bin/zookeeper-server-start.sh config/zookeeper.properties
[jun@slave1 kafka_2.11-1.1.1]$ bin/zookeeper-server-start.sh config/zookeeper.properties
另开一个终端，查看此时的系统进程，此时QuorumPeerMain就是Zookeeper服务器进程
[jun@master kafka_2.11-1.1.1]$ jps 3588 Jps 3147 QuorumPeerMain [jun@slave0 kafka_2.11-1.1.1]$ jps 3891 Jps 3547 QuorumPeerMain [jun@slave1 kafka_2.11-1.1.1]$ jps 3201 QuorumPeerMain 3538 Jps
2.启动Kafka服务
[jun@master kafka_2.11-1.1.1]$ bin/kafka-server-start.sh config/server.properties [jun@slave0 kafka_2.11-1.1.1]$ bin/kafka-server-start.sh config/server.properties [jun@slave1 kafka_2.11-1.1.1]$ bin/kafka-server-start.sh config/server.properties
另开一个终端，查看此时的系统进程，此时的Kafka就是Kafka的服务器进程
[jun@master kafka_2.11-1.1.1]$ jps 4064 Jps 3687 Kafka 3147 QuorumPeerMain [jun@slave0 kafka_2.11-1.1.1]$ jps 4021 Kafka 3547 QuorumPeerMain 4398 Jps [jun@slave1 kafka_2.11-1.1.1]$ jps 3201 QuorumPeerMain 3682 Kafka 4059 Jps
3.创建主题
· 主题（Topic）是消息中间件的基本概念，相当于文件系统的目录，其实就是用于保存消息内容的计算实体，通过主题名加以标识，就如同目录通过目录名标识一样。
（1）在master节点上创建一个名称为“test”的主题
[jun@master kafka_2.11-1.1.1]$ bin/kafka-topics.sh --create -zookeeper master:2181 --replication-factor 1 --partitions 1 --topic test Created topic "test".
（2）查看已经创建的消息主题，同样，在其他Slave结点上执行该命令也可以查看到创建的主题
[jun@master kafka_2.11-1.1.1]$ bin/kafka-topics.sh --list --zookeeper master:2181 test
4.发送消息
消息中间件是一个用于接收消息并转发消息的服务，为了检验Kafka是否能够正常工作，需要创建一个消息生产者（producer）、利用它产生消息
[jun@master kafka_2.11-1.1.1]$ bin/kafka-console-producer.sh --broker-list master:9092 --topic test >say hi [2018-07-23 20:14:28,119] WARN [Producer clientId=console-producer] Error while fetching metadata with correlation id 1 : {myfinaltest=LEADER_NOT_AVAILABLE} (org.apache.kafka.clients.NetworkClient) >hi >you >cnm >
5.接收消息
在一个新终端上执行如下的命令，可以看到接收到了producer发送的消息
[jun@slave0 kafka_2.11-1.1.1]$ bin/kafka-console-consumer.sh --bootstrap-server slave0:9092 --topic test--from-beginning say hi hi you cnm [jun@slave1 kafka_2.11-1.1.1]$ bin/kafka-console-consumer.sh --bootstrap-server slave0:9092 --topic test--from-beginning say hi hi you cnm