计划使用三台主机：11.12.112.206、11.12.112.207、11.12.112.208搭建Kafka集群环境，
使用的zookeeper集群为：11.12.112.215:2181,11.12.112.216:2181,11.12.112.217:2181
并且zookeeper集群已经启动成功；
1.下载压缩包：
kafka_2.11-0.8.2.2.tgz
并将之放在某个目录下，例如：/home/jason/bigdata
2.解压缩到当前目录下：
tar -xvf kafka_2.11-0.8.2.2.tgz
3.进入Kafka的目录
[root@clsserv206 bigdata]# cd kafka_2.11-0.8.2.2
[root@clsserv206 kafka_2.11-0.8.2.2]# ll
总用量 28
drwxr-xr-x. 3 root root 4096 9月 3 2015 bin
drwxr-xr-x. 2 root root 4096 9月 3 2015 config
drwxr-xr-x. 2 root root 4096 9月 3 2015 libs
-rw-r--r--. 1 root root 11358 9月 3 2015 LICENSE
-rw-r--r--. 1 root root 162 9月 3 2015 NOTICE
4.修改Kafka的配置文件
4.1.配置文件路径
配置文件位于kafka_2.11-0.8.2.2/config目录下，共有如下配置文件：
/home/jason/bigdata/kafka_2.11-0.8.2.2/config
[root@clsserv206 config]# ll
总用量 32
-rw-r--r--. 1 root root 1199 9月 3 2015 consumer.properties
-rw-r--r--. 1 root root 3846 9月 3 2015 log4j.properties
-rw-r--r--. 1 root root 2228 9月 3 2015 producer.properties
-rw-r--r--. 1 root root 5559 9月 3 2015 server.properties
-rw-r--r--. 1 root root 3325 9月 3 2015 test-log4j.properties
-rw-r--r--. 1 root root 993 9月 3 2015 tools-log4j.properties
-rw-r--r--. 1 root root 1023 9月 3 2015 zookeeper.properties
4.2. 修改配置文件server.properties
共修改以下配置：
#为方便，直接将broker.id设置为了ip的最后一段，当集群中有多个Kafka时，他们的这个值必须不一样
broker.id=206
#端口暂时不变
port=9092
#IP修改为本机的IP
host.name=11.12.112.206
#可选配置项，将日志输出到指定的位置
log.dirs=/home/logs/kafka-logs
#必须配置自己的zookeeper
zookeeper.connect=11.12.112.215:2181,11.12.112.216:2181,11.12.112.217:2181
4.3. 为加快配置可将这里修改好的配置文件server.properties复制到其他几个主机（207、208）
上的相应目录/home/jason/bigdata/下，使用命令示例：
scp -r kafka_2.11-0.8.2.2 root@11.12.112.207:/home/jason/bigdata/
scp -r kafka_2.11-0.8.2.2 root@11.12.112.208:/home/jason/bigdata/
【注意】拷贝过去之后要记得仿照修改配置文件server.properties中最主要的
两个配置项：broker.id和host.name，zookeeper的地址列表不变。
5.启动Kafka
进入目录：/home/jason/bigdata/kafka_2.11-0.8.2.2，执行命令：
sudo nohup ./bin/kafka-server-start.sh config/server.properties &
然后在其他集群上按照同样方式配置并启动Kafka；
6.判断Kafka是否启动成功
通过在一台server上创建producer，另外一台创建consumer，从producer上发送信息，看consumer是否能接收到，以验证集群对否成功。
6.1创建topic
使用bin目录下的kafka-topics.sh，
命令为：
sudo ./bin/kafka-topics.sh -zookeeper 11.12.112.215:2181,11.12.112.216:2181,11.12.112.217:2181 -topic test-hello-jason -replication-factor 2 -partitions 5 -create
其中：-zookeeper之后要指定zookeeper集群的地址列表，-topic之后指定topic的地址。
如下所示：
[root@clsserv206 kafka_2.11-0.8.2.2]# sudo ./bin/kafka-topics.sh -zookeeper 11.12.112.215:2181,11.12.112.216:2181,11.12.112.217:2181 -topic test-hello-jason -replication-factor 2 -partitions 5 -create
可看到输出结果：
Created topic "test-hello-jason".
6.2 查看topic创建情况
使用命令：
sudo ./bin/kafka-topics.sh -zookeeper 11.12.112.215:2181,11.12.112.216:2181,11.12.112.217:2181 -list
命令执行之后，即可看到刚才创建的topic：如下所示：
[root@clsserv206 kafka_2.11-0.8.2.2]# sudo ./bin/kafka-topics.sh -zookeeper 11.12.112.215:2181,11.12.112.216:2181,11.12.112.217:2181 -list
flume_log
idgen
new_log_1
new_ps_log_1
ng_log
nglog
ps_log1
ps_log_new - marked for deletion
ps_log_new_1
test-hello-jason
6.3 创建一个生产者
在11.12.112.206主机的Kafka上启动一个producer，使用Kafka自带的kafka-console-producer.sh来启动，命令如下：
sudo ./bin/kafka-console-producer.sh -broker-list 11.12.112.206:9092,11.12.112.207:9092,11.12.112.208:9092 -topic test-hello-jason
【注意】这里要通过-broker-list来指定Kafka集群的地址列表。Apache kafka、
执行结果如下：
[root@clsserv206 kafka_2.11-0.8.2.2]# sudo ./bin/kafka-console-producer.sh -broker-list 11.12.112.206:9092,11.12.112.207:9092,11.12.112.208:9092 -topic test-hello-jason
[2016-09-09 15:10:05,540] WARN Property topic is not valid (kafka.utils.VerifiableProperties)
6.4 创建一个消费者
在11.12.112.207主机的kafa上启动一个消费者，使用Kafka的bin目录下自带的kafka-console-consumer.sh来启动，命令如下：
sudo ./bin/kafka-console-consumer.sh -zookeeper 11.12.112.215:2181,11.12.112.216:2181,11.12.112.217:2181 - from-begining -topic test-hello-jason
【注意】这里通过-zookeeper指定zookeeper的地址列表，而不是Kafka的地址列表，要通过-topic来指定topic；
执行结果如下：
[root@cluster207 kafka_2.11-0.8.2.2]# sudo ./bin/kafka-console-consumer.sh -zookeeper 11.12.112.215:2181,11.12.112.216:2181,11.12.112.217:2181 - from-begining -topic test-hello-jason
6.5 尝试收发消息
在生产者那里发送一条消息：
hello i am jason
如下：
[2016-09-09 15:10:05,540] WARN Property topic is not valid (kafka.utils.VerifiableProperties)
hello i am jason
在接收者那里就会收到：
[root@cluster207 kafka_2.11-0.8.2.2]# sudo ./bin/kafka-console-consumer.sh -zookeeper 11.12.112.215:2181,11.12.112.216:2181,11.12.112.217:2181 - from-begining -topic test-hello-jason
hello i am jason
至此说明Kafka搭建成功
