Kafka的安装部署

第一步、 首先安装JRE/JDK 
第二步、安装Zookeeper

第三步、 下载kafka 

进入下载页面：http://kafka.apache.org/downloads.html 

选择Binary downloads下载 （Source download需要编译才能使用），也可以直接在linux终端下载： 
Shell代码  1. wget -q http://apache.fayea.com/apache-mirror/kafka/0.8.1/kafka_2.8.0-0.8.1.tgz  
 
第四步、解压
Shell代码  1. tar -xzvf kafka_2.8.0-0.8.1.tgz  2. rm kafka_2.8.0-0.8.1.tgz  3. cd kafka_2.8.0-0.8.1
 
解压后的目录：
 
Ø /bin 启动和停止命令等 
Ø /config 配置文件 
Ø /libs 类库 
注意：根据环境去修改配置文件中的信息
第五步、启动
Ø 启动Kafka server
Shell代码  1. bin/kafka-server-start.sh config/server.properties &   Ø 停止kafka server
Shell代码  1. bin/kafka-server-stop.sh  
 
 
测试：（使用控制台输出测试）
一、单机的连通性测试
Ø 运行producer： 
Shell代码  1. bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test  

早期版本的Kafka，--broker-list localhost:9092需改为--zookeeper localhost:2181  
 
Ø 运行 consumer
Shell代码  1. bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic test --from-beginning  

在producer端输入字符串并回车，查看consumer端是否显示。 

二、分布式连通性测试 

Zookeeper Server, Kafka Server, Producer都放在服务器server1上，ip地址为192.168.1.10 
Consumer放在服务器server2上，ip地址为192.168.1.12。 

分别运行server1的producer和server2的consumer， 
Shell代码  1. bin/kafka-console-producer.sh --broker-list 192.168.1.10:9092 --topic test  

Shell代码  1. bin/kafka-console-consumer.sh --zookeeper 192.168.1.10:2181 --topic test --from-beginning  
 
注意：
一、系统错误
Kafka默认开启JVM压缩指针，但只是在64位的HotSpot VM受支持，如果安装了32位的HotSpot VM，需要修改/bin/kafka-run-class.sh文件 
Shell代码  1. vi bin/kafka-run-class.sh  

找到如下行： 
Sh代码  1. KAFKA_JVM_PERFORMANCE_OPTS="-server -XX:+UseCompressedOops -XX:+UseParNewGC -XX:+UseConcMarkSweepGC -XX:+CMSClassUnloadingEnabled -XX:+CMSScavengeBeforeRemark -XX:+DisableExplicitGC -Djava.awt.headless=true"  

去除-XX:+UseCompressedOops参数 
 
二、分布式连通测试报错
在producer的console端输入字符串，consumer报Connection refused错误： 
broker, producer和consumer都注册到zookeeper上，producer和consumer的参数明确指定。问题出在broker的配置文件server.properties上： 
Properties代码  1. # Hostname the broker will bind to. If not set, the server will bind to all interfaces  2. #host.name=localhost  

host名称没有指定，就是127.0.0.1，consumer去broker拿数据就有问题。设置为192.168.1.10，重启服务就好了。
 
三、Kafka集群的自动停止
解决方案就是采取后台启动的方式：bin/kafka-server-start.sh -daemon ./config/server.properties  &