信誉好的丹徒网站建设,网站demo制作,响应式网站是做多大尺寸,给文字做网站链接kafka3.0之前是依赖于zookeeper的。
zookeeper是开源#xff0c;分布式的架构。提供协调服务#xff08;Apache项目#xff09;
基于观察者模式涉及的分布式服务管理架构。
存储和管理数据。分布式节点上的服务接受观察者的注册。一旦分布式节点上的数据发生变化#xf…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分别投自己一票交换选举信息myidA发现B的myid比A大A的这一票会转而投给B
A 0 B 2 没有半数以上结果A B会进入looking B有可能成为leader
C启动 MYID C的myid最大A和B都会把票投给C加上C自己的一票
A 0 B 0 C 3
C的状态变为leaderA和B会变为follower
只要leader确定后续的服务器都是追随者。
只有两种情况会开启选举机制
1.初始化的情况会产生选举
2.服务器之间和leader丢失了连接状态
leader已存在建立连接即可
leader不存在服务器id大的胜出EPOCH大直接胜出。EPOCH相同事务id大的胜出 EPOCH每个leader任期的一个代号没有leader大家的逻辑地位相同每投完一
次数据数据都是递增的。
事务id表示
服务器的每一次变更
服务器id用来表示zookeeper集群当中都有个id每台机器不重复和myid保持一致 实验部署
zookeeoerkafka(2.7.0)
192.168.233.10 ---zookeeperkafak 20.0.0.10
192.168.233.20 ---zookeeperkafak
192.168.233.30 ---zookeeperkafak2/4G 关闭防火墙
所有服务器
拖入kafak,zookeeper所有服务器
升级java环境
yum install -y java-1.8.0-openjdk java-1.8.0-openjdk-develjava -version line 2
tickTime2000
服务器与客户端的心跳时间ms
line 6
initLimit10
领导者和追随者之间初始连接时能够容忍的超时时间最多几次心跳超时line 10
syncLimit5
同步超时时间领导者和追随者之间同步通信超时的时间超过5*2就会认为follower丢失移出集群line15
dataDir/tmp/zookeeper
保存数据的目录需要单独创建line18
clientPort2181line15下添加
dataLogDir/opt/zookeeper/logs最后一行添加整个集群信息
server.1192.168.233.10:3188:3288
server.2192.168.233.20:3188:3288
server.3192.168.233.30:3188:3288 server.1192.168.233.10:3188:3288 server.2192.168.233.20:3188:3288 server.3192.168.233.30:3188:3288
1每个zookeeper集群的初始myid。
192.168.233.10服务器的IP地址
3188领导者和追随者之间交换信息的端口内部通信端口
3288一旦leader丢失响应开启选举3288就是用来进行选举时的服务器通信端口。 根据zoo.cfg里面的目录路径创建路径 输入myid test123
编辑脚本
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}
esacwqchmod x /etc/init.d/zookeeper每台服务器都添加
chkconfig --add zookeepertest123
service zookeeper startservice zookeeper status
查看谁是leader 消息队列KAFKA
为什么要引入消息队列MQ他是一个中间件。负责发消息。在高并发环境下同步请求来不及处理。来不及处理的请求会形成阻塞。
比如数据库会进行行锁或者表锁。请求线程满了超标了会报错too many connection。导致整个系统雪崩 消息队列的作用 1.异步处理请求核心
允许用户把一个消息放入队列但是不立即处理等用户想处理的时候再处理 2.流量消减应用解耦
耦合在软件系统当中修改一个组件需要修改其他组件高度耦合
低度耦合修改其中一个组件对其他组件影响不大无须修改所有
解耦只要通信保证其他的不影响整个集群可以独立的扩展修改降低组件之间的依赖性。
依赖点就是接口约束通过不同的端口保证集群通信。 3.可恢复性
系统当中的有一部分组件消失不影响整个系统。也就是说在消息队列当中即使有一个处理消息的进程失败一旦恢复还可以重写加入到队列当中继续处理消息。 4.缓冲
可以控制和优化数据经过系统的时间和速度。解决生产消息和消费消息处理速度不一致的问题。 5.峰值的处理能力
消息队列在峰值的情况下能够顶住突发的访问压力。避免专门为了突发情况而对系统进行修改 消息队列的模式
点对点一对一消息的生产者发送消息到队列中消费者从队列中提取消息消费者提取完之后队列中被提取的消息将会被移除。后续消费者不能再消费队列当中的消息。消息队列可以有多个消费者但是一个消息只能由一个消费者提取。
RABBITMQ 发布订阅模式一对多观察者模式。消费者提取数据之后队列当中的数据不会被清除。
生产者发布一个消息到主题。所有消费者都是通过主题获取消息。
主题topic **
topic类似一个数据流管道生产者把消息发布到主题。消费者从主题当中订阅数据。主题可以分区每个分区都有自己的偏移量。 分区---partition**
每个主题都可以分成多个分区。每个分区是数据的有序子集分区可以允许kafka进行水平扩展以处理大量数据。
消息在分钟按照偏移量存储消费者可以独立读取每个分区的数据 偏移量**
每个消息在分区当中唯一的标识。消费者可以通过偏移量来跟踪已读或者未读消息的位置。也可以提交这个偏移量来记录已处理的信息。 消费三种方式 1.begin从头开始获取所有
2.实时获取我只消费后续产生的消息
3.指定偏移量消费代码实现 生产者
生产者把数据发送kafka的主题当中负责写入消息。
消费者
从主题当中读取数据消费者可以是一个也可以是多个。每个消费者有一个唯一的消费者IDKafka通过消费者实现负载均衡和容错性。 经纪人---Broker
每个Kafka节点都有一个borker每个负责一台Kafka服务器id唯一存储主题分区当中数据处理生产和消费者的请求。
维护元数据zookeeper zookeeperzookeeper负责保存元数据元数据就是topic的相关信息发在哪台主机上指定了多少分区以及副本数偏移量
zookeeper自建一个主题_consumer_offsets
3.0之后不依赖zookeeper的核心。元数据由Kafka节点自己管理。 Kafka的工作流程 生产者写入topic的数据是持久化的默认7小时
至少一次语义只要消费者进入确保消息至少被消费一次 实验部署
test1,2,3
cd /opt
tar -xf kafka
mv kafka_2,13.. kafka设置环境变量
vim /etc/profile
最后插入
export KAFKA_HOME/opt/kafka
export PATH$PATH:$KAFKA_HOME/bin
source /etc/profilecd kafka
cd config/
lsserver.properties是主配置文件 server.properties是主配置文件
先备份 test1 2 3 每台主机配置有些许不同
vim server.propertiesline 21 60 123line 21
broker.id1
//同时也是leader的id三台主机不重复即可line 28
如果用broker进行修改该行可以不改line 42
num.network.thread3
处理网络的线程数量默认即可line 46
num.io.thread8
处理磁盘的io线程数量这个值一定要比硬盘数大line 50
socket.send.buffer.bytes102400
发送套接字的缓存大小line 54
socket.receive.buffer.bytes102400
接受者的接受套接字缓冲区大小line58
socket.request.max.bytes104857600
请求套接字的缓冲区大小line 65
log.dirs/var/log/kafkaline 70
num.partitions1
在此Kafka服务器上创建topic默认分区数。如果指定了这个配置无效了。line75
num.recovery.thread.per.data.dir1
用来恢复回收清理data下的数据的线程数量。Kafka默认不允许删除主题。line110103
log.retention.hours168
生产者发布的数据文件在主题当中保存的时间单位-小时line130123
zookeeper.connect192.168.233.10:2181,192.168.233.20:2181,192.168.233.30:2181
//指定zookeeper集群wq test1,2,3
设置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}
esacwqchmod x /etc/init.d/kafka
chkconfig --add kafka
service kafka startnetstat -antp | grep 9092 test2--创建主题
kafka所有命令都在bin下所以执行命令都得去bin目录下执行
cd /opt/kafka/bin/kafka-topics.sh --create --zookeeper 20.0.0.10:2181,20.0.0.20:2181,20.0.0.60:2181 --replication-factor 2 --partitions 3 --topic test1--replication-factor 2创建的副本数副本的作用就是冗余 创建主题成功
1.在Kafka的bin目录下是所有kafka可执行命令文件
2.--zookeeper指定的是zookeeper的地址和端口保存kafka的元数据
3.--replication-factor 2 定义每个分区的副本数
4.partitions 3 指定主题的分区数 5.--topic test1 指定主题的名称 Partition分区编号
Leader每个分区都有一个领导者Leader领导者负责处理分区的读写操作。 在上述输出中领导者的编号分别为 3、1、3。
Replicas每个分区可以有多个副本Replicas用于提供冗余和容错性。 在上述输出中Replica 3、1、2 分别对应不同的 Kafka broker。
IsrISRIn-Sync Replicas表示当前与领导者保持同步的副本。 ISR 3、1分别表示与领导者同步的副本。
test1,2,3
做映射
20.0.0.10 test1
20.0.0.20 test2
20.0.0.60 test3
发布消息
bin目录下
test2
kafka-console-producer.sh --broker-list 20.0.0.10:9092,20.0.0.20:9092,20.0.0.60:9092 --topic test1 kafka-console-consumer.sh --bootstrap-server 20.0.0.10:9092,20.0.0.20:9092,20.0.0.60:9092 --topic test2 --from-beginning test2
创建主题必须要创建分区分区一定要给副本数
kafka-topics.sh --create --zookeeper 192.168.233.20:2181 --replication-factor 1 --partitions 1 --topic gq1test3
kafka-topics.sh --create --zookeeper 192.168.233.20:2181 --replication-factor 1 --partitions 1 --topic gq2test1
kafka-console-consumer.sh --bootstrap-server 192.168.233.20:9092 --topic gq1kafka-console-consumer.sh --bootstrap-server 192.168.233.20:9092 --topic gq2test2
发起
kafka-console-producer.sh --broker-list 192.168.233.20:9092 -topic gq1
修改分区数
kafka-topics.sh --zookeeper 192.168.233.20:2181 --alter --topic gq1 --partitions 3test2
kafka-topics.sh --describe --zookeeper 192.168.233.20:2181 --alter --topic gq1 kafka-topics.sh --delete --zookeeper 192.168.233.20:2181 --topic gq1
总结
1.zookeeper主要是分布式观察者模式。统一各个服务器节点的数据。
在Kafka当中他就是收集保存Kafka的元数据。
2.kafka消息队列订阅发布模式
RABBIT MQ 轻量级
KAFKA 速度快占用资源
