zookeeper+Kafka
两个都是消息队列的工具
消息队列
出现原因:生产者产生的消息与消费者处理消息的效率相差很大。为了避免出现数据丢失而设立的中间件。
在消息的生产者与消费之间设置一个系统,负责缓存生产者与消费者之间的消息的缓存。将消息排序。
优点:
解耦
将生产者与消费者的关系解开,这样避免一端的程序出问题导致整个两端程序都不可用。
冗余
为生产者与消费者之间产生的数据做了冗余
扩展性
利用zookeeper可以具备良好的扩展性。
灵活性
消息队列的对象可以是各种不同的程序
峰值处理能力
生产者产生的消息如果具有波动性,某一时刻峰值很高,利用消息队列就可以承载压力,然后慢慢分配给消息的消费者。
可恢复性
可以采用群集备份
顺序保证
消息以排序的方式存在
缓存
在生产者与消费者之间起到了一个缓冲的作用
异步通信
生产者产生的消息与消费者消费消息可以不再同一时间。
消息队列产品
kafka
apache旗下的产品
一种高吞吐量分布式发布/订阅消息队列的开源系统
kafka术语
broker
每一个安装了kafka的服务器都可以称为一个broker
topic
一类消息的集合被称为主题,一个broker可以有多个topic。
producer
消息生产者
consumer
消息消费者
partition
存放消息的区域,每一个topic可以有多个partition,但是处理还是一个一个区域处理。每一个区域内的消息都会被分配一个id。offset (偏移量)
consumer GROUP
消费者组
messages
消息,通信的基本单位。
kafka架构
需要注意的是,kafka作为一个支持多生产者多消费者的架构,再写入消息时允许多个生产者写道同一个partition,但是消费者读取的时候一个partition仅允许一个消费者消费,但一个消费者可以消费多个partition。
kafka支持消息持久化存储,消息会被存储在缓冲区内,达成一定条件就会被写入磁盘。在写入磁盘是是按照顺序的。
topic与partition的关系
partition的数量决定了组成topic的log的数量, 因此推荐partition的数量要大于同时允许的consumer数量,要小于等于集群broker的数量。
zookeeper
一种分布式协调技术,用于解决分布式环境中多个进程之间的同步控制,让他们有序的访问某种共享资源,尽量减少脑裂的发生。
开源的分布式协调服务
解决问题
单点故障
工作原理
master启动
节点向zookeeper注册信息,在选举产生master时采用最小编号算法,既最先的编号最小。
master故障
出现故障的节点信息会被自动删除,再通过选举投票产生新的master。
zookeeper判断节点故障的依据: 节点必须与zookeeper保持心跳连接。 节点能够及时响应zookeeper的请求。
master恢复
master恢复后,会再次向zookeeper注册自身节点信息,编号就会变淡。
架构图
四个角色
leader:领导者
负责发起投票与决议,更新系统状态
3888,选举通信端口
follower:追随者
负责请求客户端的请求并将结果给客户端,在选举过程中参与投票。
observer:观察者角色
负责接收客户端请求,并将请求转发给leader。但不参与投票,目的是扩展系统,提高伸缩性。
watcher:观察者,监视者
sentinel:哨兵
client:客户端
负责向zookeeper发起请求。
zookeeper在kafka中的作用
broker注册
每一个kafka的master节点都会在zookeeper中注册信息,提供id,每个master整个id都必须唯一。
topic注册
每一个主题也会在zookeeper注册信息。
生产者与消费者的负载均衡(实际上是高可用)
分区与消费者的联系
消费进度offset记录
消费者注册
根据注册信息将消费者分配到不同的分区,以达到负载均衡的目的。
zookeeper+kafka部署
本案例以zookeeper3.6版本与kafka2.13版本为例
单节点部署kafka进行测试
可以设置hosts文件进行域名通信
安装依赖环境
yum -y install java 注意是1.8版本。
解压即可用移动到/etc/zookeeper
修改配置文件
可以单独指定数据目录
dataDir=/etc/zookeeper/zookeeper-data
然后将指定的目录创建出来
启动查看状态
./bin/zkServer.sh start
单节点部署zookeeper+kafka做测试
部署zookeeper
可以设置hosts文件进行域名通信
安装依赖环境
yum -y install java 注意是1.8版本。
解压即可用移动到/etc/zookeeper
修改配置文件
可以单独指定数据目录
dataDir=/etc/zookeeper/zookeeper-data
然后将指定的目录创建出来
之后启动查看状态
部署kafka
同样的解压即可用移动到/etc/kafka
同样的可以修改配置文件指定日志文件目录并将目录创建出来
log.dirs=/etc/kafka/kafka-logs
mkdir /etc/kafka/kafka-logs
启动kafka需要指定配置文件并可以设置为后台启动。
bin/kafka-server-start.sh config/server.properties &
需要注意启动时要先启动zookeeper,结束时要先结束kafka。
测试
创建topic
kafka-topics.sh --create --zookeeper kafka1:2181 --replication-factor 1 --partitions 1 --topic test
kafka-topic.sh --create --zookeeper broker域名或IP:端口一般2181 --replication-factor 集群master数量 --partitions 1 topic中分区数量 --topic 分区名
列出topic
kafka-topics.sh --list --zookeeper brokerIP/域名:端口(2181)
查看topic
kafka-topics.sh --describe --zookeeper kafka1:2181 --topic test
describ :描述
生产消息进入命令行模式
/kafka-console-producer.sh --broker-list kafka1:9092 -topic tes
9092kafka监听客户端连接的端口
消费消息,进入查看产生消费消息的界面
kafka-console-consumer.sh --bootstrap-server kafka1:9092 --topic test
删除topic
kafka-topic.sh --delete --zookeeper Kafka服务器:端口 --topic 主题名
集群部署zookeeper+kafka
修改主机hosts文件,使之可以使用域名访问
安装zookeeper三台主机安装部署略有不同
准备前奏
防火墙与内核机制处理
安装依赖环境java
解压移动
tar zxvf apache-zookeeper-3.6.0-bin.tar.gz
mv apache-zookeeper-3.6.0-bin /etc/zookeeper
创建数据保存目录
mkdir zookeeper-data
修改配置文件
dataDir=/etc/zookeeper/zookeeper-data
指定数据保存目录
clientPort=2181
确定监听端口
添加节点注册信息有几个添加几个
server.1=192.168.10.101:2888:3888
2888:集群内部通信端口,主要用于选举leader
3888:其他服务器在选举leader时监听的端口。
创建id文件
echo '1' > /etc/zookeeper/zookeeper-data/myid
每个节点IP不同
最后启动并查看状态
安装kafka,节点配置相同
解压并移动
tar zxvf kafka_2.13-2.4.1.tgz
mv kafka_2.13-2.4.1 /etc/kafka
修改配置文件
listeners=PLAINTEXT://192.168.10.101:9092
监听客户端连接
log.dirs=/etc/kafka/kafka-logs
指定日志文件位置
num.partitions=1
分区数量,不能超过节点数。
集群地址信息
zookeeper.connect=192.168.10.101:2181,192.168.10.102:2181,192.168.10.103:2181
创建日志目录
mkdir /etc/kafka/kafka-logs
最后启动查看状态
测试
在任意一节点上创建topic
列出任意一节点的topic
进行生产与消费的对应
ELK日志系统添加接口对接Kafka
input { beats { port => "5044" codec => "json" } } output { kafka { bootstrap_servers => "192.168.10.101:9092,192.168.10.102:9092,192.168.10.103:9092" topic_id => "httpd-filebeat" batch_size => "5" codec => "json" } }
该集群中用到的端口
zookeeper客户端连接的端口:2181
zookeerper内部通信端口:2888
zookeeper监听选举过程的端口:3888
kafka监听客户端的接口:9092
zookeeper+kafka的消息队列
2024-07-16 10:26:01 19 阅读