kafka原理详解

消息队列概述

消息队列分类

点对点

  • 组成:消息队列(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发送数据的流程(六步)

一共六步:

  1. 生产者查询Leader:producer先从zookeeper的“/brokers/.../state”节点找到该partition的leader

  2. 找到Leader之后往Leader写数据:producer将消息发送给该leader

  3. Leader落盘:leader将消息写入本地log

  4. Leader通知Follower

  5. Follower从Leader中拉取数据
    :replication写入到Follower的本地log后,follower向leader发送ack

  6. Kafka向生产者回应ACK:leader收到所有的replication的ack之后,向producer发送ack

Kafka选择分区的模式(三种)

  1. 直接指定往哪个分区写

  2. 指定key,然后kafka根据key做hash后决定写哪个分区

  3. 各个分区轮询

生产者往kafka发送数据的模式(三种)

  1. 把数据发送给Leader就认为成功,效率最高,安全性低

  2. 把数据发送给Leader,等待Leader回复Ack后则认为发送成功

  3. 把数据发送给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:

  1. 查找segment file:每一个segment文件名都包含了上一个segment最后一条消息的offset值,所以只要根据offset二分查找文件列表,就能定位到具体segment文件

  2. 通过segment file查找message:当定位到segment文件后,可以通过对应的.index元数据文件,在对应的.log文件中顺序查找对应的offset,然后即可拿到数据

3)说明:

  • kafka只能保证在同一个partition内部消息是有序的,在不同的partition之间,并不能保证消息有序

  • 为什么kafka快:因为它把对磁盘的随机读变成了顺序读

kafka安装部署及操作

kafka单机部署

安装ZooKeeper

kafka需要依赖ZooKeeper,所以需要先安装并启动ZooKeeper,kafka使用zk有两种方式:

  1. 使用kafka自带的ZooKeeper(一般不推荐使用内置的ZooKeeper)

  2. 单独搭建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:

  1. 使用kafka自带的ZooKeeper(一般不推荐使用内置的ZooKeeper)

  2. 单独搭建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)消费消息

使用kafka自带的消费者命令消费消息 (可开多个窗口,模拟消费者)

 # 消费者消费消息,也是从指定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Ͱ

标签: none

添加新评论