在Spark生态体系中，Kafka占有非常重要的位置。Kafka是一个使用Scala语言编写的基于ZooKeeper的高吞吐量低延迟的分布式发布与订阅消息系统，它可以实时处理大量消息数据以满足各种需求。在实际开发中，Kafka常常作为Spark Streaming的实时数据源，Spark Streaming从Kafka中读取实时消息进行处理，保证了数据的可靠性与实时性。

一个典型的Kafka集群中包含若干生产者（数据可以是Web前端产生的页面内容或者服务器日志等）、若干Broker、若干消费者（可以是Hadoop集群、实时监控程序、数据仓库或其他服务）以及一个ZooKeeper集群。ZooKeeper用于管理和协调Broker。当Kafka系统中新增了Broker或者某个Broker故障失效时，ZooKeeper将通知生产者和消费者。生产者和消费者据此开始与其他Broker协调工作。

Kafka通过主题对消息进行分类，一个主题可以分为多个分区，且每个分区可以存储于不同的Broker上，也就是说，一个主题可以横跨多个服务器。

一、kafka安装

1.下载

官网上下载对应安装包：

解压到相应的文件夹中：

# kafka
export KAFKA_HOME="/home/ffzs/softwares/kafka_2.12-2.8.1"
export PATH=$KAFKA_HOME/bin:$PATH

修改log目录：

[~/softwares/kafka_2.12-2.8.1]$ vim ./config/server.properties
# 最后添加日志输出地址
log.dirs=/home/ffzs/softwares/kafka_2.12-2.8.1/logs

2.启动zookeeper

kafka使用依赖于zookeeper必须要启动zookeeper才能使用kafka，这里实验环境就直接用kafka带的zookeeper就行。

(base) [~]$ zookeeper-server-start.sh -daemon $KAFKA_HOME/config/zookeeper.properties

• 通过 -daemon 来指定后台运行

3.启动kafka

不适用后台启动kafka，看看有没有什么报错的地方：

(base) [~]$ kafka-server-start.sh $KAFKA_HOME/config/server.properties

没有报错的话，通过后台启动

(base) [~]$ kafka-server-start.sh -daemon $KAFKA_HOME/config/server.properties

4.创建主题

首先创建一个topic， 因为是本地运行分区，副本都设置为1：

(base) [~]$ kafka-topics.sh --create --zookeeper ffzs-ub:2181 --replication-factor 1 --partitions 1 --topic topictest

上述代码中各参数的含义如下：

• –create：指定命令的动作是创建主题，使用该命令必须指定–topic参数。
• –topic：所创建的主题名称。
• –partitions：所创建主题的分区数。
• –zookeeper：指定ZooKeeper集群的访问地址。
• –replication-factor：所创建主题的分区副本数，其值必须小于等于Kafka的节点数。

5.查询主题

对命令对主题进行查询，看看主题是否存在：

(base) [~]$ kafka-topics.sh --list --zookeeper ffzs-ub:2181
__consumer_offsets
topictest

6.创建生产者

Kafka生产者作为消息生产角色，可以使用Kafka自带的命令工具创建一个简单的生产者。

(base) [~/softwares/kafka_2.12-2.8.1/logs]$ kafka-console-producer.sh --broker-list ffzs-ub:9092 --topic topictest

上述代码中各参数的含义如下：

• –broker-list：指定Kafka Broker的访问地址，只要能访问其中一个，即可连接成功，若想写多个，则用逗号隔开。建议将所有的Broker都写上，如果只写其中一个，该Broker失效时连接就会失败
• –topic：指定生产者发送消息的主题名称

7.创建消费者

Kafka生产者作为消息生产角色，可以使用Kafka自带的命令工具创建一个简单的生产者。

(base) [~/softwares/kafka_2.12-2.8.1/logs]$ kafka-console-consumer.sh --bootstrap-server ffzs-ub:9092 --topic topictest

8、测试

通过在生产者中输入数据，可以观察到消费者中打印出了相应的数据。