大数据系列 | Kafka架构分析及应用

大数据系列 | Kafka架构分析及应用

  • 1. Kafka原理分析
  • 2. Kafka架构分析
  • 3. Kafka的应用
    • 3.1. 安装Zookeeper集群
    • 3.2. 安装Kafka集群
    • 3.3. 生产者和消费者使用
      • 3.3.1. 生产者使用
      • 3.3.1. 消费者使用
  • 4. Kafka Controller控制器

1. Kafka原理分析

   Kafka是一个高吞吐量、 持久性的分布式发布/订阅消息系统。 其有以下特点。
   ● 高吞吐量:可以满足每秒百万级别消息的生产和消费。
   ● 持久性:有一套完善的消息存储机制, 可以确保数据高效、 安全地持久化。
   ● 分布式:基于分布式架构, 安全, 稳定

Kafka 的数据是存储在磁盘中的, 为什么可以满足每秒百万级别消息的生产和消费? 主要是因为 Kafka 用到了磁盘顺序,所以其读写速度超过内存随机(往硬盘的)读写速度。

   Kafka主要应用在实时数据计算领域。 利用Flume实时采集日志文件中的新增数据, 然后将其存储到Kafka中, 最后在Kafka 后对接实时计算程序。 这其实是一个典型的实时数据计算流程。

2. Kafka架构分析

   Kafka 中包含 Broker、Topic、Partition、Message、Producer和Consumer等组件
在这里插入图片描述

   ● Broker: 消息的代理。 Kafka 集群中的节点(机器) 被称为 Broker。
   ● Topic: 主题。 这是一个逻辑概念, 负责存储 Kafka 中的数据, 相同类型的数据一般会存储到同一个 Topic 中。 可以把 Topic 认为是数据库中的表。
   ● Partition: Topic 物理上的分组。 1 个 Topic 在 Broker 中被分为 1 个或者多个 Partition。分区是在创建 Topic 时指定的, 每个 Topic 都是有分区的, 至少 1 个。 Kafka 中的数据实际上存储在 Partition 中。
   ● Message: 消息, 是数据通信的基本单位。 每个消息都属于 1 个 Partition。
   ● Producer: 消息和数据的生产者, 向 Kafka 的 Topic 生产数据。
   ● Consumer: 消息和数据的消费者, 从 Kafka 的 Topic 消费数据。

   Zookeeper 并不属于 Kafka 的组件, 但是 Kafka 可以根据需求选择依赖 Zookeeper。自Kafka 2.8版本开始,它“抛弃”了Zookeeper,引入了KIP-500架构升级,使用Raft协议实现去中心化。在新的架构中,每个Controller节点都保存所有元数据,通过KRaft协议保证副本的一致性,从而解决了Zookeeper带来的复杂度增加、必须具备Zookeeper运维能力、Controller故障处理麻烦、分区瓶颈等问题8。因此,Kafka并不再是必须依赖Zookeeper。

3. Kafka的应用

3.1. 安装Zookeeper集群

   官方地址:https://downloads.apache.org/zookeeper/
在这里插入图片描述
   选择xx-bin.tar.gz结尾的软件包

root@Agent1:~# cat /etc/hosts
127.0.0.1 localhost
127.0.1.1 zabbix-server# The following lines are desirable for IPv6 capable hosts
::1     ip6-localhost ip6-loopback
fe00::0 ip6-localnet
ff00::0 ip6-mcastprefix
ff02::1 ip6-allnodes
ff02::2 ip6-allrouters192.168.77.252 Agent1
192.168.77.253 Agent2
192.168.77.254 Agent3

   zookeeper1安装:

root@Agent1:~# tar xf /opt/apache-zookeeper-3.8.4-bin.tar.gz -C /root
root@Agent1:~# cd apache-zookeeper-3.8.4-bin/conf/
root@Agent1:~/apache-zookeeper-3.8.4-bin/conf# cp zoo_sample.cfg zoo.cfg

   修改Zookeeper的配置文件:

