E.1 安装Kafka
E.1.1 下载Kafka
Kafka是由LinkedIn设计的一个高吞吐量、分布式、基于发布订阅模式的消息系统，使用Scala编写，它以可水平扩展、可靠性、异步通信和高吞吐率等特性而被广泛使用。目前越来越多的开源分布式处理系统都支持与Kafka集成，其中Spark Streaming作为后端流引擎配合Kafka作为前端消息系统正成为当前流处理系统的主流架构之一。
Kafka安装包可以在其官网下载页面下载，下载地址如下，为加快下载速度可以选择中国境内的镜像，选择稳定版本kafka_2.11-0.9.0.1.tgz安装包。
http://kafka.apache.org/downloads.html
E.1.2 解压并配置环境变量
下载后把安装包方放在目录/home/spark/work目录下，用下面命令解压缩Kafka安装包，并把解压后的目录移动到/app/soft目录下：
$cd /home/spark/work/
$tar -zxf kafka_2.11-0.9.0.1.tgz
$mv kafka-0.9.0.1 /app/soft
$ll /app/soft
为了方便运行Kafka相关脚本，将Kafka的bin路径加入到/etc/profile中，设置如下内容（分发到各节点后，在各节点上做同样设置）：
export KAFKA_HOME=/app/soft/kafka-0.9.0.1
export PATH=$PATH:$KAFKA_HOME/bin
设置完毕后使用如下命令使配置生效：
$source /etc/profile
E.1.3 修改Kafka的配置文件
在Kafka的根目录下建立log目录用于存放日志文件：
$mkdir /app/soft/kafka-0.9.0.1/logs/
修改$KAFKA_HOME/config/server.properties配置文件内容（仅列出重要配置）：
################ Server Basics ################
#建议根据ip区分,这里使用ZooKeeper中的id来设置，如master节点设置为0，slave1节点设置为1，slave2节点设置为2
broker.id=1
################ Socket Server Settings ################
#broker用于接收producer消息的端口
port=9092
#broker的hostname
host.name=master
#配置PRODUCER/CONSUMER连上来的时候使用的地址
advertised.host.name=master
################ Log Basics ################
#kafka存放消息文件的路径
log.dirs=/app/soft/kafka-0.9.0.1/logs/
#topic的默认分区数
num.partitions=2
################ ZooKeeper ################
#ZooKeeper集群连接地址信息
zookeeper.connect=master:2181,slave1:2181,slave2:2181
#连接ZooKeeper超时时间，单位为毫秒
zookeeper.connection.timeout.ms=6000
E.1.4 分发Kafka到各节点
使用scp命令到Kafka分发到slave1和slave2节点上：
$cd /app/soft/kafka-0.9.0.1
$scp -r kafka-0.9.0.1 [email protected]:/app/soft
$scp -r kafka-0.9.0.1 [email protected]:/app/soft
分发完毕后，修改server.properties配置文件中broker.id、host.name、advertised.host. name和zookeeper.connect等配置项。
E.2 启动并验证
1. 启动ZooKeeper
分别在master、slave1和slave2节点上启动ZooKeeper服务：
$zkServer.sh start
2. 启动Kafka
分别在master、slave1和slave2节点上启动Kafka服务：
$kafka-server-start.sh $KAFKA_HOME/config/server.properties
图 附录E?1 启动Kafka
3. 在master节点上新建主题Topic
$kafka-topics.sh --create --topic kafkaTopic --replication-factor 3 --partitions 2 --zookeeper master:2181
图 附录E?2 在Kafka中创建主题
4. 在slave1模拟消息生产者，发送消息至Kafka
$kafka-console-producer.sh --broker-list master:9092 --sync --topic kafkaTopic
当消费者连接后，在发送消息的终端输入：hello kafka/who are you?
图 附录E?3 slave1模拟消息生产者发送消息
5. 在slave2模拟消息消费者，显示消息的消费
$kafka-console-consumer.sh --zookeeper master:2181 --topic kafkaTopic --from-beginning
由于设置接收从开始到现在的消息，以前发送的消息也显示在slave2终端上。
图 附录E?4 slave2模拟消息消费者收到消息