Kafka的安装与配置一、知识目标理解Apache Kafka分布式消息队列系统的基本概念、核心架构及其在大数据生态系统中的角色定位。掌握Kafka集群的架构组成和工作原理包括生产者、消费者、Broker、Topic、Partition等核心概念。了解Kafka与ZooKeeper的依赖关系以及集群部署的注意事项。二、技能目标能够独立完成Kafka 3.3.1Scala 2.12版本的下载、解压和环境配置。能够正确配置Kafka集群的核心配置文件并启动集群。能够创建Topic、生产消息和消费消息验证Kafka集群的运行状态。三、素质目标培养在分布式系统部署中的耐心和细致调试精神。提升在Linux集群环境下进行软件安装、配置与维护的综合实践能力。增强版本管理和配置一致性意识。四、重难点重点正确配置server.properties文件中的broker.id、log.dirs和zookeeper.connect三个核心参数。难点多节点环境下的配置一致性管理和集群网络连通性保障。五、任务描述本次任务是Kafka 3.3.1Scala 2.12集群模式的安装与配置使用3台机器master、worker1、worker2。最终效果能够正常创建Topic、生产消息和消费消息验证Kafka集群的运行状态。六、环境要求操作系统CentOS 7LinuxJava环境JDK 1.8 或更高版本Kafka 3.3.1推荐使用JDK 8/11ZooKeeper集群已安装并配置好ZooKeeper集群Kafka依赖ZooKeeper管理集群元数据SSH免密登录Master节点需能免密登录所有Worker节点主机规划主机名IP地址运行服务master192.168.47.201ZooKeeper Kafka Brokerworker1192.168.47.202ZooKeeper Kafka Brokerworker2192.168.47.203ZooKeeper Kafka Broker七、Kafka集群架构7.1 核心组件Broker代理服务器一个Kafka集群由多个Broker组成每个Broker就是一个Kafka服务器实例。每个Broker都有一个唯一的IDbroker.id在集群中必须全局唯一。Topic主题Topic是消息的逻辑分类生产者将消息发送到特定的Topic消费者从Topic订阅消息。Partition分区每个Topic可以划分为多个Partition每个Partition是一个有序的消息队列。分区可以实现并行处理和数据冗余。Producer生产者负责向Kafka集群发送消息的应用程序。Consumer消费者负责从Kafka集群读取消息的应用程序。7.2 Kafka依赖ZooKeeperKafka使用ZooKeeper来管理集群的元数据包括Broker注册、Topic配置、分区Leader选举等。因此在启动Kafka之前必须先启动ZooKeeper集群。注意自Kafka 2.8版本开始Kafka引入了KRaft模式可以不依赖ZooKeeper。但本文档部署的Kafka 3.3.1版本仍然推荐使用ZooKeeper模式以保证稳定性和兼容性。八、任务实施步骤1上传并解压Kafka安装包上传Kafka安装包到master节点的/opt/packages/目录下步骤2配置环境变量进入到/opt/packages/目录下解压Kafka安装包到/opt/programs/目录tar -zxvf kafka_2.12-3.3.1.tgz -C /opt/programs/进入/opt/programs/目录下重命名目录为kafkamv kafka_2.12-3.3.1 kafka步骤3配置环境变量在master节点上配置环境变量在etc目录下编辑环境变量配置文件profilevi profile# 添加以下内容# KAFKA_HOMEexport KAFKA_HOME/opt/programs/kafkaexport PATH$PATH:$KAFKA_HOME/bin使配置生效source profile验证配置echo $KAFKA_HOME在master节点上的etc目录下的配置文件profile分发到worker1和worker2节点发送profile文件到worker1节点scp profile rootworker1:/etc/发送profile文件到worker2节点scp profile rootworker2:/etc/在worker1节点上使配置生效source /etc/profile在worker2节点上使配置生效source /etc/profile步骤4修改Kafka核心配置文件进入到Kafka配置目录/opt/programs/kafka/config/4.1 修改 server.properties 配置文件在master节点上编辑server.propertiesvi server.properties修改或添加以下内容# 基础配置# Broker ID集群中全局唯一master节点设为0broker.id0# Kafka数据存储目录需手动创建log.dirs/opt/programs/kafka/datas# ZooKeeper集群连接地址注意需要先启动ZooKeeper集群zookeeper.connectmaster:2181,worker1:2181,worker2:2181/kafka# 网络配置#监听地址和端口listenersPLAINTEXT://master:9092#对外暴露的地址集群部署需配置为节点IP或主机名advertised.listenersPLAINTEXT://master:9092# Topic配置#默认分区数num.partitions1#默认副本因子default.replication.factor1#允许自动创建Topicauto.create.topics.enabletrue# 日志配置#日志保留时间小时log.retention.hours168#日志段文件大小字节log.segment.bytes1073741824注意复制过去后需把中文删除配置参数说明broker.id每个Broker的唯一标识集群中必须全局唯一。master设为0worker1设为1worker2设为2。log.dirsKafka存储消息数据的目录需要手动创建。zookeeper.connectZooKeeper集群的连接地址多个地址用逗号分隔。/kafka是ZooKeeper的chroot路径用于隔离不同Kafka集群的数据。listenersKafka Broker监听的地址和端口。advertised.listeners对外暴露的地址供生产者和消费者连接。num.partitions默认分区数影响Topic的并行处理能力。default.replication.factor默认副本因子集群部署建议设为2或3以提高数据可靠性。4.2在/opt/programs/kafka/目录下创建Kafka数据目录mkdir -p datas步骤5分发Kafka到Worker节点在master节点上进入/opt/programs目录下分发到worker1节点scp -r kafka rootworker1:/opt/programs/分发到worker2节点scp -r kafka rootworker2:/opt/programs/步骤6修改Worker节点的配置文件6.1 修改worker1节点的配置在worker1节点上的/opt/programs/kafka/config/目录下修改server.propertiesvi server.properties修改以下内容broker.id1listenersPLAINTEXT://worker1:9092advertised.listenersPLAINTEXT://worker1:90926.2 修改worker2节点的配置在worker2节点上的/opt/programs/kafka/config/目录下修改server.propertiesvi server.properties修改以下内容broker.id2listenersPLAINTEXT://worker2:9092advertised.listenersPLAINTEXT://worker2:9092步骤7启动ZooKeeper集群在启动Kafka之前必须先启动ZooKeeper集群在master、worker1、worker2三个节点上分别执行zkServer.sh start查看ZooKeeper进程步骤8启动Kafka集群在三个节点上分别启动Kafka服务在master节点kafka-server-start.sh -daemon /opt/programs/kafka/config/server.properties在worker1节点kafka-server-start.sh -daemon /opt/programs/kafka/config/server.properties在worker2节点kafka-server-start.sh -daemon /opt/programs/kafka/config/server.properties步骤9验证集群运行状态9.1 查看进程在各节点使用jps命令检查进程master、worker1和worker2节点节点应看到QuorumPeerMain (ZooKeeper)、Kafka (Kafka Broker)、Jps9.2 查看ZooKeeper中的Kafka节点连接ZooKeeperzkCli.sh查看Kafka相关的ZooKeeper节点ls /kafka/brokers/ids应显示[0, 1, 2]表示三个Broker都已注册步骤10测试Kafka集群10.1 创建Topic# 创建一个名为test的Topic3个分区副本因子为2kafka-topics.sh --bootstrap-server master:9092,worker1:9092,worker2:9092 \--create --replication-factor 2 --partitions 3 --topic test10.2 查看Topic列表kafka-topics.sh --bootstrap-server master:9092,worker1:9092,worker2:9092 --list10.3 查看Topic详情kafka-topics.sh --bootstrap-server master:9092,worker1:9092,worker2:9092 \--describe --topic test10.4 生产消息# 启动生产者发送消息kafka-console-producer.sh --broker-list master:9092,worker1:9092,worker2:9092 --topic test hello kafka welcome to big data world test message这里是发送了三条消息 hello kafka welcome to big data world test message图中只发送了一条消息10.5 消费消息在worker1或worker2打开一个新的终端窗口启动消费者# 启动消费者从开头消费消息kafka-console-consumer.sh --bootstrap-server master:9092,worker1:9092,worker2:9092 \--topic test --from-beginning10.6 在master节点删除Topickafka-topics.sh --bootstrap-server master:9092,worker1:9092,worker2:9092 \--delete --topic test恭喜你学会了Kafka的安装与配置附录集群启停脚本逐台机器手动关闭分别登录到三台机器上执行关闭命令/opt/programs/kafka/bin/kafka-server-stop.sh然后关闭ZooKeeper/opt/programs/zookeeper/bin/zkServer.sh stop如果还存在ConsoleProducer 进程可以使用以下命令# 查找并强制终止所有 ConsoleProducer 进程ps -ef | grep ConsoleProducer | grep -v grep | awk {print $2} | xargs kill -9