root@Agent1:~/apache-zookeeper-3.8.4/conf# vim zoo.cfg
dataDir=/root/apache-zookeeper-3.8.4-bin/data
server.0=Agent1:2888:3888
server.1=Agent2:2888:3888
server.2=Agent3:2888:3888

   server.x,server后面的编号需要和kafka配置文件中的broker.id对应
   2888表示Flower跟Leader的通信端口,简称服务端内部通信的端口,默认是2888。3888表示是选举端口,默认是3888
   clientPort是客户端连接的接口,客户端连接zookeeper服务器的端口,zookeeper会监听这个端口,接收客户端的请求访问。默认是2181
   在Zookeeper中创建data目录以保存 myid 文件:

root@Agent1:~/apache-zookeeper-3.8.4-bin# pwd
/root/apache-zookeeper-3.8.4-bin
root@Agent1:~/apache-zookeeper-3.8.4-bin# mkdir data
root@Agent1:~/apache-zookeeper-3.8.4-bin# echo 0 > ./data/myid

   zookeeper2安装:

root@Agent2:~# tar xf /opt/apache-zookeeper-3.8.4-bin.tar.gz -C /root
root@Agent2:~# cd apache-zookeeper-3.8.4-bin/
root@Agent2:~/apache-zookeeper-3.8.4-bin# mkdir data

   zookeeper3安装:

root@Agent3:~# tar xf /opt/apache-zookeeper-3.8.4-bin.tar.gz -C /root
root@Agent3:~# cd apache-zookeeper-3.8.4-bin/
root@Agent3:~/apache-zookeeper-3.8.4-bin# mkdir data

   把Agent1节点上修改好配置的 Zookeeper 安装包复制到另外两台机器中:

root@Agent1:~/apache-zookeeper-3.8.4-bin# scp conf/zoo.cfg 192.168.77.253:/root/apache-zookeeper-3.8.4-bin/conf
root@Agent1:~/apache-zookeeper-3.8.4-bin# scp conf/zoo.cfg 192.168.77.254:/root/apache-zookeeper-3.8.4-bin/confroot@Agent1:~/apache-zookeeper-3.8.4-bin# scp data/myid  192.168.77.253:/root/apache-zookeeper-3.8.4-bin/data
root@Agent1:~/apache-zookeeper-3.8.4-bin# scp data/myid  192.168.77.254:/root/apache-zookeeper-3.8.4-bin/data

   修改Agent2和Agent3上Zookeeper中myid文件的内容:

root@Agent2:~/apache-zookeeper-3.8.4-bin# cat data/myid
1
root@Agent3:~/apache-zookeeper-3.8.4-bin# cat data/myid
2

   启动zookeeper集群:

./bin/zkServer.sh start-foreground :表示在前台启动

root@Agent1:~/apache-zookeeper-3.8.4-bin# pwd
/root/apache-zookeeper-3.8.4-binroot@Agent1:~/apache-zookeeper-3.8.4-bin# ./bin/zkServer.sh start
ZooKeeper JMX enabled by default
Using config: /root/apache-zookeeper-3.8.4-bin/bin/../conf/zoo.cfg
Starting zookeeper ... STARTED
root@Agent2:~/apache-zookeeper-3.8.4-bin# ./bin/zkServer.sh start
ZooKeeper JMX enabled by default
Using config: /root/apache-zookeeper-3.8.4-bin/bin/../conf/zoo.cfg
Starting zookeeper ... STARTED
root@Agent3:~/apache-zookeeper-3.8.4-bin# ./bin/zkServer.sh start
ZooKeeper JMX enabled by default
Using config: /root/apache-zookeeper-3.8.4-bin/bin/../conf/zoo.cfg
Starting zookeeper ... STARTED

   验证Zookeeper 集群运行状态:
   分别在Agent1、Agent2和Agent3上执行jps命令, 以验证是否有QuorumPeerMain进程。 如果都有则说明Zookeeper集群启动成功了, 否则需要到对应机器中Zookeeper的logs目录下查看tail -f -n 200 /root/apache-zookeeper-3.8.4-bin/logs/zookeeper*-*.out日志文件中的报错信息。

root@Agent1:~# jps
159674 Jps
159628 QuorumPeerMain

   zookeeper leader角色主机会开启288838882181三个端口,其余follower节点只有38882181两个端口

