一、什么是Kafka？

数据工程中最具挑战性的部分之一是如何从不同点收集和传输大量数据到分布式系统进行处理和分析。需要通过消息队列正确地分离大量数据，因为如果一部分数据无法传送，则可以在系统恢复时传输和分析其他数据。有两种消息排队，对于上述目的，它们都是可靠的和异步的。点对点(Point to point)和发布者——订阅者(publisher-subscriber)。下图展示了一个典型的消息系统，其中：消息的生产者负责产生消息；消息的消费者负责处理消息。

Kafka是一个分布式发布——订阅消息传递系统。Kafka快速、可扩展且耐用。它保留主题中的消息源。生产者将数据写入主题，消费者从主题中读取数据。
Zookeeper需要覆盖Kafka生态系统，因此有必要下载它，更改其属性并最终设置环境。在运行Zookeeper之后，应该下载Kafka，然后开发人员可以借助一些指令创建代理，集群和主题。

二、消息队列的分类

• 点对点（Queue）

在点对点或一对一中，有一个发件人和正在监听发件人的多个消费者。当一个消费者从队列收到消息时，该特定消息将从队列中消失，而其他消费者无法获得该消息。

• 发布和订阅系统（Topic）

在发布者——订阅者中，发布者向同时收听发布者的多个消费者或订阅者发送消息，并且每个订阅者可以获得相同的消息。数据应通过数据管道传输，数据管道负责整合来自数据源的数据。

三、Kafka的体系架构

• 主题和发布者

有一个发布者发送消息。消息根据主题进行分类，每个主题都有一个或多个分区，并有自己的偏移地址。例如，如果我们为一个主题分配复制因子= 2，那么Kafka将为每个分区创建两个相同的副本并在群集中找到它。

• 集群和Brokers

Kafka集群包括代理——服务器或节点，每个代理可以位于不同的机器中，并允许订户选择消息。因此，复制就像备份分区一样，这意味着Kafka是持久的，这有助于容错。

• Zookeeper

Kafka集群不保留其自身生态系统的元数据，因为它是无状态的。因此，Kafka依赖于Zookeeper来跟踪元数据。Zookeeper应该首先启动。实际上，Zookeeper是brokers和consumers之间的接口，它的存在是容错的必要条件。Kafka代理负责负载平衡，假设该主题有一个主题和多个分区，每个分区都有一个领导者，定期确认其与Zookeeper的偏移量。因此，如果一个节点或代理失败，Kafka可以从Zookeeper请求的最后一个偏移地址继续操作，因此Zookeeper在崩溃情况下在Kafka恢复中起着至关重要的作用。

四、Kafka单机单Broker的部署

• 部署ZooKeeper

配置/root/training/zookeeper-3.4.6/conf/zoo.cfg文件
dataDir=/root/training/zookeeper-3.4.6/tmp
server.1=hadoop112:2888:3888

在/root/training/zookeeper-3.4.6/tmp目录下创建一个myid的空文件
echo 1 > /root/training/zookeeper-3.4.6/tmp/myid

启动ZooKeeper
zkServer.sh  start

查看ZooKeeper的状态
zkServer.sh  status

由于我们部署的是单节点的ZooKeeper，所以ZooKeeper的状态将是Standalone。

• 部署Kafka

修改server.conf文件
broker.id=0
port=9092
log.dirs=/root/training/kafka_2.11-2.4.0/logs/broker0
zookeeper.connect=localhost:2181

启动Kafka
bin/kafka-server-start.sh config/server.properties &

• 使用JPS查看后台进程

五、测试Kafka

创建Topic
bin/kafka-topics.sh --create --zookeeper bigdata111:2181 --replication-factor 1 --partitions 3 --topic mytopic1

发送消息
bin/kafka-console-producer.sh --broker-list bigdata111:9092 --topic mytopic1

接收消息
消息消费
bin/kafka-console-consumer.sh --bootstrap-server bigdata111:9092 --topic mytopic1

从开始位置消费
bin/kafka-console-consumer.sh --bootstrap-server bigdata111:9092 --from-beginning --topic topicName

显示key消费
bin/kafka-console-consumer.sh --bootstrap-server bigdata111:9092 --property print.key=true --topic mytopic1