简介
Kafka是一种分布式流处理平台，它具有高吞吐量、可扩展性、可靠性、实时性和灵活性等优点。它能够支持每秒数百万条消息的传输，并且可以通过增加节点来增加吞吐量和存储容量。Kafka通过将数据复制到多个节点来实现数据冗余和高可用性，即使某个节点故障，也可以保证数据不会丢失。它能够快速地处理和传输数据，支持实时数据的处理和分析。此外，Kafka可以与各种不同的数据处理和分析工具集成，包括流处理、批处理、数据挖掘等等。
Kafka的优点包括高吞吐量、可扩展性、可靠性、实时性和灵活性。它能够处理大量的数据，支持高并发的场景，并且可以水平扩展以支持更大的负载。此外，Kafka还提供了消息持久化的能力，可以保证数据的可靠性和不丢失。
Kafka的缺点包括消息可能会重复消费影响数据精确度、数据达不到真正的实时以及只能支持统一分区内消息有序等。此外，Kafka不支持事务和消息的持久化，可能会在系统故障时丢失数据。
Kafka的同类产品包括RabbitMQ、ActiveMQ、Amazon SQS等等。这些产品都是消息队列系统，具有类似的功能和用途。它们可以用于解耦和缓冲应用程序之间的消息传递，支持异步和同步通信，提供消息持久化和可靠性保证等功能。
Kafka解决的场景问题包括实时数据流的处理和分析、大规模数据的分布式处理和存储、解耦和缓冲应用程序之间的消息传递等等。它可以用于在线购物平台、智能家居产品等互联网产品的消息传递和数据处理，以及金融、物流等行业的实时数据处理和分析。
一个典型的Kafka集群包含若干个生产者（Producer）、若干Kafka集群节点（Broker）、若干消费者（Consumer）以及一个Zookeeper集群。Kafka通过Zookeeper管理集群配置，选举Leader以及在消费者发生变化时进行负载均衡。生产者使用推（Push）模式将消息发布到集群节点，而消费者使用拉（Pull）模式从集群节点中订阅并消费消息。
安装
zookeeper安装
hadoop安装
Kafka安装
kafka安装、配置都比较简单，没有主从节点之分。
进入官网选择二进制安装
所有节点均执行以下操作，我们在安装没有一个组件的时候都会设置软连接，主要是方便其他组件的挂接和后期组件的升级。
[hadoop@vm02 ~]$ tar -zxf kafka_2.12-3.6.0.tgz
[hadoop@vm02 ~]$ rm -rf kafka_2.12-3.6.0.tgz
[hadoop@vm02 ~]$ ln -s kafka_2.12-3.6.0/ kafka
配置文件设置
进入到kafka的配置目录（/home/hadoop/kafka/config）中间，设置相关配置参数
zookeeper.properties配置文件
该文件主要是挂接上zookeeper，参数kafka集群直接的选主实现运行的可靠性和高可用性
在kafka安装包中自带有zookeeper的程序，本文不适用自带的zookeeper程序，我们独立安装了zookeeper，用于管理hadoop中的所有组件，本文的kafka也将其挂到hadoop统一的zookeeper的程序中去。
在默认文件中设置了五个参数（其中一个被#注释），所有节点均按照下图配置
# the directory where the snapshot is stored.
dataDir=/home/hadoop/zookeeper/zkdata
# the port at which the clients will connect
clientPort=2181
# disable the per-ip limit on the number of connections since this is a non-production configq
maxClientCnxns=10
# Disable the adminserver by default to avoid port conflicts.
# Set the port to something non-conflicting if choosing to enable this
admin.enableServer=false
# admin.serverPort=8080
-
maxClientCnxns： 这个参数控制单个 IP 地址可以同时连接到 ZooKeeper 服务的最大数量。默认值是60。如果你的应用程序需要大量的客户端连接到 ZooKeeper，或者你发现客户端因为达到这个限制而被拒绝连接，你可以考虑增加这个值。但是请注意，过高的设置可能会导致服务器资源耗尽或拒绝服务攻击。 -
admin.enableServer： 这个参数用来启用或禁用 ZooKeeper 的管理界面。默认情况下，这个参数的值为false，这意味着 ZooKeeper 不会监听任何端口来提供管理界面。如果你想独立使用kafka内置的zookeeper程序，你可以打开，本问是独立安装的zookeeper，所以不适用参数，即设置为false -
dataDir：在参数的地址可以查看zookeeper具体的路径进行配置
[hadoop@vm02 ~]$ cd zookeeper/ [hadoop@vm02 zookeeper]$ ll total 40 drwxr-xr-x. 2 hadoop hadoop 4096 Nov 16 22:59 bin drwxr-xr-x. 2 hadoop hadoop 87 Nov 15 15:44 conf drwxr-xr-x. 5 hadoop hadoop 4096 Oct 4 17:50 docs drwxrwxr-x. 2 hadoop hadoop 4096 Nov 15 15:07 lib -rw-r--r--. 1 hadoop hadoop 11358 Oct 4 17:50 LICENSE.txt drwxrwxr-x. 2 hadoop hadoop 46 Nov 15 15:41 logs -rw-r--r--. 1 hadoop hadoop 2084 Oct 4 17:50 NOTICE.txt -rw-r--r--. 1 hadoop hadoop 2335 Oct 4 17:50 README.md -rw-r--r--. 1 hadoop hadoop 3570 Oct 4 17:50 README_packaging.md drwxrwxr-x. 3 hadoop hadoop 63 Nov 28 00:33 zkdata drwxrwxr-x. 3 hadoop hadoop 23 Nov 15 15:41 zklog -
clientPort=2181：指定zookeeper对外的端口号，这里我们在独立部署zookeeper中使用的2181端口号
consumer.properties配置文件
consumer.properties 文件是一个用于配置消费者客户端的属性文件。这个文件通常包含了消费者如何连接到 Kafka 集群、如何消费消息以及其它相关设置的信息
-
bootstrap.servers： 这个参数指定了一个或多个 Kafka Broker 的地址列表（格式为：hostname:port），用逗号分隔。消费者将首先连接到这些服务器中的一个，并从中获取集群元数据，如 Topic 的分区信息和 Leader 服务器的位置等。然后，消费者会直接连接到负责其订阅 Topic 分区的 Leader 服务器来拉取消息。本文不采用 -
group.id： 这个参数指定了消费者的组 ID。同一组内的所有消费者共享订阅 Topic 的分区，以便实现负载均衡。每个分区只能被该组内的一个消费者实例消费。本文不采用 -
zookeeper.connect：用于指定zookeeper集群节点。hostname:port,hostname:port,hostname:port使用逗号隔开，列出zookeeper集群的所有节点。本文
consumer.properties配置文件仅采用此配置。
所有节点按照以下标准配置
zookeeper.connect=vm02:2181,vm03:2181,vm04:2181
producer.properties配置文件
producer.properties 文件是一个用于配置生产者客户端的属性文件。
学过kafka读者会知道metadata.broker.list参数，现在在kafka较新版本中已经使用bootstrap.servers参数进行替换。
bootstrap.servers（旧版本metadata.broker.list）：启动时producer查询borokers的列表，可以是集群中所有brokersl的一个子集。注意，这个参数只是用来获取topic的元信息用，producer会从元信息中挑选合适的broker并与之建立socket;连接。格式是：host1:port1,host2:port2。本文仅对该参数进行配置。所有节点均采用以下配置
bootstrap.servers=vm02:9092,vm03:9092,vm04:9092
# specify the compression codec for all data generated: none, gzip, snappy, lz4, zstd
compression.type=none
server.properties配置文件
server.properties 文件是一个用于配置 Kafka Broker 的属性文件。
以下讲解一下几个比较重要的参数。
-
Broker ID (
broker.id)： 每个 Broker 必须有一个唯一的 ID。这个 ID 用于在 ZooKeeper 中注册 Broker。kafka集群中每一个节点的设置均不相同。本文采用（broker.id=2broker.id=3broker.id=4
的值分别设置在vm02\vm03\vm04节点中去） -
Listeners (
listeners)： 指定 Broker 监听客户端连接的 IP 地址和端口。例如：PLAINTEXT://localhost:9092。本文采用（PLAINTEXT://vm02:9092 PLAINTEXT://vm03:9092
PLAINTEXT://vm04:9092。
分别设置在vm02\vm03\vm04节点中去） -
ZooKeeper 连接 (
zookeeper.connect)： 指定 ZooKeeper 集群的地址列表（格式为：hostname:port/path），用逗号分隔。本文采用（zookeeper.connect=vm05:2181,vm06:2181,vm07:2181的值，所有节点同步） -
日志目录 (
log.dirs)： 指定 Kafka 存储 Topic 分区日志的目录。可以指定多个路径以多副本日志模式增加kafka的可靠性。本文采用（
log.dirs=/home/hadoop/zookeepr/zklog/kafka-logs的值，所有节点同步） -
自动创建主题 (
auto.create.topics.enable)： 确保设置为false，避免意外创建主题。本文采用(auto.create.topics.enable=false的值，所有节点同步) -
日志保留策略 (
log.retention.{ms,minutes,hours})：{ms,minutes,hours}是时间单位，据需求设置消息的保留时间或大小限制，建议保留七天。
本文采用（log.retention.hours=128的值，所有节点同步） -
自动领导选举 (
unclean.leader.election.enable)： 默认情况下，该参数的值为true，这意味着当 ZooKeeper 发现一个 Broker 不可用时，它会允许从那些与 Leader 失去同步的副本（Out-of-Sync Replicas, OSRs）中选举新的 Leader。这种方式被称为 "不干净的 Leader 选举"（Unclean Leader Election）。这种选举方式的优点是速度快，因为它不需要等待所有副本都完成数据同步。然而，缺点是可能会导致数据丢失或不一致。建议为false。本文采用（unclean.leader.election.enable=false的值，所有节点同步）
进入该文件中的添加参数，添加在文末尾，会覆盖前面的参数，其余的参数使用默认值。
##以下参数每一个节点值不能一样
broker.id=2
listeners=PLAINTEXT://vm02:9092
##以下参数所有节点配置均一致
zookeeper.connect=vm05:2181,vm06:2181,vm07:2181
log.dirs=/home/hadoop/zookeepr/zklog/kafka-logs
auto.create.topics.enable=false
log.retention.hours=128
unclean.leader.election.enable=false
环境变量配置
博主喜欢把环境变量全部配置在/etc/profile中，这样方便运维，一样望去可以知道服务器安装了哪些程序。然后在对应的用户（hadoop用户下）.bash_profile用户环境变量文件中增加source /etc/profile 使得对应的用户能读到所有的环境变量，当然可以根据个人需求，在.bash_profile用户环境变量文件中只配置对应其用户开放的环境变量路径。
#使用root用户编辑该文件
vim /etc/profile增加以下变量（所有节点同步）
export KAFKA_HOME=/home/hadoop/kafka
export PATH=$KAFKA_HOME/bin:$PATH
export SERVER_PROPERTIES=$KAFKA_HOME/config/server.properties
记住：wq 退出保存后，需要使用source /etc/profile 加载一遍环境变量
博主在该服务器上已经配置了一些其他程序，读者将就看吧
切换到hadoop用户下，编辑.bash_profile用户环境变量增加source /etc/profile
启动Kafka集群
启动zookeeper
#使用以下指令启动zookeeper（所有节点同步）
zkServer.sh start 启动Hadoop
#使用以下指令启动hadoop集群的hdfs\yarn\等服务（主节点vm02）
start-all start 启动hbase
#使用以下指令启动hbase（hbase主节点vm02）
start-hbase.sh启动kafka
#使用以下指令启动kafka（所有节点同步）
kafka-server-start.sh -daemon $SERVER_PROPERTIES
查看所有进程使用jps查看所有进程
![[二分查找]LeetCode2040:两个有序数组的第 K 小乘积](https://img-blog.csdnimg.cn/f95ddae62a4e43a68295601c723f92fb.gif#pic_center)