root@Agent3:~/apache-zookeeper-3.8.4-bin# ./bin/zkServer.sh status
ZooKeeper JMX enabled by default
Using config: /root/apache-zookeeper-3.8.4-bin/bin/../conf/zoo.cfg
Client port found: 2181. Client address: localhost. Client SSL: false.
Mode: leaderroot@Agent3:~/apache-zookeeper-3.8.4-bin# ss -tnlp
State      Recv-Q     Send-Q         Local Address:Port          Peer Address:Port    Process
LISTEN     0          50                         *:3888                     *:*        users:(("java",pid=177883,fd=65))
LISTEN     0          50                         *:2181                     *:*        users:(("java",pid=177883,fd=53))
LISTEN     0          50                         *:2888                     *:*        users:(("java",pid=177883,fd=67))

3.2. 安装Kafka集群

   官方地址:https://kafka.apache.org/downloads

   kafka1安装:

root@Agent1:~# tar xf /opt/kafka_2.13-3.7.0.tgz -C /root
root@Agent1:~# cd kafka_2.13-3.7.0/config/
root@Agent1:~/kafka_2.13-3.7.0/config# vim server.properties

   修改Kafka的配置文件:
   主要修改server.properties配置文件中的broker.idlog.dirszookeeper.connect参数

broker.id=0
log.dirs=/data/kafka-logs
#zookeeper.connect=localhost:2181
zookeeper.connect=Agent1:2181,Agent2:2181,Agent2:2181root@Agent1:~# mkdir -p /data/kafka-logs

   ● broker.id:Kafka集群中Broker的编号, 默认是从0开始的, 所以Agent1主机中的broker.id 值为 0。
   ● log.dirs: Kafka 中的数据存储目录。 建议指定到存储空间比较大的磁盘中, 因为在实际工作中 Kafka 中会存储很多数据。
   ● zookeeper.connect: Zookeeper 集群的地址, 多个地址之间使用逗号分隔。

root@Agent1:~/kafka_2.13-3.7.0# bin/kafka-server-start.sh -daemon config/server.properties
root@Agent1:~/kafka_2.13-3.7.0# jps
160227 Jps
159628 QuorumPeerMain
160141 Kafka

   同样的方式安装Agent2和Agent3主机,并修改broker.id参数的值

   kafka2安装:

root@Agent2:~/kafka_2.13-3.7.0/config# vim server.properties
broker.id=1
log.dirs=/data/kafka-logs
zookeeper.connect=Agent1:2181,Agent2:2181,Agent2:2181
root@Agent2:~/kafka_2.13-3.7.0# bin/kafka-server-start.sh -daemon config/server.properties
root@Agent2:~/kafka_2.13-3.7.0# jps
154018 Kafka
153505 QuorumPeerMain
154070 Jps

   kafka3安装:

root@Agent3:~/kafka_2.13-3.7.0/config# vim server.properties
broker.id=2
log.dirs=/data/kafka-logs
zookeeper.connect=Agent1:2181,Agent2:2181,Agent2:2181
root@Agent3:~/kafka_2.13-3.7.0# bin/kafka-server-start.sh -daemon config/server.properties
root@Agent3:~/kafka_2.13-3.7.0# jps
162163 Jps
162116 Kafka
161589 QuorumPeerMain

   启动Kafka集群:

root@Agent1:~/kafka_2.13-3.7.0# bin/kafka-server-start.sh -daemon config/server.properties
root@Agent2:~/kafka_2.13-3.7.0# bin/kafka-server-start.sh -daemon config/server.properties
root@Agent3:~/kafka_2.13-3.7.0# bin/kafka-server-start.sh -daemon config/server.properties

   验证Kafka集群的运行状态:
   分别在Agent1、 Agent2和Agent3上执行jps命令验证是否有Kafka进程, 如果都有则说明Kafka集群启动成功了, 否则需要到对应的机器上查看 Kafka 的日志信息

root@Agent1:~/kafka_2.13-3.7.0# jps
924314 Jps
507311 QuorumPeerMain
160799 Kafka
169837 Application

3.3. 生产者和消费者使用

3.3.1. 生产者使用

   创建Topic:
   在安装好Kafka集群之后, 还需要先在Kafka中创建Topic, 之后就可以基于 Kafka 生产和消费数据了。

