Kafka 安装和测试
1. 简介
kafka （官网地址：http://kafka.apache.org）是一款分布式消息发布和订阅的系统，具有高性能和高吞吐率。
i. 消息的发布（publish）称作producer，消息的订阅（subscribe）称作consumer，中间的存储阵列称作broker。
ii. 多个broker协同合作，producer、consumer和broker三者之间通过zookeeper来协调请求和转发。
iii. producer产生和推送(push)数据到broker，consumer从broker拉取(pull)数据并进行处理。
iv. broker端不维护数据的消费状态，提升了性能。
v. 直接使用磁盘进行存储，线性读写，速度快：避免了数据在JVM内存和系统内存之间的复制，减少耗性能的创建对象和垃圾回收。
vi. Kafka使用scala编写，可以运行在JVM上。
2. 安装：
a. 首先安装JRE/JDK
Linux安装JDK
b. 下载kafka
进入下载页面：http://kafka.apache.org/downloads.html
选择Binary downloads下载 （Source download需要编译才能使用）
也可以直接在linux终端下载：
c. 解压
目录：
/bin 启动和停止命令等。
/config 配置文件
/libs 类库
d. 修改配置
Kafka默认开启JVM压缩指针，但只是在64位的HotSpot VM受支持，如果安装了32位的HotSpot VM，需要修改/bin/kafka-run-class.sh文件
找到如下行：
去除-XX:+UseCompressedOops参数
3. 启动和停止
启动Zookeeper server：
&是为了能退出命令行
启动Kafka server:
停止Kafka server
停止Zookeeper server:
4. 单机连通性测试
运行producer：
早版本的Kafka，--broker-list localhost:9092需改为--zookeeper localhost:2181
运行consumer：
在producer端输入字符串并回车，查看consumer端是否显示。
5. 分布式连通性测试
Zookeeper Server, Kafka Server, Producer都放在服务器server1上，ip地址为192.168.1.10
Consumer放在服务器server2上，ip地址为192.168.1.12。
分别运行server1的producer和server2的consumer，
在producer的console端输入字符串，consumer报Connection refused错误：
broker, producer和consumer都注册到zookeeper上，producer和consumer的参数明确指定。问题出在broker的配置文件server.properties上：
host名称没有指定，就是127.0.0.1，consumer去broker拿数据就有问题。设置为192.168.1.10，重启服务就好了。
kafka （官网地址：http://kafka.apache.org）是一款分布式消息发布和订阅的系统，具有高性能和高吞吐率。
i. 消息的发布（publish）称作producer，消息的订阅（subscribe）称作consumer，中间的存储阵列称作broker。
ii. 多个broker协同合作，producer、consumer和broker三者之间通过zookeeper来协调请求和转发。
iii. producer产生和推送(push)数据到broker，consumer从broker拉取(pull)数据并进行处理。
iv. broker端不维护数据的消费状态，提升了性能。
v. 直接使用磁盘进行存储，线性读写，速度快：避免了数据在JVM内存和系统内存之间的复制，减少耗性能的创建对象和垃圾回收。
vi. Kafka使用scala编写，可以运行在JVM上。
2. 安装：
a. 首先安装JRE/JDK
Linux安装JDK
b. 下载kafka
进入下载页面：http://kafka.apache.org/downloads.html
选择Binary downloads下载 （Source download需要编译才能使用）
也可以直接在linux终端下载：
Shell代码
- wget-qhttp://apache.fayea.com/apache-mirror/kafka/0.8.1/kafka_2.8.0-0.8.1.tgz
c. 解压
Shell代码
- tar-xzvfkafka_2.8.0-0.8.1.tgz
- rmkafka_2.8.0-0.8.1.tgz
- cdkafka_2.8.0-0.8.1
目录：
/bin 启动和停止命令等。
/config 配置文件
/libs 类库
d. 修改配置
Kafka默认开启JVM压缩指针，但只是在64位的HotSpot VM受支持，如果安装了32位的HotSpot VM，需要修改/bin/kafka-run-class.sh文件
Shell代码
- vibin/kafka-run-class.sh
找到如下行：
Sh代码
- KAFKA_JVM_PERFORMANCE_OPTS="-server-XX:+UseCompressedOops-XX:+UseParNewGC-XX:+UseConcMarkSweepGC-XX:+CMSClassUnloadingEnabled-XX:+CMSScavengeBeforeRemark-XX:+DisableExplicitGC-Djava.awt.headless=true"
去除-XX:+UseCompressedOops参数
3. 启动和停止
启动Zookeeper server：
Shell代码
- bin/zookeeper-server-start.shconfig/zookeeper.properties&
&是为了能退出命令行
启动Kafka server:
Shell代码
- bin/kafka-server-start.shconfig/server.properties&
停止Kafka server
Shell代码
- bin/kafka-server-stop.sh
停止Zookeeper server:
Shell代码
- bin/zookeeper-server-stop.sh
4. 单机连通性测试
运行producer：
Shell代码
- bin/kafka-console-producer.sh--broker-listlocalhost:9092--topictest
早版本的Kafka，--broker-list localhost:9092需改为--zookeeper localhost:2181
运行consumer：
Shell代码
- bin/kafka-console-consumer.sh--zookeeperlocalhost:2181--topictest--from-beginning
在producer端输入字符串并回车，查看consumer端是否显示。
5. 分布式连通性测试
Zookeeper Server, Kafka Server, Producer都放在服务器server1上，ip地址为192.168.1.10
Consumer放在服务器server2上，ip地址为192.168.1.12。
分别运行server1的producer和server2的consumer，
Shell代码
- bin/kafka-console-producer.sh--broker-list192.168.1.10:9092--topictest
Shell代码
- bin/kafka-console-consumer.sh--zookeeper192.168.1.10:2181--topictest--from-beginning
在producer的console端输入字符串，consumer报Connection refused错误：
broker, producer和consumer都注册到zookeeper上，producer和consumer的参数明确指定。问题出在broker的配置文件server.properties上：
Properties代码
- #Hostnamethebrokerwillbindto.Ifnotset,theserverwillbindtoallinterfaces
- #host.name=localhost
host名称没有指定，就是127.0.0.1，consumer去broker拿数据就有问题。设置为192.168.1.10，重启服务就好了。