1.zookeeper
kafka3.0之前依赖于zookeeper
zookeeper是一个开源,分布式的架构,提供协调服务(Apache项目)
基于观察者模式涉及的分布式服务管理架构
存储和管理数据,分布式节点上的服务接受观察者的注册,一旦分布式节点上的数据如果发生变化,由zookeeper来负责通知分布式节点上的服务
zookeeper分为领导者,追随者 leader follower组成的集群
只要有一半以上的集群存活,zookeeper集群就可以正常工作,适用于安装奇数台的服务集群
全局数据一致,每个zookeeper每个节点都保存相同的数据,维护监控服务的数据一致。(主要作用)
数据更新的原子性,要么都成功,要么都失败
实时性,只要有变化,立刻同步。
zookeeper的应用场景
1.统一命名服务,在分布式的环境下,对所有的应用和服务进行统一命名
2.统一配置管理,配置文件同步,kafka的配置文件被修改,可以快速同步到其他节点
3.统一集群管理,实时掌握所有节点的状态
4.服务器动态上下限
5.负载均衡,把访问的服务器的数据,发送到访问最少的服务器处理客户端的请求
领导者和追随者:zookeeper的选举机制
三台服务器:A B C
A 先启动 发起第一选举,投票投给自己,只有一票,不满半数,A的状态是looking
B 启动 在发起一次选举,A和B分别投自己一票,交换选票信息,myid A发现B的myid比A的大,A的这票会转而投给B,A 0 B 2 没有半数以上结果,A B 会进入looking 、B有可能leader
C 启动 MYID C的myid最大 A和B都会把票投给C A 0 B 0 C 3
C的状态变为leader A B变为follower
只有leader确定,后续的服务器都是追随者
只有两种情况会开启选举机制:
1.初始化的情况会产生选举
2.服务器之间和eader丢失了连接状态
特殊情况下:
leader已经存在,建立连接即可
leader不存在,leader不存在
1.服务器ID大的胜出
2.EPOCH大,直接胜出
3.EPOCH相同,事务ID大的胜出
EPOCH每个leader任期的代号,没有leader,大家的逻辑地址相同。每投完一次之后,数据时递增
事务id表示服务器的每一次更新,每变更一次id变化一次‘
服务器ID
zookeeper 当中所有机器。每台机器不重复,和mysql保存一直
zookeeper+kafka(2.7.0)
kafka (3.4.1)
实现
zookeeper集群
192.168.233.10 zookeeper+kafka
192.168.233.20 zookeeper+kafka
192.168.233.30 zookeeper+kafka
所有
systemctl stop firewalld
setenforce 0
cd /opt
拖进去 apache-zookeeper-3.5.7-bin.tar.gz kafka_2.13-2.7.0
yum install -y java-1.8.0-openjdk java-1.8.0-openjdk-devel
java -version
cd /opt
tar -xf apache-zookeeper-3.5.7-bin.tar.gz
mv apache-zookeeper-3.5.7-bin zookeeper
cd zookeeper
cd /opt/zookeeper/conf
cp zoo_sample.cfg zoo.cfg
vim zoo.cfg
tickTime=2000
#服务器与客户端之间心跳时间,2秒检测一次服务器和客户端之间的通信
initLimit=10
#领导者和追随者之间,最对几次心跳数超时 10*2S 20S
syncLimit=5
#同步超时时间,领导者和追随者之间,同步通信超时的时间。5*2s leader会认为follower丢失,移除集群
16行
dataDir=/opt/zookeeper/data
dataLogDir=/opt/zookeeper/logs (需要改)
添加
server.1=192.168.233.10:3188:3288
server.2=192.168.233.20:3188:3288
server.3=192.168.233.30:3188:3288
##
server.1=192.168.233.10:3188:3288
1 定义每个zookeeper集群的初始myid
192.168.233.10 :服务器的ip地址
3188:领导者和追随者之间交换信息的端口(内部通信的端口)
3288:一旦leader丢失响应,开启选举,3288就是用来执行选举时的通信端口
##
wq!
mkdir /opt/zookeeper/data
mkdir /opt/zookeeper/logs
三台分别创建
10:
cd ..
cd data/
echo 1 > myid
20:
cd ..
cd data/
echo 2 > myid
30:
cd ..
cd data/
echo 3 > myid
所有
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
wq!
三台服务器分别启动(一个一个起)
chmod +x /etc/init.d/zookeeper
chkconfig --add zookeeper
service zookeeper start
起完以后看状态
service zookeeper status
2.kafka
为什么要引入消息队列(MQ),首先他也是一个中间件。在高并发环境下,同步请求太多来不及处理,来不及处理的请求会阻塞,比方数据库就会形成行锁或者表锁,请求线程满了,超标了,too many connection。整个系统雪崩
消息队列的作用:异步处理请求 流量削峰,应用解耦 可恢复性 缓冲机制
解耦:
耦合:在软件系统当中,修改一个组件需要修改所有其他组件,高度耦合
低度耦合:修改其中一个组件,对其他最贱影响不大,无需修改所有
A B C
只有通信保证,其他的修改不影响整个集群,每个组件可以独立的扩展,修改,降低组件之间的依赖性
依赖点就是接口约束,通过不同的端口,保证集群通信
可恢复性:系统当中的有一部分组件消失,不影响整个系统,也就是在消息队列当中,即使有一个处理消息的进程失败,一旦恢复还可以重新加入到队列当中,继续处理消息。
缓冲机制:可以控制和优化数据经过系统的时间和速度,解决生产消息和消费消息处理速度不一致的问题
峰值的处理能力:消息队列在峰值情况之下,能够顶住突发的访问压力。避免专门为了突然情况而对系统进行修改
异步通信:允许用户把一个消息放入队列,但是不立即处理,等我想处理的时候在处理
消息队列的模式:
点对点 一对一:消息的生产者发送消息到队列中,消费者从队列中提取消息,消费者提取完之后,队列中被提取的消息将会被移除,后续消费者不能再消费队列当中的消息,消息队列可以有多个消费者,但是一个消息,只能由一个消费者提取
RABBITMQ
发布/订阅模式:一对多,又叫做观察者模式。消费者提取数据之后,队列当中的消息不会被清除
生产者发布一个消息到对象(主题)所有消费者都是通过主题获取消费之后,队列当中的消息不会消除
主题:topic topic类似于一个数据流的管道,生产者把消息发布到主题,消息从主题当中订阅数据,主题可以分区,每个分区都要自己的偏移量
分区:partition 每个主题都可以分成多个分区,每个分区是数据的有序子集,分区可以运行kafka进行水平扩展,以处理大量数据
消息在分钟按照偏移量存储,消费者可以独立每个分区的数据
偏移量:是每个消息在分区中唯一的标识,消费者可以通过偏移量来跟踪获取已读或者未读消息的位置,也可以提交偏移量来记录已处理的信息
生产者:producer 生产者把数据发送kafka的主题当中 负责写入消息
消费者:consumer从主题当中读取数据,消费者可以是一个也可以是多个,每个消费者有一个唯一的消费者组ID,kafka实现负载均衡和容错性
经纪人:Broker每个kafka节点都有一个Broker,每个Broker负责一台kafka服务器,id唯一,存储主题分区当中的数据,处理生产和消费者的请求,维护元数据(zookeeper)
zookeeper:zookeeper负责保存元数据,元数据就是topic的相关信息(发布在哪台主机上,指定了多少分区,以及副本数,偏移量)
zookeeper自建一个主题:_consumer_offsets,
3.0之后不依赖zookeeper的核心 元数据由kafka节点自己管理
消费的方式:
begining,从头开始
实时更新
指定位置,用代码编写
kafka的工作流程:
生产者向主题里面发送数据,主题里的分区保存数据,消费者根据消费方式来消费数据。
生产者写入的topic的数据时持久化,默认7个小时
至少一次语义:只要消费者进入,确保消息至少被消费一次。
实验
所有
cd /opt
tar -xf kafka_2.13-2.7.0.tgz
mv kafka_2.13-2.7.0 kafka
vim /etc/profile
export KAFKA_HOME=/opt/kafka
export PATH=$PATH:$KAFKA_HOME/bin
source /etc/profile
cd /opt/kafka/config
cp server.properties server.properties.bak
10:
vim server.properties
21行:
65行 修改
130行 修改
zookeeper.connect=192.168.233.10:2181,192.168.233.20:2181,192.168.233.30:2181
配置zookeeper的集群
20:
所有:
vim /etc/init.d/kafka
#!/bin/bash
#chkconfig:2345 22 88
#description:Kafka Service Control Script
KAFKA_HOME='/opt/kafka'
case $1 in
start)
echo "---------- Kafka 启动 ------------"
${KAFKA_HOME}/bin/kafka-server-start.sh -daemon ${KAFKA_HOME}/config/server.properties
;;
stop)
echo "---------- Kafka 停止 ------------"
${KAFKA_HOME}/bin/kafka-server-stop.sh
;;
restart)
$0 stop
$0 start
;;
status)
echo "---------- Kafka 状态 ------------"
count=$(ps -ef | grep kafka | egrep -cv "grep|$$")
if [ "$count" -eq 0 ];then
echo "kafka is not running"
else
echo "kafka is running"
fi
;;
*)
echo "Usage: $0 {start|stop|restart|status}"
esac
wq!
chmod +x /etc/init.d/kafka
chkconfig --add kafka
service kafka start
netstat -antp | grep 9092
所有:
cd /opt/kafka/bin
20:
kafka-topics.sh --create --zookeeper 192.168.233.10:2181,192.168.233.20:2181,192.168.233.30:2181 --replication-factor 2 --partitions 3 --topic test1
kafka-topics.sh --create --zookeeper 192.168.66.15:2181,192.168.66.16:2181,192.168.66.17:2181 --replication-factor 2 --partitions 3 --topic test1
##
创建主题
1.在kafka的bin目录下创建,是所有kafka可执行命令的文件
2. --zookeeper指定的是zookeeper的地址和端口,保存kafka的元数据
3. --replication-factor 2 定义每个分区的副本数
4.partition 3 指定主题的分区数
5. --topic test1 指定主题的名称
10:
kafka-topics.sh --describe --zookeeper 192.168.233.10:2181,192.168.233.20:2181,192.168.233.30:2181
kafka-topics.sh --describe --zookeeper 192.168.233.10:2181,192.168.233.20:2181,192.168.233.30:2181 --topic test1
kafka-topics.sh --describe --zookeeper 192.168.66.15:2181,192.168.66.16:2181,192.168.66.17:2181
kafka-topics.sh --describe --zookeeper 192.168.66.15:2181,192.168.66.16:2181,192.168.66.17:2181 --topic test3
所有
vim /etc/hosts
192.168.233.10 test1
192.168.233.20 test2
192.168.233.30 test3
记得改主机名
20:
kafka-console-producer.sh --broker-list 192.168.66.15:9092,192.168.66.16:9092,192.168.66.17:9092 --topic test3
kafka-console-producer.sh --broker-list 192.168.233.10:9092,192.168.233.20:9092,192.168.233.30:9092 --topic test1
30:
从头开始
kafka-console-consumer.sh --bootstrap-server 192.168.66.15:9092,192.168.66.16:9092,192.168.66.17:9092 --topic test3 --from-beginning
kafka-console-consumer.sh --bootstrap-server 192.168.233.10:9092,192.168.233.20:9092,192.168.233.30:9092 --topic test1 --from-beginning
实时消息
kafka-console-consumer.sh --bootstrap-server 192.168.66.15:9092,192.168.66.16:9092,192.168.66.17:9092 --topic test3
kafka-console-consumer.sh --bootstrap-server 192.168.233.10:9092,192.168.233.20:9092,192.168.233.30:9092 --topic test1
20:创建
kafka-topics.sh --create --zookeeper 192.168.233.20:2181 --partitions 1 --replication-factor 1 --topic guoqi1
kafka-topics.sh --create --zookeeper 192.168.66.16:2181 --partitions 1 --replication-factor 1 --topic guoqi1
30:创建
kafka-topics.sh --create --zookeeper 192.168.233.20:2181 --partitions 1 --replication-factor 1 --topic guoqi2
kafka-topics.sh --create --zookeeper 192.168.66.17:2181 --partitions 1 --replication-factor 1 --topic guoqi2
10:连接
kafka-console-consumer.sh --bootstrap-server 192.168.233.20:9092 --topic guoqi1
kafka-console-consumer.sh --bootstrap-server 192.168.233.20:9092 --topic guoqi2
kafka-console-consumer.sh --bootstrap-server 192.168.66.16:9092 --topic guoqi1
kafka-console-consumer.sh --bootstrap-server 192.168.66.17:9092 --topic guoqi2
20:发送
kafka-console-producer.sh --broker-list 192.168.233.20:9092 --topic guoqi1
kafka-console-producer.sh --broker-list 192.168.66.16:9092 --topic guoqi1
30:发送
kafka-console-producer.sh --broker-list 192.168.233.30:9092 --topic guoqi2
kafka-console-producer.sh --broker-list 192.168.66.17:9092 --topic guoqi2
如何修改分区数
20:kafka-topics.sh --zookeeper 192.168.233.20:2181 --alter --topic guoqi1 --partitons 3
删除
kafka-topics.sh --delete --zookeeper 192.168.233.20:2181 --topic guoqi1
查看元数据
cd /opt/zookeeper/bin
./zkCli.sh -server 192.168.233.10:2181