root@Agent3:~/kafka_2.13-3.7.0# bin/kafka-topics.sh --create --topic hello --partitions 5 --replication-factor 2 --bootstrap-server localhost:9092
Created topic hello.

   ● --create: 创建 Topic。
   ● --zookeeper: 指定Kafka集群使用的Zookeeper集群地址, 指定1个或者多个都可以, 多个用逗号分隔。
   ● --partitions: 指定Topic中的分区数量。
   ● --replication-factor: 指定Topic中分区的副本因子, 这个参数的值需要小于或等于Kafka集群中Broker的数量。
   ● --topic: 指定Topic的名称

root@Agent3:~/kafka_2.13-3.7.0# bin/kafka-topics.sh --describe --topic hello --bootstrap-server localhost:9092
Topic: hello    TopicId: wvzIUg03RN6FMc9iA4-QNw PartitionCount: 5       ReplicationFactor: 2    Configs:Topic: hello    Partition: 0    Leader: 0       Replicas: 0,1   Isr: 0,1Topic: hello    Partition: 1    Leader: 2       Replicas: 2,0   Isr: 2,0Topic: hello    Partition: 2    Leader: 1       Replicas: 1,2   Isr: 1,2Topic: hello    Partition: 3    Leader: 0       Replicas: 0,2   Isr: 0,2Topic: hello    Partition: 4    Leader: 2       Replicas: 2,1   Isr: 2,1

   启动基于控制台的生产者并向指定 Topic 中生产数据:
   Kafka 默认提供了基于控制台的生产者 , 直接使用Kafka的bin目录下的kafka-console-producer.sh 即可, 方便测试
   启动基于控制台的生产者之后, 生产测试数据: hehe

root@Agent3:~/kafka_2.13-3.7.0# bin/kafka-console-producer.sh --topic hello --bootstrap-server localhost:9092
>hehe
>

   ● broker-list: 指定Kafka集群的地址, 指定1个或者多个都可以, 指定多个时用逗号隔开。
   ● topic: 指定要生产数据的Topic名称

3.3.1. 消费者使用

   kafka默认提供了基于控制台的消费者, 直接使用Kafka 的 bin目录下的kafka-consoleconsumer.sh即可, 方便测试。

root@Agent1:~# cd kafka_2.13-3.7.0/
root@Agent1:~/kafka_2.13-3.7.0# bin/kafka-console-consumer.sh --topic hello --from-beginning --bootstrap-server localhost:9092
hehe

   ● bootstrap-server: 指定Kafka集群的地址, 指定1个或者多个都可以, 指定多个时用逗号分隔。
   ● topic: 指定要消费数据的Topic名称
   ● --from-beginning:Kafka消费者默认是消费最新生产的数据, 如果想消费之前生产的数据, 则需要添加参数–from-beginning, 表示从头消费。

   Kafka的生产者和消费者也可以使用Java代码来实现。 不过在实际工作中并不会经常这么用, 因为和 Kafka 经常对接使用的技术框架(例如 Flume) 已经内置了对应的消费者和生产者代码, 在使用时只需要进行简单的配置即可。
   kafka和zookeeper对接成功之后,可以在zookeeper中查看kafka的信息:

[zk: localhost:2181(CONNECTED) 5] ls /brokers
[ids, seqid, topics][zk: localhost:2181(CONNECTED) 1] ls /brokers/ids
[0, 1, 2][zk: localhost:2181(CONNECTED) 2] get /brokers/ids/0
{"features":{},"listener_security_protocol_map":{"PLAINTEXT":"PLAINTEXT"},"endpoints":["PLAINTEXT://Agent1:9092"],"jmx_port":-1,"port":9092,"host":"Agent1","version":5,"timestamp":"1711374513109"}

4. Kafka Controller控制器

   kafka控制器的作用是在Zookeeper的帮助下管理和协调整个Kafka集群,任意一台Broker可以成为Controller,但是只有一台会Controller
   leader和控制器可以不是同一台机器。ZooKeeper集群中会有一个选举过程来决定哪台服务器作为Leader,这个选举是基于ZooKeeper的内部机制进行的。一旦选举完成,Leader就负责处理所有的客户端请求,并确保集群中的其他服务器(被称为Follower)与它保持同步。
   控制器负责维护集群的元数据信息,处理客户端的创建、删除节点等操作,并将这些操作同步到其他服务器上。因此,控制器和Leader虽然有所区别,但在实践中,控制器通常是Leader,因为Leader负责处理所有客户端请求。但理论上,它们可以在不同的服务器上。

   控制器的选取:
   Broker启动时,会尝试创建ZK中 /controller znode,第一个创建/controller的Broker会是Controller

