沂水县住房和建设局网站,经营网站 备案信息,设置网站的关键词,乐山做网站的公司注#xff1a;本章使用的Kafka为2.7.0版本
Zookeeper概述
1.Zookeeper定义
Zookeeper是一个开源的分布式的#xff0c;为分布式框架提供协调服务的Apache项目。
2.Zookeeper工作机制
Zookeeper从设计模式角度来理解#xff1a;是一个基于观察者模式设计的分布式服务管理…注本章使用的Kafka为2.7.0版本
Zookeeper概述
1.Zookeeper定义
Zookeeper是一个开源的分布式的为分布式框架提供协调服务的Apache项目。
2.Zookeeper工作机制
Zookeeper从设计模式角度来理解是一个基于观察者模式设计的分布式服务管理框架它负责存储和管理大家都关心的数据然后接受观察者的注册一旦这些数据的状态发生变化Zookeeper就将负责通知已经在Zookeeper上注册的那些观察者做出相应的反应。
也就是说 Zookeeper 文件系统 通知机制。
3.Zookeeper特点
1Zookeeper一个领导者(Leader)多个跟随者(Follower)组成的集群。2Zookeepe集群中只要有半数以上节点存活Zookeeper集群就能正常服务。所以Zookeeper适合安装奇数台服务器。3全局数据一致每个Server保存一份相同的数据副本Client无论连接到哪个Server数据都是一致的。4更新请求顺序执行来自同一个Client的更新请求按其发送顺序依次执行即先进先出。5数据更新原子性一次数据更新要么成功要么失败。6实时性在一定时间范围内Client能读到最新数据。
4.Zookeeper数据结构
ZooKeeper数据模型的结构与Linux文件系统很类似整体上可以看作是一棵树每个节点称做一个ZNode。每一个ZNode默认能够存储1MB的数据每个ZNode都可以通过其路径唯一标识。
5.Zookeeper应用场景
提供的服务包括统一命名服务、统一配置管理、统一集群管理、服务器节点动态上下线、软负载均衡等。
●统一命名服务 在分布式环境下经常需要对应用/服务进行统一命名便于识别。例如IP不容易记住而域名容易记住。
●统一配置管理 1分布式环境下配置文件同步非常常见。一般要求一个集群中所有节点的配置信息是一致的比如Kafka集群。对配置文件修改后希望能够快速同步到各个节点上。
2配置管理可交由ZooKeeper实现。可将配置信息写入ZooKeeper上的一个Znode。 各个客户端服务器监听这个Znode。一旦 Znode中的数据被修改ZooKeeper将通知各个客户端服务器。
●统一集群管理 1分布式环境中实时掌握每个节点的状态是必要的。可根据节点实时状态做出一些调整。
2ZooKeeper可以实现实时监控节点状态变化。可将节点信息写入ZooKeeper上的一个ZNode。 监听这个ZNode可获取它的实时状态变化。
●服务器动态上下线 客户端能实时洞察到服务器上下线的变化。
●软负载均衡 在Zookeeper中记录每台服务器的访问数让访问数最少的服务器去处理最新的客户端请求。
5.Zookeeper选举机制
●第一次启动选举机制 1服务器1启动发起一次选举。服务器1投自己一票。此时服务器1票数一票不够半数以上3票选举无法完成服务器1状态保持为LOOKING
2服务器2启动再发起一次选举。 服务器1和2分别投自己一票并交换选票信息此时服务器1发现服务器2的myid比自己目前投票推举的服务器1大更改选票为推举服务器2。此时服务器1票数0票服务器2票数2票没有半数以上结果选举无法完成服务器12状态保持LOOKING
3服务器3启动发起一次选举。此时服务器1和2都会更改选票为服务器3。 此次投票结果服务器1为0票服务器2为0票服务器3为3票。此时服务器3的票数已经超过半数服务器3当选Leader。服务器12更改状态为FOLLOWING服务器3更改状态为LEADING
4服务器4启动发起一次选举。此时服务器123已经不是LOOKING状态不会更改选票信息。交换选票信息结果服务器3为3票服务器4为1票。此时服务器4服从多数更改选票信息为服务器3并更改状态为FOLLOWING
5服务器5启动同4一样当小弟。
●非第一次启动选举机制1当ZooKeeper 集群中的一台服务器出现以下两种情况之一时就会开始进入Leader选举 1服务器初始化启动。
2服务器运行期间无法和Leader保持连接。
2而当一台机器进入Leader选举流程时当前集群也可能会处于以下两种状态 1集群中本来就已经存在一个Leader。 对于已经存在Leader的情况机器试图去选举Leader时会被告知当前服务器的Leader信息 对于该机器来说仅仅需要和 Leader机器建立连接并进行状态同步即可。
2集群中确实不存在Leader。 假设ZooKeeper由5台服务器组成SID分别为1、2、3、4、5ZXID分别为8、8、8、7、7 并且此时SID为3的服务器是Leader。某一时刻3和5服务器出现故障因此开始进行Leader选举。
选举Leader规则
1.EPOCH大的直接胜出2.EPOCH相同事务id大的胜出3.事务id相同服务器id大的胜出
注
SID服务器ID。用来唯一标识一台ZooKeeper集群中的机器每台机器不能重复和myid一致。
ZXID事务ID。ZXID是一个事务ID用来标识一次服务器状态的变更。 在某一时刻集群中的每台机器的ZXID值不一定完全一致这和ZooKeeper服务器对于客户端“更新请求”的处理逻辑速度有关。
Epoch每个Leader任期的代号。没有Leader时同一轮投票过程中的逻辑时钟值是相同的。 每投完一次票这个数据就会增加
部署Zookeeper集群
1.实验组件
#准备3台服务器做Zookeeper集群
20.0.0.10
20.0.0.20
20.0.0.30
2.安装前准备
#关闭防火墙
systemctl stop firewalld
systemctl disable firewalld
setenforce 0#安装JDK
yum install -y java-1.8.0-openjdk java-1.8.0-openjdk-devel
java -version#将apache-zookeeper-3.5.7-bin.tar.gz压缩包上传至/opt目录
3.安装Zookeeper
#三台服务器一齐操作
cd /opt
tar -zxvf apache-zookeeper-3.5.7-bin.tar.gz
mv apache-zookeeper-3.5.7-bin /opt/zookeeper#修改配置文件
cd /opt/zookeeper/conf/
cp zoo_sample.cfg zoo.cfg
vim zoo.cfg
--2--
tickTime2000
#通信心跳时间Zookeeper服务器与客户端心跳时间单位毫秒--5--
initLimit10
#Leader和Follower初始连接时能容忍的最多心跳数tickTime的数量这里表示为10*2s--8--
syncLimit5
#Leader和Follower之间同步通信的超时时间这里表示如果超过5*2sLeader认为Follwer死掉并从服务器列表中删除Follwer--12--修改
dataDir/opt/zookeeper/data
#指定保存Zookeeper中的数据的目录目录需要单独创建--添加--
dataLogDir/opt/zookeeper/logs
#指定存放日志的目录目录需要单独创建--15--
clientPort2181
#客户端连接端口--末尾添加集群信息--
server.120.0.0.10:3188:3288
server.220.0.0.20:3188:3288
server.320.0.0.30:3188:3288#在每个节点上创建数据目录和日志目录
mkdir /opt/zookeeper/data
mkdir /opt/zookeeper/logs#在每个节点的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概述
1.为什么需要消息队列
主要原因是由于在高并发环境下同步请求来不及处理请求往往会发生阻塞。比如大量的请求并发访问数据库导致行锁表锁最后请求线程会堆积过多从而触发 too many connection 错误引发雪崩效应。
我们使用消息队列通过异步处理请求从而缓解系统的压力。消息队列常应用于异步处理流量削峰应用解耦消息通讯等场景。
当前比较常见的 MQ 中间件有 ActiveMQ、RabbitMQ、RocketMQ、Kafka 等。
2.使用消息队列的好处
1解耦 允许你独立的扩展或修改两边的处理过程只要确保它们遵守同样的接口约束。
在软件系统中解耦的概念与上述比喻类似。高度耦合的系统中各个组件之间的关系非常紧密修改其中一个组件可能会影响其他组件。而低耦合的系统中各个组件之间的关系更加独立修改一个组件不会波及其他组件。
解耦使得系统更容易维护、扩展和修改因为它减少了组件之间的依赖性。
2可恢复性 系统的一部分组件失效时不会影响到整个系统。消息队列降低了进程间的耦合度所以即使一个处理消息的进程挂掉加入队列中的消息仍然可以在系统恢复后被处理。
3缓冲 有助于控制和优化数据流经过系统的速度解决生产消息和消费消息的处理速度不一致的情况。
4灵活性 峰值处理能力 在访问量剧增的情况下应用仍然需要继续发挥作用但是这样的突发流量并不常见。 如果为以能处理这类峰值访问为标准来投入资源随时待命无疑是巨大的浪费。 使用消息队列能够使关键组件顶住突发的访问压力而不会因为突发的超负荷的请求而完全崩溃。
5异步通信 很多时候用户不想也不需要立即处理消息。消息队列提供了异步处理机制允许用户把一个消息放入队列但并不立即处理它。想向队列中放入多少消息就放多少然后在需要的时候再去处理它们。
3.消息队列的两种模式
1点对点模式一对一消费者主动拉取数据消息收到后消息清除 消息生产者生产消息发送到消息队列中然后消息消费者从消息队列中取出并且消费消息。消息被消费以后消息队列中不再有存储所以消息消费者不可能消费到已经被消费的消息。消息队列支持存在多个消费者但是对一个消息而言只会有一个消费者可以消费。
2发布/订阅模式一对多又叫观察者模式消费者消费数据之后不会清除消息 消息生产者发布将消息发布到 topic 中同时有多个消息消费者订阅消费该消息。和点对点方式不同发布到 topic 的消息会被所有订阅者消费。 发布/订阅模式是定义对象间一种一对多的依赖关系使得每当一个对象目标对象的状态发生改变则所有依赖于它的对象观察者对象都会得到通知并自动更新。
RabbitMQ:RabbitMQ支持点对点模型其中消息生产者将消息发送到队列而消息消费者从队列中接收和处理消息。
Apache Kafka: Kafka的消息发布/订阅模型也可以被视为一种点对点通信模型特别是在具有多个消费者时。 每个消费者组中的消费者将收到相同的消息实现了一对多的发布/订阅和一对一的点对点模型。
4.Kafka定义
Kafka是一个分布式的基于发布/订阅模式的消息队列MQMessage Queue主要应用于大数据实时处理领域。
5.Kafka特性
●高吞吐量、低延迟 Kafka 每秒可以处理几十万条消息它的延迟最低只有几毫秒。 每个 topic 可以分多个 PartitionConsumer Group 对 Partition 进行消费操作提高负载均衡能力和消费能力。
●可扩展性 kafka 集群支持热扩展
●持久性、可靠性 消息被持久化到本地磁盘并且支持数据备份防止数据丢失
●容错性 允许集群中节点失败多副本情况下若副本数量为 n则允许 n-1 个节点失败
●高并发 支持数千个客户端同时读写
6.Kafka系统架构
经纪人BrokerKafka集群由多个经纪人组成每个经纪人是一台运行Kafka服务的服务器。 经纪人存储主题分区中的数据处理生产者和消费者的请求以及维护元数据。
主题TopicKafka数据流的基本单元是主题它类似于一个数据流管道。生产者将数据发布到主题而消费者从主题中订阅数据。主题可以被分区每个分区可以看作是主题的子数据流具有自己的偏移量Offset。
分区Partition每个主题可以分成多个分区每个分区是数据的有序子集。分区允许Kafka进行水平扩展以处理大量数据。消息在分区中按照偏移量有序存储消费者可以独立读取每个分区的数据。
副本Replica为保证集群中的某个节点发生故障时该节点上的 partition 数据不丢失且 kafka 仍然能够继续工作kafka 提供了副本机制一个 topic 的每个分区都有若干个副本一个 leader 和若干个 follower。
Leader每个 partition 有多个副本其中有且仅有一个作为 LeaderLeader 是当前负责数据的读写的 partition。
FollowerFollower 跟随 Leader所有写请求都通过 Leader 路由数据变更会广播给所有 FollowerFollower 与 Leader 保持数据同步。Follower 只负责备份不负责数据的读写。如果 Leader 故障则从 Follower 中选举出一个新的 Leader。当 Follower 挂掉、卡住或者同步太慢Leader 会把这个 Follower 从 ISRLeader 维护的一个和 Leader 保持同步的 Follower 集合 列表中删除重新创建一个 Follower。
生产者Producer生产者是将数据发布到Kafka主题的组件。它负责将消息写入主题生产者可以将消息发送到指定的主题和分区。
消费者Consumer消费者是从Kafka主题中读取数据的组件。它可以订阅一个或多个主题 然后从主题的分区中读取消息。每个消费者都有一个唯一的消费者组IDKafka通过消费者组来实现负载均衡和容错性。
消费者组Consumer Group由多个 consumer 组成所有的消费者都属于某个消费者组即消费者组是逻辑上的一个订阅者。可为每个消费者指定组名若不指定组名则属于默认的组。将多个消费者集中到一起去处理某一个 Topic 的数据可以更快的提高数据的消费能力。消费者组内每个消费者负责消费不同分区的数据一个分区只能由一个组内消费者消费防止数据被重复读取。消费者组之间互不影响。
偏移量Offset偏移量是每个消息在分区中的唯一标识消费者使用偏移量来跟踪已读取的消息位置。偏移量存储在Kafka中消费者可以通过提交偏移量来记录已处理的消息。
ZookeeperKafka 通过 Zookeeper 来存储集群的 meta 信息。由于 consumer 在消费过程中可能会出现断电宕机等故障consumer 恢复后需要从故障前的位置的继续消费所以 consumer 需要实时记录自己消费到了哪个 offset以便故障恢复后继续消费。zookeeper的作用就是生产者push数据到kafka集群就必须要找到kafka集群的节点在哪里这些都是通过zookeeper去寻找的。消费者消费哪一条数据也需要zookeeper的支持从zookeeper获得offsetoffset记录上一次消费的数据消费到哪里这样就可以接着下一条数据进行消费。
7.Kafka工作流程
生产者将消息发送到指定的主题每个消息都附带一个可选的键Key和值Value。
主题可以有多个分区生产者将消息写入一个分区。
经纪人Broker接收到消息后将其存储在分区中并分配一个唯一的偏移量。
在 Kafka 中偏移量Offset是一个用于标识消费者在一个特定分区中已经读取到的消息位置的标记。 每个分区的每个消费者都会有一个偏移量它表示下一条将要读取的消息的位置。 偏移量是一个64位的整数表示在分区中消息的唯一位置。
Kafka 的消息是持久化的一旦消息被写入分区它就会保留一段时间供消费者在需要时读取。 通过使用偏移量消费者可以准确地指定它想要从哪个位置开始消费消息。
偏移量的重要性体现在以下几个方面 消息的唯一标识 每个消息都有一个在分区内的唯一偏移量它可以被用作消息的全局唯一标识。
精确的消费位置 消费者可以通过指定偏移量精确地确定从哪个位置开始消费消息。 这是在重新启动消费者或者切换到不同的消费者时非常有用的功能。
消费者组的偏移量管理 在消费者组中Kafka 会追踪每个消费者的偏移量。 当新的消费者加入或离开消费者组时Kafka 负责协调和管理这些消费者的偏移量。
至少一次语义 消费者可以选择手动提交偏移量确保消息至少被消费一次。 这是通过将偏移量与成功处理消息的应用程序逻辑关联来实现的。
在 Kafka 中有两种类型的偏移量
消费者组偏移量Group Offsets 由 Kafka 内部管理用于追踪消费者组中每个消费者的偏移量。
自定义偏移量Commit Offsets 由应用程序管理消费者可以选择手动提交偏移量以实现更细粒度的偏移量控制。
消费者订阅一个或多个主题并从指定分区中读取消息。消费者使用偏移量来跟踪已读取的消息。
消费者可以以不同的速率读取消息Kafka支持多个消费者组每个组可以独立消费消息从而实现负载均衡和水平扩展。
Kafka保留消息一定的时间这个时间可以配置一般情况下已读取的消息会被保留一段时间以便允许消费者重新读取。
Kafka的分布式特性、数据持久性、高吞吐量和水平扩展性使其成为处理大规模数据流的强大工具可用于多种用途 包括日志收集、实时数据分析、事件驱动架构等。
部署Kafka集群
1.安装Kafka
cd /opt
--上传kafka_2.13-2.7.0.tgz包--
tar zxvf kafka_2.13-2.7.0.tgz
mv kafka_2.13-2.7.0 kafka/#修改配置文件
cd /opt/kafka/config/
cp server.properties server.properties.bak
vim server.properties
--21--
broker.id0
#broker的全局唯一编号每个broker不能重复因此要在其他机器上配置 broker.id1、broker.id2--31--
listenersPLAINTEXT://20.0.0.10:9092
#指定监听的IP和端口如果修改每个broker的IP需区分开来也可保持默认配置不用修改--42--
num.network.threads3
#broker 处理网络请求的线程数量一般情况下不需要去修改--45--
num.io.threads8
#用来处理磁盘IO的线程数量数值应该大于硬盘数--48--
socket.send.buffer.bytes102400
#发送套接字的缓冲区大小--51--
socket.receive.buffer.bytes102400
#接收套接字的缓冲区大小--54--
socket.request.max.bytes104857600
#请求套接字的缓冲区大小--60--
log.dirs/var/log/kafka
#kafka运行日志存放的路径也是数据存放的路径--65--
num.partitions1
topic在当前broker上的默认分区个数会被topic创建时的指定参数覆盖--69--
num.recovery.threads.per.data.dir1
#用来恢复和清理data下数据的线程数量--103--
log.retention.hours168
#segment文件数据文件保留的最长时间单位为小时默认为7天超时将被删除--110--
log.segment.bytes1073741824
#一个segment文件最大的大小默认为 1G超出将新建一个新的segment文件
#Kafka 以日志文件的形式维护其数据而这些日志文件被分割成多个日志段。当一个日志段达到指定的大小时就会创建一个新的日志段。--123--
zookeeper.connect20.0.0.10:2181,20.0.0.20:2181,20.0.0.30:2181
#配置连接Zookeeper集群地址#修改全局配置
vim /etc/profile
--添加--
export KAFKA_HOME/opt/kafka
export PATH$PATH:$KAFKA_HOME/binsource /etc/profile
#配置Zookeeper启动脚本
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 ];thenecho kafka is not runningelseecho kafka is runningfi
;;
*)echo Usage: $0 {start|stop|restart|status}
esac#设置开机自启
chmod x /etc/init.d/kafka
chkconfig --add kafka#分别启动Kafka
service kafka start
2.Kafka命令行操作
#创建topic
kafka-topics.sh --create --zookeeper 20.0.0.10:2181,20.0.0.20:2181,20.0.0.30:2181 --replication-factor 2 --partitions 3 --topic test1----------------------------------------------------------------------------------------
--bootstrap-server定义 bootstrap-server 集群服务器地址如果有多个 IP 地址使用逗号分割一般使用一个IP即可
--replication-factor定义分区副本数1 代表单副本建议为 2
--partitions定义分区数
--topic定义 topic 名称
----------------------------------------------------------------------------------------#查看当前服务器钟的所有topic
kafka-topics.sh --list --zookeeper 20.0.0.10:2181,20.0.0.20:2181,20.0.0.30:2181#查看某个topic的详情
[rootapache config]# kafka-topics.sh --describe --zookeeper 20.0.0.10:2181,20.0.0.20:2181,20.0.0.30:2181
Topic: test1 PartitionCount: 3 ReplicationFactor: 2 Configs: Topic: test1 Partition: 0 Leader: 0 Replicas: 0,2 Isr: 0Topic: test1 Partition: 1 Leader: 1 Replicas: 1,0 Isr: 1,0Topic: test1 Partition: 2 Leader: 2 Replicas: 2,1 Isr: 2,1----------------------------------------------------------------------------------------
Partition分区编号 Leader每个分区都有一个领导者Leader领导者负责处理分区的读写操作。Replicas每个分区可以有多个副本Replicas用于提供冗余和容错性。IsrISRIn-Sync Replicas表示当前与领导者保持同步的副本。
----------------------------------------------------------------------------------------#发布消息
kafka-console-producer.sh --broker-list 20.0.0.10:9092,20.0.0.20:9092,20.0.0.30:9092 --topic test1#消费消息
kafka-console-consumer.sh --bootstrap-server 20.0.0.10:9092,20.0.0.20:9092,20.0.0.30:9092 --topic test1 --from-beginning
#--from-beginning会把主题中以往所有的数据都读取出来#修改分区数
kafka-topics.sh --zookeeper 20.0.0.10:2181,20.0.0.20:2181,20.0.0.30:2181 --alter --topic test1 --partitions 6#删除 topic
kafka-topics.sh --delete --zookeeper 20.0.0.10:2181,20.0.0.20:2181,20.0.0.30:2181 --topic test1
