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发送数据的流程（六步）
一共六步：
生产者查询Leader：producer先从zookeeper的“/brokers/.../state”节点找到该partition的leader
找到Leader之后往Leader写数据：producer将消息发送给该leader
Leader落盘：leader将消息写入本地log
Leader通知Follower
Follower从Leader中拉取数据
：replication写入到Follower的本地log后，follower向leader发送ackKafka向生产者回应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=truekafka集群部署
环境信息
| 节点 | 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）消费消息
使用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Ͱ