[zk: localhost:2181(CONNECTED) 2] get /controller
{"version":2,"brokerid":1,"timestamp":"1711374505881","kraftControllerEpoch":-1}

   控制器的功能:

   ● topic管理:当使用kafka-topics脚本时,后台工作通过 controller 完成
   ● 分区重分配:对已有 topic 分区进行细粒度的分配
   ● Preferred Leader 选举:Kafka为了避免部分Broker负载过重而提供的一种换Leader的方案
   ● 集群成员管理(新增 Broker、Broker 主动关闭、Broker 宕机):自动检测 Broker、controller 通过 watch 机制检查 ZK 的 /brokers/ids 子节点数量变更
   ● 数据服务:controller向其他Broker提供数据服务、controller上保存了最全的集群元数据
在这里插入图片描述

   控制器故障转移Failover:
   Controller存在单点
   故障转移:当运行中的Controller突然宕机,Kafka能够快速感知,并立即启用备用Controller代替之前失败的Controller,这个过程称为 Failover
在这里插入图片描述

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.mzph.cn/news/786365.shtml

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

电商运营自动化新里程:取数宝引领数字化转型实践

随着电子商务行业的高速发展及复杂化,精细化运营已成为电商企业提升竞争力的关键所在。尤其是在海量数据处理与实时分析方面,自动化工具的引入对企业管理和决策带来了革命性变化。其中,“取数宝”作为一种先进的电商运营自动化解决方案&#…

dockerfile制作-pytoch+深度学习环境版

你好你好! 以下内容仅为当前认识,可能有不足之处,欢迎讨论! 文章目录 文档内容docker相关术语docker常用命令容器常用命令根据dockerfile创建容器dokerfile文件内容 docker问题:可能的原因和解决方法示例修改修改后的D…

解析Apache Kafka:在大数据体系中的基本概念和核心组件

关联阅读博客文章:探讨在大数据体系中API的通信机制与工作原理 关联阅读博客文章:深入解析大数据体系中的ETL工作原理及常见组件 关联阅读博客文章:深度剖析:计算机集群在大数据体系中的关键角色和技术要点 关联阅读博客文章&a…

LuaJIT源码分析(二)数据类型

LuaJIT源码分析(二)数据类型 LuaJIT支持的lua数据类型和官方的lua 5.1版本保持一致,它的源文件中也有一个lua.h: // lua.h /* ** basic types */ #define LUA_TNONE (-1)#define LUA_TNIL 0 #define LUA_TBOOLEAN 1 #define L…

【数据结构】顺序表的实现——动态分配

🎈个人主页:豌豆射手^ 🎉欢迎 👍点赞✍评论⭐收藏 🤗收录专栏:数据结构 🤝希望本文对您有所裨益,如有不足之处,欢迎在评论区提出指正,让我们共同学习、交流进…

vscode通过ssh连接服务器(吐血总结)

一、通过ssh连接服务器 1、打开vscode,进入拓展(CtrlShiftX),下载拓展Remote - SSH。 2、点击远程资源管理器选项卡,选择远程(隧道/SSH)类别。 3、点击SSH配置。 4、在中间上部分弹出的配置文件…

LangChain入门:9.使用FewShotPromptTemplate实现智能提示工程

在构建智能提示工程时,LangChain 提供了强大的 FewShotPromptTemplate 模型,它可以帮助我们更好地利用示例来指导大模型生成更加优质的提示。 在这篇博文中,我们将使用 LangChain 的 FewShotPromptTemplate 模型来设计一个智能提示工程&#…

游戏引擎中的粒子系统

一、粒子基础 粒子系统里有各种发射器(emitter),发射器发射粒子(particle)。 粒子是拥有位置、速度、大小尺寸、颜色和生命周期的3D模型。 粒子的生命周期中,包含产生(Spawn)、与环…

AcrelEMS-EV 汽车制造能效管理系统解决方案

