附录E 安装Kafka

喜欢ヅ旅行 2022-03-30 07:29 166阅读 0赞

**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 spark@slave1:/app/soft

$scp -r kafka-0.9.0.1 spark@slave2:/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

[![clip\_image002][clip_image002]][clip_image002_clip_image002]

图 附录E‑1 启动Kafka

3.在master节点上新建主题Topic

$kafka-topics.sh --create --topic kafkaTopic --replication-factor 3 --partitions 2 --zookeeper master:2181

[![clip\_image004][clip_image004]][clip_image004_clip_image004]

图 附录E‑2 在Kafka中创建主题

4.在slave1模拟消息生产者，发送消息至Kafka

$kafka-console-producer.sh --broker-list master:9092 --sync --topic kafkaTopic

当消费者连接后，在发送消息的终端输入：hello kafka/who are you?

[![clip\_image006][clip_image006]][clip_image006_clip_image006]

图 附录E‑3 slave1模拟消息生产者发送消息

5.在slave2模拟消息消费者，显示消息的消费

$kafka-console-consumer.sh --zookeeper master:2181 --topic kafkaTopic --from-beginning

由于设置接收从开始到现在的消息，以前发送的消息也显示在slave2终端上。

[![clip\_image008][clip_image008]][clip_image008_clip_image008]

图 附录E‑4 slave2模拟消息消费者收到消息

[clip_image002]: https://images2015.cnblogs.com/blog/107289/201612/107289-20161218224515089-1703124351.jpg
[clip_image002_clip_image002]: /images/20220330/50f0a915b5774a8ca14fbbb40be4bdbf.png
[clip_image004]: https://images2015.cnblogs.com/blog/107289/201612/107289-20161218224516167-1333969799.jpg
[clip_image004_clip_image004]: /images/20220330/bcdd446d282b4c608fc981a3ec72acc3.png
[clip_image006]: https://images2015.cnblogs.com/blog/107289/201612/107289-20161218224517214-715077972.jpg
[clip_image006_clip_image006]: /images/20220330/1d424c8fe9ab4c33a511b9242b3ddc65.png
[clip_image008]: https://images2015.cnblogs.com/blog/107289/201612/107289-20161218224518714-1133454603.jpg
[clip_image008_clip_image008]: /images/20220330/2b5cc82de0ce44c2aea7c0b2ba252dd7.png