消息队列分类
点对点
-
组成:消息队列(Queue)、发送者(Sender)、接收者(Receiver)
-
特点:一个生产者生产的消息只能被一个接受者接收,消息一旦被消费,消息就不在消息队列中了
发布/订阅
-
组成:消息队列(Queue)、发布者(Publisher)、订阅者(Subscriber)、主题(Topic)
-
特点:每个消息可以有多个消费者,彼此互不影响,即发布到消息队列的消息能被多个接受者(订阅者)接收
常见的消息系统
-
ActiveMQ: 历史悠久,支持性较好,性能相对不高
-
RabbitMQ: 可靠性高、安全
-
Kafka: 分布式、高性能、高吞吐量、跨语言
-
RocketMQ: 阿里开源的消息中间件,纯Java实现
kafka架构
kafka介绍
Kafka是一个分布式的发布/订阅消息系统,最初由LinkedIn(领英)公司发布,使用Scala语言编写,后成为Apache的顶级项目。
kafka主要用于处理活跃的数据,如登录、浏览、点击、分享等用户行为产生的数据。
kafka架构组成
Broker
-
broker表示kafka的节点,kafka集群包含多个kafka服务节点,每个kafka服务节点就称为一个broker
Topic
-
主题,用来存储不同类别的消息(kafka的消息数据是分主题存储在硬盘上的)
-
存储消息时,需要指定存储在哪个主题下面,如发帖,发哪种类型的
Partition
-
分区,每个topic包含一个或多个partition,在创建topic时指定包含的partition数据(目的是为了进行分布式存储)
-
分区可以提高负载(每个分区是不同的磁盘,所以会提高负载)
Replication
-
副本,每个partition分区可以有多个副本,分布在不同的Broker上
-
kafka会选出一个副本作为Leader,所有的读写请求都会通过Leader完成,Follower只负责备份数据
-
所有Follower会自动从Leader中复制数据,当Leader宕机后,会从Follower中选出一个新的Leader继续提供服务,实现故障自动转移
Message
-
消息,是通信数据的基本单位,每个消息都属于一个Partition,消息都是放在Partition里面的
Producer
-
消息的生产者,向kafka的一个topic发布消息,发布消息时,需要指定发布到哪个topic主题
Consumer
-
消息的消费者,订阅Topic并读取其发布的消息,消费或订阅哪个topic主题里的消息,可以订阅多个主题的消息(类似订阅多个微信公众号)
Consumer Group
-
消费者组,每个Consumer属于一个特定的Consumer Group,多个Consumer可以属于同一个Consumer Group
-
各个consumer可以组成一个组,每个消息只能被组中的一个consumer消费,如果一个消息可以被多个consumer消费的话,那么这些consumer必须在不同的组。
ZooKeeper
-
协调Kafka的正常运行,kafka将元数据信息保存在ZooKeeper中,但发送给Topic本身的消息数据并不存储在ZK中,而是存储在磁盘文件中
-
元数据信息包括:kafka有多少个节点、有哪些主题,主题叫什么,有哪些分区的等(消息自身的数据不在ZK中,而是在磁盘上对应的分区中)
kafka的工作流程
生产者向kafka发送数据的流程(六步)
一共六步:
-
生产者查询Leader:producer先从zookeeper的“/brokers/.../state”节点找到该partition的leader
-
找到Leader之后往Leader写数据:producer将消息发送给该leader
-
Leader落盘:leader将消息写入本地log
-
Leader通知Follower
-
Follower从Leader中拉取数据:replication写入到Follower的本地log后,follower向leader发送ack
-
Kafka向生产者回应ACK:leader收到所有的replication的ack之后,向producer发送ack
Kafka选择分区的模式(三种)
-
直接指定往哪个分区写
-
指定key,然后kafka根据key做hash后决定写哪个分区
-
各个分区轮询
生产者往kafka发送数据的模式(三种)
-
把数据发送给Leader就认为成功,效率最高,安全性低
-
把数据发送给Leader,等待Leader回复Ack后则认为发送成功
-
把数据发送给Leader,确保Follower从Leader拉取数据回复Ack给Leader,Leader再向生产者回复Ack才认为发送成功,安全性最高
数据消费
多个消费者可以组成一个消费者组,并用一个标签来标识这个消费者组(一个消费者实例可以运行在不同的进程甚至不同的服务器上)
-
如果所有的消费者实例都在同一个消费者组中,那么消息记录会被很好的均衡发送到每个消费者实例
-
如果所有的消费者实例都在不同的消费者组,那么每一条消息记录会被广播到每一个消费者实例
各个consumer可以组成一个组,每个消息只能被组中的一个consumer消费,如果一个消息可以被多个consumer消费的话,那么这些consumer必须在不同的组
注意:每个消费者实例可以消费多个分区,但是每一个分区最多只能被消费者组中的一个实例消费
kafka的文件存储机制
topic、partition和segment
1)在kafka文件存储中,同一个topic下有多个不同的partition:
-
每个partition就是一个目录,partition的命名规则为:topic名称+有序序号
-
第一个partition序号从0开始,序号最大值为partition数量减一
2)每个partition的目录下面会有多组segment文件:
-
每个partition(目录)相当于一个巨型大文件被平均分配到多个大小都相等的segment数据文件中(但每个segment file消息数量不一定相等,这种特性方便old segment file快速被删除)
-
每组segment文件包含:.index文件、.log文件、.timeindex文件(.log文件就是实际存储message的地方,.index和.timeindex文件为索引文件,用于检索消息)
-
每个partition只需要支持顺序读写就行了,segment文件生命周期由服务端配置参数决定
-
这样做能快速删除无用文件,有效提高磁盘利用率
3)segment文件
-
segment文件由2大部分组成,分别为index file和data file,此2个文件一一对应,成对出现,后缀".index"和“.log”分别表示为segment索引文件、数据文件
-
segment文件命名规则:partion全局的第一个segment从0开始,后续每个segment文件名为上一个segment文件最后一条消息的offset值。数值最大为64位long大小,19位数字字符长度,没有数字用0填充
存储和查找message的过程
1)数据写入过程
每个Partition都是一个有序并且不可改变的消息记录集合(每个partition都是一个有序队列),当新的数据写入时,就被追加到partition的末尾。
在每个partition中,每条消息都会被分配一个顺序的唯一标识,这个标识被称为Offset(偏移量),用于partition唯一标识一条消息。
2)数据查找过程
在partition中通过offset查找message:
-
查找segment file:每一个segment文件名都包含了上一个segment最后一条消息的offset值,所以只要根据offset二分查找文件列表,就能定位到具体segment文件
-
通过segment file查找message:当定位到segment文件后,可以通过对应的.index元数据文件,在对应的.log文件中顺序查找对应的offset,然后即可拿到数据
3)说明:
-
kafka只能保证在同一个partition内部消息是有序的,在不同的partition之间,并不能保证消息有序
-
为什么kafka快:因为它把对磁盘的随机读变成了顺序读
kafka安装部署及操作
kafka单机部署
安装ZooKeeper
kafka需要依赖ZooKeeper,所以需要先安装并启动ZooKeeper,kafka使用zk有两种方式:
-
使用kafka自带的ZooKeeper(一般不推荐使用内置的ZooKeeper)
-
单独搭建ZooKeeper
使用kafka自带的ZooKeeper:
# kafka的bin目录中,有自带的zk的启动命令 /usr/local/kafka/bin/zookeeper-server-start.sh # kafka的config目录中,有自带的zk的配置文件 /usr/local/kafka/bin/zookeeper.properties
如果要使用kafka内置的ZooKeeper,修改好配置文件 ./config/zookeeper.properties
(主要修改zk的data位置和端口),直接启动即可
# 后台启动,并指定配置文件 zookeeper-server-start.sh -daemon ../config/zookeeper.properties
安装kafka
kafka需要java环境,需要安装jdk
# 1.安装jdk yum install -y java-1.8.0-openjdk # 2.准备kafka安装包 tar zxvf kafka_2.11-2.2.0.tgz -C /usr/local/ ln -s /usr/local/kafka_2.11-2.2.0 /usr/local/kafka mkdir -pv /data/kafka/data/ # 创建kafka数据存储目录 # 配置环境变量 sed -i '$aPATH="/usr/local/kafka/bin:$PATH"' /etc/profile source /etc/profile # 3.修改kafka配置文件 vim /usr/local/kafka/config/server.properties listeners=PLAINTEXT://10.0.0.80:9092 # kafka默认监听端口号为9092, log.dirs=/data/kafka/data # 指定kafka数据存放目录 zookeeper.connect=localhost:2181 # 指定ZooKeeper地址,kafka要将元数据存放到zk中,这里会在本机启动一个zk # 4.启动kafka kafka-server-start.sh -daemon /usr/local/kafka/config/server.properties # 5.查看进程及端口 ps -ef | grep kafka ss -tnl | grep 9092 # kafka监听在9092端口
kafka脚本程序及配置文件
几个kafka的操作脚本
-
kafka-server-start.sh kafka启动程序
-
kafka-server-stop.sh kafka停止程序
-
kafka-topics.sh 创建topic程序
-
kafka-console-producer.sh 命令行模拟生产者生产消息数据程序
-
kafka-console-consumer.sh 命令行模拟消费者消费消息数据程序
kafka的配置文件
vim /usr/local/kafka/config/server.properties
############################# Server Basics ############################# # broker的id,值为整数,且必须唯一,在一个集群中不能重复,默认为0 broker.id=0 ############################# Socket Server Settings ############################# # kafka默认监听的端口为9092 #listeners=PLAINTEXT://:9092 # 处理网络请求的线程数量,默认为3个 num.network.threads=3 # 执行磁盘IO操作的线程数量,默认为8个 num.io.threads=8 # socket服务发送数据的缓冲区大小,默认100KB socket.send.buffer.bytes=102400 # socket服务接受数据的缓冲区大小,默认100KB socket.receive.buffer.bytes=102400 # socket服务所能接受的一个请求的最大大小,默认为100M socket.request.max.bytes=104857600 ############################# Log Basics ############################# # kafka存储消息数据的目录 log.dirs=../data # 每个topic默认的partition数量 num.partitions=1 # 在启动时恢复数据和关闭时刷新数据时每个数据目录的线程数量 num.recovery.threads.per.data.dir=1 ############################# Log Flush Policy ############################# # 消息刷新到磁盘中的消息条数阈值 #log.flush.interval.messages=10000 # 消息刷新到磁盘中的最大时间间隔 #log.flush.interval.ms=1000 ############################# Log Retention Policy ############################# # 日志保留小时数,超时会自动删除,默认为7天 log.retention.hours=168 # 日志保留大小,超出大小会自动删除,默认为1G,log.retention.bytes这是指定 Broker 为消息保存的总磁盘容量大小 #log.retention.bytes=1073741824 # 日志分片策略,单个日志文件的大小最大为1G,超出后则创建一个新的日志文件 log.segment.bytes=1073741824 # 每隔多长时间检测数据是否达到删除条件 log.retention.check.interval.ms=300000 ############################# Zookeeper ############################# # Zookeeper连接信息,如果是zookeeper集群,则以逗号隔开 zookeeper.connect=localhost:2181 # 连接zookeeper的超时时间 zookeeper.connection.timeout.ms=6000 # 是否可以删除topic,默认为false delete.topic.enable=true
kafka集群部署
环境信息
节点 | IP | ZK Port | Kafka Port | OS |
---|---|---|---|---|
node01 | 10.0.0.80 | 2181 | 9092 | CentOS7.9 |
node02 | 10.0.0.81 | 2181 | 9092 | CentOS7.9 |
node03 | 10.0.0.82 | 2181 | 9092 | CentOS7.9 |
部署ZooKeeper集群
kakfa依赖ZooKeeper,可以用以下两种方式使用ZooKeeper:
-
使用kafka自带的ZooKeeper(一般不推荐使用内置的ZooKeeper)
-
单独搭建ZooKeeper
搭建ZooKeeper集群见ZooKeeper文档。
部署kafka集群
所有节点(node01、node02、node03)上操作:
# 1.安装jdk yum install -y java-1.8.0-openjdk # 2.准备kafka安装包 tar zxvf kafka_2.11-2.2.0.tgz -C /usr/local/ ln -s /usr/local/kafka_2.11-2.2.0 /usr/local/kafka mkdir -pv /data/kafka/data/ # 创建kafka数据存储目录 # 配置环境变量 sed -i '$aPATH="/usr/local/kafka/bin:$PATH"' /etc/profile source /etc/profile # 3.修改kafka配置文件 broker.id=1 # 各自节点的id号,每个节点都有自己的id,值为整数,且必须唯一,在一个集群中不能重复,默认为0 listeners=PLAINTEXT://10.0.0.80:9092 # kafka默认监听的端口号为9092,指定各自节点的地址和端口 log.dirs=/data/kafka/data # 指定kafka数据的存放目录 zookeeper.connect=10.0.0.80:2181,10.0.0.81:2181,10.0.0.82:2181 # zookeeper的连接信息,kafka要将元数据信息存放到zk中 zookeeper.connection.timeout.ms=600000 #连接zk超时时间调大,否则可能起不来,默认: 6000 # 4.启动kafka kafka-server-start.sh -daemon /usr/local/kafka/config/server.properties # 5.查看进程及端口 ps -ef | grep kafka ss -tnl | grep 9092 # kafka监听在9092端口
生产和消费消息测试
-
kafka-server-start.sh kafka启动程序
-
kafka-server-stop.sh kafka停止程序
-
kafka-topics.sh 创建topic程序
-
kafka-console-producer.sh 命令行模拟生产者生产消息数据程序
-
kafka-console-consumer.sh 命令行模拟消费者消费消息数据程序
topic相关操作
操作topic使用kafka-topic.sh
脚本
# 查看主题topic列表,需指定zk的地址 kafka-topics.sh --list --zookeeper 10.0.0.80:2181 # 创建topic hello kafka-topics.sh --create --zookeeper 10.0.0.80:2181 --replication-factor 1 --partitions 3 --topic hello # --create 是创建主题topic # --zookeeper localhost:2181 主题topic信息是存储在zk中,需要指定zk服务的地址 # --replication-factor 1 主题topic信息的副本数,因为现在只要一个节点,所以只能是1,有多个节点时候,可以指定副本数多个 # --partitions 3 主题topic有多少个分区 # --topic test-topic 指定主题topic的名字 # 查看某个具体的主题topic消息 kafka-topics.sh --describe --zookeeper 10.0.0.80:2181 --topic hello # 修改主题topic信息,增加到5个分区 kafka-topics.sh --alter --zookeeper 10.0.0.80:2181 --topic hello --partitions 5 # 删除主题topic hello kafka-topics.sh --delete --zookeeper 10.0.0.80:2181 --topic hello
生产和消费命令
-
生产消息:
kafka-console-producer.sh
-
消费消息:
kafka-console-consumer.sh
1)生产消息
使用kafka自带的生产者命令生产消息 (可开一个窗口,模拟生产者)
# 生产者生产消息,是往topic里发送消息的,需要指明kafka地址和topic的名字 kafka-console-producer.sh --broker-list 10.0.0.80:9092 --topic test-topic >hello >test1 >test2 >
2)消费消息文章来源:https://www.toymoban.com/news/detail-784824.html
使用kafka自带的消费者命令消费消息 (可开多个窗口,模拟消费者)文章来源地址https://www.toymoban.com/news/detail-784824.html
# 消费者消费消息,也是从指定topic里取出的,需要指明kafka地址和topic的名字,加--from-beginning是从头开始收,不加就从当前状态开始收 kafka-console-consumer.sh --bootstrap-server 10.0.0.80:9092 --topic test-topic --from-beginning
查看消息本体及相关数据
查看kafka存放的消息
# 来到kafka的数据目录,查看kafka存放的消息 cd /data/kafka/data/ ls -d ./test-topic* # kafka存放的消息会被分布存储在各个分区,这里目录名test-topic就表示对应的topic名称,后缀-0就表示对应的分区 ./test-topic-0 # 有几个分区就会有几个这样的目录,消息被分布存储在各个目录(目录名称格式: topic名称-分区编号) # 查看对应分区下的文件(每个分区中存放的消息内容都不一样) ls ./test-topic-0/ 00000000000000000000.index 00000000000000000000.log 00000000000000000000.timeindex leader-epoch-checkpoint # 查看消息本体 cat ./test-topic-0/00000000000000000000.log =CͰÿÿÿÿÿÿÿÿÿÿÿÿÿÿ hello=Mľ5ÿÿÿÿÿÿÿÿÿÿÿÿÿÿ test1<{y輁ÿÿÿÿÿÿÿÿÿÿÿÿÿexit<.پÿÿÿÿÿÿÿÿÿÿÿÿÿquit=徐±ÿÿÿÿÿÿÿÿÿÿÿÿÿÿ hello=Hڥÿÿÿÿÿÿÿÿÿÿÿÿÿÿ test1=z ȁÿÿÿÿÿÿÿÿÿÿÿÿÿÿ test2BΘ艁ÿÿÿÿÿÿÿÿÿÿÿÿÿÿ hahahahaha
查看kafka存放在ZooKeeper中的元数据
# 客户端连接zk zkCli.sh # 如果什么都不指定,则默认连接本机的2181端口 # zkCli.sh -server 10.0.0.80:2181 # 指定IP和端口,可以连接集群中任何一个节点 # 查看/根节点下有哪些数据 [zk: localhost:2181(CONNECTED) 0] ls / [mytest, cluster, controller, brokers, zookeeper, admin, isr_change_notification, log_dir_event_notification, controller_epoch, testNode, consumers, latest_producer_id_block, config] # 查看/brokers下的数据 [zk: localhost:2181(CONNECTED) 1] ls /brokers [ids, topics, seqid] # 查看当前brokers的节点编号 [zk: localhost:2181(CONNECTED) 2] ls /brokers/ids [0] # 查看主题topic [zk: localhost:2181(CONNECTED) 3] ls /brokers/topics [test_conf, test-topic, xxxxxx, web_log, xxx_conf, __consumer_offsets, hg_test, aaa_conf] # 查看test-topic这个主题的分区 [zk: localhost:2181(CONNECTED) 4] ls /brokers/topics/test-topic [partitions] # 查看test-topic这个主题的分区情况 [zk: localhost:2181(CONNECTED) 5] ls /brokers/topics/test-topic/partitions [0] # 使用get命令查看test-topic这个主题的相关信息 [zk: localhost:2181(CONNECTED) 6] get /brokers/topics/test-topic {"version":1,"partitions":{"0":[0]}} cZxid = 0x200000147 ctime = Sat Mar 18 10:18:27 CST 2023 mZxid = 0x200000147 mtime = Sat Mar 18 10:18:27 CST 2023 pZxid = 0x200000148 cversion = 1 dataVersion = 0 aclVersion = 0 ephemeralOwner = 0x0 dataLength = 36 numChildren = 1
到了这里,关于kafka的原理及集群部署详解的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!