安科瑞电气股份有限公司 祁洁 15000363176 一、行业现状 1、政府、市场越来越关注碳排放指标。 2、用能设备缺乏完整的在线监视分析系统,无法及时发现用能异常和能源利用效率。 3、不能生产全流程监测和分析能源利用水平,无法及时发现浪费。 4、用…

用计算困难问题的视角看密码学算法

从计算困难问题的视角看密码学算法 计算困难问题是理论计算机和密码学的交叉论题,密码学的加密算法都基于计算困难问题(一般来说是NP-Complete和NP-Hard问题),在这篇文章里我们将讨论计算困难问题和各种加密算法的关系,从而引出我们的观点:密码学算法其实就是利用验证容易但是求…

Hippo4j线程池实现技术

文章目录 🔊博主介绍🥤本文内容部署运行模式集成线程池监控配置参数默认配置 📢文章总结📥博主目标 🔊博主介绍 🌟我是廖志伟,一名Java开发工程师、Java领域优质创作者、CSDN博客专家、51CTO专家…

力扣热题100_链表_142_环形链表 II

文章目录 题目链接解题思路解题代码 题目链接 142. 环形链表 II 给定一个链表的头节点 head ,返回链表开始入环的第一个节点。 如果链表无环,则返回 null。 如果链表中有某个节点,可以通过连续跟踪 next 指针再次到达,则链表中…

【Spring】之AOP详解

AOP 什么是AOP? AOP:Aspect Oriented Programming,面向切面编程。 切面指的是某一类特定问题,因此面向切面编程也可以理解为面向特定方法编程。例如,在任何一个系统中,总有一些页面不是用户可以随便访问…

Matlab|配电网三相不平衡潮流计算【隐式Zbus高斯法】【可设定变压器数量、位置、绕组方式】

目录 主要内容 部分代码 结果一览 1.以33节点为例 2.以12节点系统为例 下载链接 主要内容 该模型基于隐式Zbus高斯法实现对配电网的三相不平衡潮流计算,通过选项可实现【不含变压器】和【含变压器】两种方式下的潮流计算,并且通过参数设置…

游戏引擎中的声音系统

一、声音基础 1.1 音量 声音振幅的大小 压强p:由声音引起的与环境大气压的局部偏差 1.2 音调 1.3 音色 1.4 降噪 1.5 人的听觉范围 1.6 电子音乐 将自然界中连续的音乐转换成离散的信号记录到内存中 采样 - 量化 - 编码 香农定理:采样频率是信…

如何查询网站是否被搜索引擎收录

怎么看网站有没有被百度收录 对于网站所有者来说,了解自己的网站是否被百度搜索引擎收录是非常重要的。只有被收录,网站才能在百度搜索结果中展现,从而获取流量和曝光。下面介绍几种方法,让您快速了解自己的网站是否被百度收录。…

SpringBoot+uniApp宠物领养小程序系统 附带详细运行指导视频

文章目录 一、项目演示二、项目介绍三、运行截图四、主要代码1.保存宠物信息代码2.提交订单信息代码3.查询评论信息代码 一、项目演示 项目演示地址: 视频地址 二、项目介绍 项目描述:这是一个基于SpringBootuniApp框架开发的宠物领养微信小程序系统。…

Elasticsearch的倒排索引是什么?

文章目录 什么是ES?什么是倒排索引?为什么叫做倒排索引?分词器的使用 什么是ES? Elasticsearch是基于 Apache Lucene【lusen】的搜索引擎,支持Restful API风格【可以使用常见的HTTP请求来访问】,并且搜索速…

在js中本地存储的数组如何转成对象

一、此方法仅适用一维数组; 二、效果图 使用后 三、js代码。 function gong(s){console.log(s);let data;let kk1;// 检查ask_id是否不为空 if (s.ask_id null ) { kk1}else{kk2let dd;dds.data;sessionStorage.setItem(wenda,JSON.stringify(dd[0]))window.l…

1、jvm基础知(一)

什么是JVM? 1、定义:JVM 指的是Java虚拟机( Java Virtual Machine )。JVM 本质上是一个运行在计算机上的程序, 他的职责是运行Java字节码文件,Java虚拟机上可以运行Java、Kotlin、Scala、Groovy等语言 任…