网站建设合作签约报道,Linux做视频网站网速均衡,app编写软件,大连开发区着火kafka3.0之前依赖于zookeeper
zookeeper开源#xff0c;分布式的架构#xff0c;提供协调服务#xff08;Apache项目#xff09;
基于观察者模式涉及的分布式服务管理架构
存储和管理数据#xff0c;分布式节点上的服务接受观察者的注册#xff0c;一旦分布式节点上的…
kafka3.0之前依赖于zookeeper
zookeeper开源分布式的架构提供协调服务Apache项目
基于观察者模式涉及的分布式服务管理架构
存储和管理数据分布式节点上的服务接受观察者的注册一旦分布式节点上的数据发生变化由zookeeper负责通知分布式节点上的服务
zookeeper特点
1、分为领导者 追随者 leader follow组成的集群
2、只要有一半以上的集群存活zookeeper集群就可以正常工作适用于安装奇数台的服务集群
3、全局数据一致每一个zookeeper每个节点都保存相同的数据维护监控服务的数据一致
4、数据更新的原子性要么都成功要么都失败
5、实时性只要有变化立刻同步特性 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 A 0 B 0 C 3
C的状态变为leader A和B变成follower
只要leader确定后续的服务器都是追随者 只有两种情况会开启选举机制
1、初始化的情况会产生选举
2、服务器之间和leader丢失了连接状态 leader已经存在建立连接即可
leader不存在leader不存在
1、服务器ID大的胜出
2、EPOCH大直接胜出
3、EPOCH相同事务ID大的胜出 EPOCH每个leader任期的代号如果没有leader所有的逻辑地位相同每投完一次之后数据是递增
事务ID表示服务器的每一次变更每变更一次事务ID变化一次
服务器IDzookeeper集群当中的机器都有一个ID每台机器不重复和myid保持一致 zookeeperkafka2.7.0集群部署
zookeeper集群
20.0.0.10 zookeeperkafka
20.0.0.11 zookeeperkafka
20.0.0.12 zookeeperkafka
所有服务器都是2核4G
#所有服务器#关闭防火墙和安全机制
#把zookeeper和kafka源码包拖到opt目录下#升级java环境
yum install -y java-1.8.0-openjdk java-1.8.0-openjdk-develjava -version#解压源码包
cd /opt
tar -xf apache-zookeeper-3.5.7-bin.tar.gz
mv apache-zookeeper-3.5.7-bin zookeeper
cd zookeeper/confcp zoo_sample.cfg zoo.cfg
vim zoo.cfgtickTime2000
#服务器与客户端之间心跳时间2秒检测一次服务器和客户端之间的通信
initLimit10
#领导者和追随者之间,初始连接时能够容忍的超时时间10*2s 20s
syncLimit5
#同步超时时间领导者和追随者之间同步通信超时的时间5*2sleader会认为follower丢失移除集群
dataDIr/opt/zookeeper/data
#保存数据的目录需要单独创建
dataLogDir/opt/zookeeper/logs
#保存日志的目录需要单独创建
clientPort2181
#端口号
server.120.0.0.10:3188:3288
server.220.0.0.11:3188:3288
server.320.0.0.12:3188:3288
#server.120.0.0.10:3188:3288
#1 每个zookeeper集群的初始myid
# 20.0.0.10 服务器的IP地址
# 3188 领导者和追随者之间交换信息的端口内部通信的端口
# 3288 一旦leader丢失响应开启选举3288就是用来选举时的服务之间通信端口cd /opt/zookeeper
mkdir data logs#给每台服务器分配myid
cd /data
echo 1 myidcd /data
echo 2 myidcd /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#给脚本权限
chmod x /etc/init.d/zookeeper
chkconfig --add zookeeper#重启服务
service zookeeper start
#查看状态(查看谁是主)
service zookeeper status
消息队列kafka
为什么要引入消息队列MQ他也是一个中间件在高并发环境下同步请求来不及处理来不及处理的请求会形成阻塞
比方说数据库就会形成行锁或者表锁请求线程满了超标了too many connection 整个系统雪崩
消息队列的作用
1、异步处理请求
异步处理的流程图 2、流量削峰应用解耦
解耦只要通信保证其他的修改不影响珍格格集群每个组件可以的独立的扩展修改降低组件之间的依赖性
依赖点就是接口约束通过不同的端口保证集群通信
耦合在软件系统中修改一个组件需要修改所有其他组件高度耦合
低度耦合修改其中一个对其他组件影响不大无需修改所有
3、可恢复性系统当中的有一部分组件消失不影响整个系统也就是说在消息队列当中即使有一个处理消息的进程失败一旦恢复还可以重启加入队列当中继续处理消息
4、缓中可以控制和优化数据经过系统的时间和速度解决生产消息和消费消息处理速度不一致的问题
5、峰值的处理能力消息队列在峰值情况下能够顶住突发的访问压力避免专门为了突发情况而对系统进行修改
6、异步通信允许用户把一个消息放入队列但是不立刻处理等用户想处理的时候在处理 消息队列的模式
1、点对点 一对一消息的生产者发送消息队列中消费者从队列中提取消息消费者提取完之后队列中被提取的消息将会被移除后续消费者不能再消费队列当中的消息消息队列可以有多个消费者但是一个消息只能由一个消费者提取
RABBITMQ
2、发布/订阅模式一对多又叫做观察者模式消费者提取数据之后队列当中的消息不会被清除
生产者发布一个消息到主题所有消费者都是通过主题获取消息
消息队列的重要组件
1、主题topic topic类似一个数据流的管道生产者把消息发布到主题消费者从那个主题当中订阅数据主题可以分区每个分区都有自己的偏移量
2、分区partition 每个主题都可以多个分区每个分区是数据的有序子集分区可以允许kafka进行水平拓展以处理大量数据消息在分钟按照偏移量存储消费者可以独立读取每个分区的数据
3、偏移量是每个消息在分区中唯一的标识消费者可以通过偏移量来跟踪获取已读或者未读消息的位置也可以提交偏移量来记录已处理的信息
4、生产者producer生产者把数据发送kafka的主题当中负责写入消息
5、消费者从主题当中读取数据消费者可以是一个也可以十多个每个消费者有一个唯一的消费者组IDkafka通过消费者实现负载均衡和容错性
6、经纪人Broker 每个kafka节点都有一个Broker每个负责一台kafka服务器id唯一存储主题分区当中数据处理生产和消费者的请求
维护元数据zookeeper
zookeeperzookeeper负责保存元数据元数据就是topic的相关信息发布在哪台主机上指定了多少分区以及副本数偏移量
zookeeper自建一个主题__consumer_offsets
3.0之后不依赖zookeeper的核心元数据由kafka节点自己管理 第一种begin从头开始获取所有
第二种实时获取我只消费后续产生的消息
第三种指定偏移量消费代码实现 kafka的工作流程
1、生产者发送数据到主题topic中
2、可以分区每个分区都有自己的偏移量
3、消费者从主题当中读取数据
工作流程图 生产者写入topic的数据时持久化的默认是7小时
至少一次语义只要消费者进入确保消息至少被消费一次
kafka实验部署
#解压kafka的源码包
tar -xf kafkamv kafkavim /etc/profile
export KAFKA_HOME/opt/kafka
export PATH$PATH:$KAFKA_HOME/bin#10服务器
#修改kafka的配置文件
cd /opt/kafka
cd /config
vim server.properties#21行
broker.id1
#全局唯一编号不可以重复
#28行
#如果全局遍布改变不需要改
#42行
num.network.threads3
#处理网络请求的线程数量默认即可
#46行
num.io.threads8
#处理磁盘的IO线程数量一定要比硬盘数大
#50行
socket.receive
#发送套接字的缓冲区大小
54行#接受者的接受套接字缓冲区大小套接字就是端口
58行#请求套接字的缓冲区大小
#65行
log.dirs/var/log/kafka
#日志路径
#70行
num.partitions1
#在此kafka服务器上创建topic默认分区数如果指定了这个配置无效了
#75行
num.recovery.threads.per.data.dir1
#用来恢复回收清理data下的数据的线程数量kafka的默认不允许删除主题的
#110行
log.retention.hours168
#生产者发布的数据文件在主题当中保存的时间168小时默认是7天
#130行
zookeeper.connect20.0.0.10:2181,20.0.0.11:2181,20.0.0.12:2181
#指定zookeeper集群#11服务器
vim server.properties#21行
broker.id2
#65行
log.dirs/var/log/kafka
#日志路径
#130行
zookeeper.connect20.0.0.10:2181,20.0.0.11:2181,20.0.0.12:2181
#指定zookeeper集群#12服务器
vim server.properties#21行
broker.id3
#65行
log.dirs/var/log/kafka
#日志路径
#130行
zookeeper.connect20.0.0.10:2181,20.0.0.11:2181,20.0.0.12:2181
#指定zookeeper集群#kafka的启动脚本
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
#查看kafka端口
netstat -antp | grep 9092
kafka创建主题
#所有可执行kafka的命令文件都要在bin目录下执行
cd /opt/kafka/bin#创建主题
kafka-topics.sh --create --zookeeper 20.0.0.10:2181,20.0.0.11:2181,20.0.0.12:2181 --replication-factor 2 --partitions 3 --topic xiaobu
#--zookeeper 指定的是zookeeper的地址和端口保存kafka的元数据
#--replication-factor 2 定义每个分区的副本数
#--partitions 3 指定主题的分区数
# --topic xiaobu 指定主题名可以自定义#查看主题
kafka-topics.sh --describe --zookeeper 20.0.0.10:2181,20.0.0.11:2181,20.0.0.12:2181#查看指定的主题详细信息
kafka-topics.sh --describe --zookeeper 20.0.0.10:2181,20.0.0.11:2181,20.0.0.12:2181 --topic xiaobu#Partition分区编号
#Leader每个分区都有一个领导者Leader领导者负责处理分区的读写操作。
#在上述输出中领导者的编号分别为 3、1、3。
#Replicas每个分区可以有多个副本Replicas用于提供冗余和容错性。
#在上述输出中Replica 3、1、2 分别对应不同的 Kafka broker。
#IsrISRIn-Sync Replicas表示当前与领导者保持同步的副本。
#ISR 3、1分别表示与领导者同步的副本。#做映射
vim /etc/hosts
20.0.0.10 test1
20.0.0.11 test2
20.0.0.12 test3随便选择一个主机发布消息
kafka-console-producer.sh --broker-list 20.0.0.10:9092,20.0.0.11:9092,20.0.0.12:9092 --topic xiaobu订阅信息从头开始
kafka-console-consumer.sh --bootstrap-server 20.0.0.10:9092,20.0.0.11:9092,20.0.0.12:9092 --topic xiaobu --from-beginning 订阅信息实时更新
kafka-console-consumer.sh --bootstrap-server 20.0.0.10:9092,20.0.0.11:9092,20.0.0.12:9092 --topic xiaobu在11上创建主题
kafka-topics.sh --create --zookeeper 20.0.0.11:2181 --replication-factor 1 --partitions 1 --topic xiaobu1
在10上订阅信息
kafka-console-consumer.sh --bootstrap-server 20.0.0.11:9092 --topic xiaobu1
在11上创建主题
kafka-topics.sh --create --zookeeper 20.0.0.12:2181 --replication-factor 1 --partitions 1 --topic xiaobu2
在10上订阅信息
kafka-console-consumer.sh --bootstrap-server 20.0.0.12:9092 --topic xiaobu2修改分区数
kafka-topics.sh --zookeeper 20.0.0.11:2181 --alter -topic xiaobu1 --partitions 3
#查看指定的主题详细信息
kafka-topics.sh --describe --zookeeper 20.0.0.11:2181 --topic xiaobu1
删除主题
kafka-topics.sh --delete --zookeeper 20.0.0.12:2181 --topic xiaobu1
#此时命令执行后只是打赏打上删除的标记并没有完全删除。还是保存在元数据当中
查看zookeeper保存信息
cd zookeeper/bin
./zkCli.sh -server 20.0.0.11:2181
#-server 20.0.0.11:2181指定需要登录的IP地址
ls /brokes/topics
#保存Kafka的元数据信息的位置。
#你无法对保存的元数据进行任何信息只可以查询
get /brokes/topics
#查看元数据信息
kafka3.0创建主题
#所有可执行kafka的命令文件都要在bin目录下执行
cd /opt/kafka/bin#创建主题
kafka-topics.sh --create --bootstrap-server 20.0.0.13:9092,20.0.0.40:9092,20.0.0.41:9092 --replication-factor 2 --partitions 3 --topic test1#查看主题
kafka-topics.sh --list --bootstrap-server 20.0.0.10:9092,20.0.0.11:9092,20.0.0.12:9092
指定查看主题
kafka-topics.sh --list --bootstrap-server 20.0.0.10:9092,20.0.0.11:9092,20.0.0.12:9092 test随便选择一个主机发布消息
kafka-console-producer.sh --broker-list 20.0.0.10:9092,20.0.0.11:9092,20.0.0.12:9092 test
kafkaelk集群 #两个es主机
#把filebeat的源码包拖到opt目录下
解压
cd filebeat
yum -y insatll nginx
systemctl restart nginx
vim /var/share/nginx/html/index.htmlvim filebeat.yml
- type: logenabled: truepaths:- /var/log/nginx/access.log- /var/log/nginx/error.logtags: [nginx]fields:service_name: 20.0.0.10_nginxlog_type: nginxfrom: 20.0.0.10
output.kafka:enabled: truehosts: [20.0.0.40:9092,20.0.0.41:9092,20.0.0.42:9092]topic: nginx运行filebeatnohup ./filebeat -e -c filebeat.yml filebeat.out cd /opt/kafka.conf
input {kafka {bootstrap_servers 20.0.0.40:9092,20.0.0.41:9092,20.0.0.42:9092topics nginxtype nginx_kafkacodec json解析json格式的代码auto_offset_reset earliest从头拉取latestdecorate_event true传递给es实例中的信息包含kafka的属性数据}}output{if nginx in [tags] {elasticsearch {hosts [20.0.0.10:9200,20.0.0.11:9200]index %{[fields][service_name]}-%{YYYY.MM.dd}}}}logstash -f kafka.conf --path.data /opt/nginx1去kafka上查看kafka-topic.sh --list --bootstrap-server 20.0.0.40:9092,20.0.0.41:9092,20.0.0.42:9092
