Kafka3.0之前依赖于zookeeper Zookeeper开源,分布式的架构,提供协调服务(apache项目) 基于观察者模式设计的分布式服务管理架构 存储和管理数据,分布式节点上的服务接受观察者的注册,一旦分布式节点上的数据发生变化,由zookeeper开负责通知分布式节点上的服务 |
Zookeeper 分为领导者和追随者 leaderfoollower 只有一般以上的集群存活,zookeeper集群可以正常工作,适用于安装奇数台的服务集群 全局数据一致,每个zookeeper每个节点都保存相同的数据,维护监控服务的数据一致 数据更新的原子性,要么都成功,要么都失败。 |
Zookeeper的应用场景
1 | 统一命名服务,在分布式的环境下,对所有的应用和服务进行统一的命名 |
2 | 统一配置管理,配置文件同步,kafka的配置文件被修改,可以快速同步到其他节点 |
3 | 统一集群管理,实时掌握所有节点的状态 |
4 | 服务器动态上下线 |
5 | 负载均衡,把访问服务器的数据,发送到访问最少的服务器处理客户端的请求 |
zookeeper的选举机制
三台服务器 A B C A 先启动,发起第一次选举,投票投给自己,只有一票,A的状态是looking B 启动,再发起一次选举,A和B分别投主机一票,交换选票信息,myid-A发现myid-B的myid比A大,A的这一票转而投给B A 0 B 2 没有半数以上结果,A B会进入looking C 启动 MYID c的myid最大 A和B都会把票投给C A 0 B 0 C 3,C的状态变为leader,A和B变成follower 此时结构已确定,后续加的服务器都是追随者。 |
只有两种情况会开启选举机制
1 | 初始化的情况会产生选举 |
2 | 服务器之间的leader丢失了连接状态 |
Leader已经存在,建立连接即可,如Leader不存在
1 | 服务器ID大的胜出 |
2 | EPOCH大,直接胜出 |
3 | EPOCH相同,事务ID大的胜出 |
4 | EPOCH每个leader任期的代号,没有leader,大家的逻辑地址相同,每次投完一次之后,数据都是递增,事务id,表示服务器的每一次变更,每变更一次事务id变化一次 |
服务器ID zookeeper集群当中都有一个id,每台机器不重复,和myid保持一致 |
消息队列:kafka
为什么要引入消息队列这个机制(MQ) 首先它是一个中间件,他负责发消息,客户到中间到服务端,服务端到中间件再到客户端,在高并发环境下,同步请求来不及处理,来不及处理的请求会形成阻塞,比方说数据库就会形成行锁或者表锁,请求线程满了,超标了,too manyconnection,出现就会引发系统雪崩 |
消息队列的作用
异步处理请求,流量削峰,应用解耦 | |
耦合 | 在软件系统当中,修改一个组件,需要修改所有其他组件,高度耦合 |
低度耦合 | 修改其中一个组件,对其他组件影响不大,无需修改所有 |
解耦 | 只要通信保证,其他的修改不影响整个集群,每个组件可以独立的跨站,修改,降低组件之间的依赖性 |
可恢复性 | 系统当中的有一部分组件小时,不影响整个系统,也就是说在消息队列当中,即使有一个处理消息的进程失败,一旦恢复还可以重新加入到队列当中,继续处理消息 |
缓冲 | 可以控制和优化数据经过系统的时间和速度,解决生产消息和消费消息处理速度不一致问题 |
峰值的处理能力 | 消息队列在峰值情况之下,能够顶住突发的访问压力,避免了专门为了突发情况而对系统进行修改 |
异步通信 | 允许用户把一个消息放入队列,但是不立即处理,等用户想处理的时候再处理 |
消息队列的模式
点对点 一对一 | 消息的生产者发送消息到队列中,消费者从队列中提取消息,消费者提取完之后,队列将不再存储被提取的消息,队列中被提取的消息会被移除,后续消费者不能在消费队列当中的消息,消息队列可以有多个消费者,但是一个消息只能有一个消费者 |
发布/订阅模式 | 1对多,又叫观察者模式,消费者提取数据之后,队列当中的消息不会被清除 |
生产者发布一个消息到主题,所有消费者都是通过主题获取消息 |
主题 | topic topic类似一个数据流的管道,生产者把消息发布到主题,消费者从主题当中订阅数据,主题可以分区,每个分区都有自己的偏移量 |
分区 | partion 每个主题可以分成多个分区,每个分区都是数据的有序子集,分区可以允许kafka进行水平拓展,,以处理大量数据消息在分钟按照偏移量存储,消费者可以独立读取每个分区的数据 |
偏移量 | 是每个消息在分区中唯一的标识,消费者可以通过偏移量来跟踪获取已读或者未读消息的位置,也可以提交偏移量来记录已处理的信息 |
生产者 | producer生产者把数据发送kafka的主题当中,负责写入消息 |
消费者 | 从主题当中读取数据,消费者可以是一个也可以十多个,每个消费者有一个唯一的消费者组ID,kafka通过消费者实现负载均衡和容错性 |
经纪人 | broker,每个kafka节点都有一个borker,每个负责一个kafka服务器,id唯一,存储主题分区当中的数据,处理生产和消费者的请求,以及维护元数据(zookeeper),3.0之后不依赖zookeeper的核心,元数据由kafka节点自己管理 |
Zookeeper | zookeeper负责保存元数据,元数据就是topic的相关信息(发布在哪台主机,制定了多少分区,以及副本数,偏移量) |
Zookeeper会自建一个主题:_consumer_offsets |
|
Kafka的工作流程
实验
zookeeper安装
yum install -y java-1.8.0-openjdk java-1.8.0-openjdk-devel
tar -xf apache-zookeeper-3.5.7-bin.tar.gz
mv apache-zookeeper-3.5.7-bin zookeeper
cd zookeeper/conf/
cp zoo_sample.cfg zoo.cfgvim zoo.cfg2
tickTime=2000
#服务器与客户端之间心跳时间,2秒检测一次服务器和客户端之间的通信
6
initLimit=10
#领导者和追随者之间,初始连接时容忍的时间 10*2S 20S
10
syncLimit=5
同步超时时间,领导者和追随者之间,同步通信超时的时间,5*2=10s,leader会认为follower丢失,移除集群。
15
dataDir=/opt/zookeeper/data
#保存数据的目录,需要单独创建
dataLogDir=/opt/zookeeper/logs
19
clientPort=2181
34
server.1=192.168.10.100:3188:3288
server.2=192.168.10.101:3188:3288
server.3=192.168.10.102:3188:3288
注
server.X=192.168.10.xxx:3188:3288
X:每个zookeeper集群的初始myid。
192.168.10.xxx 服务器的IP地址
3188 领导者和追随者之间交换信息的端口(内部通信的端口)
3288 一旦leader丢失响应,开启选举,3288就是用来执行选举时的服务器之问通信端口。2
tickTime=2000
#服务器与客户端之间心跳时间,2秒检测一次服务器和客户端之间的通信
6
initLimit=10
#领导者和追随者之间,初始连接时容忍的时间 10*2S 20S
10
syncLimit=5
同步超时时间,领导者和追随者之间,同步通信超时的时间,5*2=10s,leader会认为follower丢失,移除集群。
15
dataDir=/opt/zookeeper/data
#保存数据的目录,需要单独创建
dataLogDir=/opt/zookeeper/logs
19
clientPort=2181
34
server.1=192.168.10.100:3188:3288
server.2=192.168.10.101:3188:3288
server.3=192.168.10.102:3188:3288
注
server.X=192.168.10.xxx:3188:3288
X:每个zookeeper集群的初始myid。
192.168.10.xxx 服务器的IP地址
3188 领导者和追随者之间交换信息的端口(内部通信的端口)
3288 一旦leader丢失响应,开启选举,3288就是用来执行选举时的服务器之问通信端口。//在每个节点上创建数据目录和日志目录
mkdir /opt/zookeeper/data
mkdir /opt/zookeeper/logscd /opt/zookeeper/data/
//在每个节点的dataDir指定的目录下创建一个 myid 的文件,不同节点分配1、2、3
echo 1 > /opt/zookeeper/data/myid
echo 2 > /opt/zookeeper/data/myid
echo 3 > /opt/zookeeper/data/myid//配置 Zookeeper 启动脚本
vim /etc/init.d/zookeeper#!/bin/bash
#chkconfig:2345 20 90
#description:Zookeeper Service Control Script
ZK_HOME='/opt/zookeeper'
case $1 in
start)echo "---------- zookeeper 启动 ------------"$ZK_HOME/bin/zkServer.sh start
;;
stop)echo "---------- zookeeper 停止 ------------"$ZK_HOME/bin/zkServer.sh stop
;;
restart)echo "---------- zookeeper 重启 ------------"$ZK_HOME/bin/zkServer.sh restart
;;
status)echo "---------- zookeeper 状态 ------------"$ZK_HOME/bin/zkServer.sh status
;;
*)echo "Usage: $0 {start|stop|restart|status}"
esac// 设置开机自启
chmod +x /etc/init.d/zookeeper
chkconfig --add zookeeper//分别启动 Zookeeper
service zookeeper start//查看当前状态
service zookeeper status
安装kafka
tar -xf kafka 2.13-2.7.0.tgz
vim /etc/profileexport KAFKA_HOME=/opt/kafka
export PATH=$PATH:$KAFKA_HOME/binsource /etc/profile
cd /opt/kafka/config
vim server.properties60
log.dirs=/var/log/kafka
#日志存放路径
zookeeper.connect=192.168.10.100:2181,192.168.10.101:2181,192.168.10.102:2181broker.id=1、2、3
42
num.network.threads=3
#处理网络请求的线程数量,默认即可
46
num.io.threads=8
#处理磁盘的IO线程数量,一定要比硬盘数大。
50
socket.send.buffer.bytes=102400
#发送套接字的缓冲去大小
socket.receive.buffer.bytes=102400
#接受者的接受套接字缓冲区大小。
socket.request,max.bytes=104857600
#请求套接字的缓冲区大小。(字节)65
log.dirs=/var/log/kafka
#日志存放路径
70
num.partitions=1
#在此kafka服务器上创建topic,默认分区数。如果指定了,这个配置无效了
75
num.recoverythreads.per.data.dir=1
#用来恢复,回收,清理data下的数据的线程数量。kafka默认不允许删除主题。110
log.retention.hours=168
#生产者发布的数据文件在主题当中保存的时间。168小时=7天
130
zookeeper.connect=192.168.10.100:2181,192.168.10.101:2181,192.168.10.102:2181vim /etc/init.d/kafka
chmod +x /etc/init.d/kafka
chkconfig --add kafka
service kafka start #启动Kafka
netstat -antp | grep 9092 #Kafka的默认端口号9092,查看服务端口是否启动创建主题
1、在kafka的bin日录下,是所有的kafka可执行命令文件
2、--zookeeper 指定的是zookeeper的地址和端口,保存kafka的元数据
3、--replication-factor 2 定义每个分区的副本数
4、partitions 3 指定主融的分区数
5、--topic test1 指定主题的名称kafka-topics.sh --create --zookeeper IP地址:2181 --replication-factor 副本数 --partitions 分区数 --topic 主题名
--replication-factor 副本数
--partitions 分区数
--topic 主题名 例:
kafka-topics.sh --create --zookeeper 192.168.10.100:2181,192.168.10.101:2181,192.168.10.102:2181 --replication-factor 2 --partitions 3 --topic test1查看主题查看所有主题
查看所有的topic
kafka-topics.sh --describe --zookeeper IP地址:2181
例
kafka-topics.sh --describe --zookeeper 192.168.10.100:2181,192.168.10.101:2181,192.168.10.102:2181查看指定主题
查看指定主题
kafka-topics.sh --describe --zookeeper IP地址:2181 --topic 主题名
例
kafka-topics.sh --describe --zookeeper 192.168.10.100:2181,192.168.10.101:2181,192.168.10.102:2181 --topic test2声明内容
Partition:分区编号 Leader:每个分区都有一个领导者(Leader),领导者负责处理分区的读写操作。
在上述输出中,领导者的编号分别为 3、1、3。Replicas:每个分区可以有多个副本(Replicas),用于提供冗余和容错性。
在上述输出中,Replica 3、1、2 分别对应不同的 Kafka broker。Isr:ISR(In-Sync Replicas)表示当前与领导者保持同步的副本。
ISR 3、1分别表示与领导者同步的副本。vim /etc/hosts192.168.10.100 test1
192.168.10.101 test2
192.168.10.102 test3生产者kafka-console-producer.sh --broker-list IP地址:9092 --topic 主题名
例
kafka-console-producer.sh --broker-list 192.168.10.100:9092,192.168.10.101:9092,192.168.10.102:9092 --topic test1消费者同步所有的
kafka-console-consumer.sh --bootstrap-server IP地址:9092 --topic 主题名 --from-beginning
例:
kafka-console-consumer.sh --bootstrap-server 192.168.10.100:9092,192.168.10.101:9092,192.168.10.102:9092 --topic test1 --from-beginning实时同步
kafka-console-consumer.sh --bootstrap-server IP地址:9092 --topic 主题名
例:
kafka-console-consumer.sh --bootstrap-server 192.168.10.100:9092,192.168.10.101:9092,192.168.10.102:9092 --topic test1修改分区数
格式
kafka-topics.sh --zookeeper ip地址:2181 --alter --topic 主题名 --partitions 分区数
例
kafka-topics.sh --zookeeper 192.168.10.100:2181,192.168.10.101:2181,192.168.10.102:2181 --topic test1 --partitions 3删除主题
格式
kafka-topics.sh --delete --zookeeper ip地址:2181 --topic 主题名
例:
kafka-topics.sh --delete --zookeeper 192.168.10.100:2181,192.168.10.101:2181,192.168.10.102:2181 --topic test1