目录
zookeeper集群
概念
kafka 3.4.1 在命令中已经不需要zookeeper
zookeeper是一个开源的,分布式的,为分布式架构提供协调服务的APACHE的项目。
zookeeper的工作机制:观察者模式设计的分布式服务器管理架构。负责存储和管理元数据,记录集群的变化,保存集群的变化的信息。
zookeeper的特点:
1.在集群中分为领导者和追随者组成的集群
2.只要有半数以上的节点正常工作,整个zookeeper就可以正常工作。zookeeper在部署时一般选择奇数台
3.全局数据一致,每个zookeeper不论是领导者还是追随者,在访问他们的数据时都是一致的。
4.数据更新的原子性,一次更新数据,那么都成功,那么都失败
5.数据更新的实时性
6.领导者和追随者投票产生
选举机制:A、B、C
1)服务器A启动,发起一次选举,A会投自己一票,此时A有一票,但不够半数,选举无法完成,A进行looking
2)服务器B启动,再发起一次选举,服务器B也投自己一票,服务A和服务B会做个myid比较,比较谁的myid大,如果A比B小,A会把票改投给B,此时B有2票,B自动当选为leader
3)C启动,自动成为追随者,A也会成为追随者
数据流向图
zookeeper集群实验
负责管理和收集zookeeper数据
架构
192.168.233.61
192.168.233.62
192.168.233.63
实现步骤
1.安装ntpdate(所有设备都要安装)
yum -y install ntpdate -y
然后查看时间是否一致
2.安装java(所有设备都要安装)
yum install -y java-1.8.0-openjdk java-1.8.0-openjdk-devel
3.安装zookeeper(所有设备都要安装)
把apache-zookeeper-3.5.7-bin.tar拖入到三台虚拟机中
tar -xf apache-zookeeper-3.5.7-bin.tar.gz
mv apache-zookeeper-3.5.7-bin /opt/zookeeper
cd zookeeper/
cd conf/
cp zoo_sample.cfg zoo.cfg
vim zoo.cfg
server.1 数字id,也就是服务器对应的myid
3188:zookeeper集群内部通信的端口
3288:重新选举端口,万一leader挂了,用这个端口进行内部通信,选举新的leader
依旧是三台机同步操作
mkdir /opt/zookeeper/data
mkdir /opt/zookeeper/logs
然后分别在61 echo 1 > /opt/zookeeper/data/myid
在62 echo 2 > /opt/zookeeper/data/myid
在63 echo 3 > /opt/zookeeper/data/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
kafka集群
概念
kafka就是消息队列(MQ)
应用场景:在高并发环境下,同步的请求来不及处理,请求太多会造成阻塞,比如说大量请求并发到数据库,会报错too many connection,此时消息队列,会使用异步处理方式,可以缓解系统处理请求的压力。
消息队列流程图
kafka的作用:
1.异步处理
2.系统解耦
解耦:每个系统之间独立运行,互相之间没有必然的依赖关系。
应用场景:微服务架构中的通信对于解耦来说至关重要。各个微服务之间独立运行,分别处理各自的请求和消息。提高整个系统的吞吐量和处理能力。
电商的订单系统、网站的工单系统,都是典型的一个消息队列场景。
3.负载均衡
消息队列的负载均衡:把任务发送到多个消费者,多个消费者可以并行处理队列中的消息
4.流量控制和限流
通过延迟方法,处理生成速率和消费者的处理速度(代码控制)
5.数据同步和分发
跨系统的数据同步和日志收集
6.任务调度和定时任务
7.实时数据处理
8.备份和恢复
消息队列的模式
1.点对点:即一对一 ,可以理解为一个生产者对应一个消费者(淘汰) 消费者消费完数据之后,生产者会自动清除已消费的数据
2.发布/订阅模式:即一对多,又叫观察者模式 消费者数据在消费完之后不会被清除,会保留一段时间)
一对多:生产者发布一个消息,可以是一个消费者使用,也可以是多个消费者同时使用(主流)
kafka就是发布/订阅模式的消息队列,主要用于大数据的实时处理领域
RAbbitMQ也是发布/订阅模式的消息队列,小集群内部使用
kafka的特性:
1.高吞吐量,低延迟
每秒可以处理几十万条数据,延迟只有几毫秒
2.集群的可拓展性(热拓展)
3.消息的持久化
生成者发布的消息可以保存到磁盘当中,防止数据丢失(有时间限制)
4容错性
挂了一个也可以继续使用
5.高并发
数千个客户端可以同时丢写
kafka的组件
topic又叫主题:是kafka的基本单元,所有生产者发布的消息都是发布到主题
消费者订阅主题,然后消费生产者发布的消息
生产者:生产者把消息发布到主题
消费者:订阅主题,消费生产者发布的消息
分区:每个主题都可以分成多个分区,每个分区都是数据的有序子集
分区当中保留数据,按照偏移量来有序的存储数据。消费者可以根据偏移量来消费指定分区当中的消息(一般不用)
分区还有备份的作用:我们在创建主题时创建分区,创建分区时要指定副本数。
分区和我们执行的集群机器数量一般是保持一致的。
副本:备份分区中的消息,最少要2个,互为备份。
偏移量:消息在分区当中的唯一标识,就是跟踪和定位消息所在的位置,消费者可以根据偏移量来处理信息
经纪人 broker:经纪人处理生成者和消费者的请求(现在都是kafka处理)元数据(还是zookeeper)
zookeeper:保存元数据
kafka的工作流程
生产者将消息发布到指定的主机,每个消息都附带一个key和value。
主题是有多个分区的,生产者把消息写入一个分区(带偏移量)
经纪人(kafka):分配和处理生产者的发布请求,偏移量也是经纪人分配(在分区中是唯一的)
消费者订阅主题,获取全量的消费者的消费信息(默认模式),也可以从执行的分区获取消息(代码来完成,一般不用)
生产者发布的消息会在本地保留一段时间,防止消费者有延迟或者处理速度过慢,导致没有成功消费。保留时间:7天
kafka集群实验
实验基于zookeeper实验
1.安装kafka(三台虚拟机同时操作)
把kafka_2.13-3.4.1拖入三台虚拟机
tar -xf kafka_2.13-3.4.1.tgz
mv kafka_2.13-3.4.1 /usr/local/kafka
cd /usr/local/kafka/config/
cp server.properties server.properties.bak
vim server.properties
这里的id就是经纪人的id,每台机器不能一样
指定本机地址,每台机都要改
了解即可,一般不动
指定日志位置,三台机都一样
指定zookeeper集群的ip地址,三台机同步
然后 vim /etc/profile (三台机同步)
在最后一行添加
export KAFKA_HOME=/usr/local/kafka
export PATH=$PATH:$KAFKA_HOME/bin
source /etc/profile
然后配置kafka的启动脚本(三台同时操作)
vim /etc/init.d/kafka
#!/bin/bash
#chkconfig:2345 22 88
#description:Kafka Service Control Script
KAFKA_HOME='/usr/local/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
chmod +x /etc/init.d/kafka
chkconfig --add kafka
service kafka start
netstat -antp | grep 9092 查看kafka的端口是否启动
创建主题
kafka-topics.sh --create --bootstrap-server 192.168.233.61:9092,192.168.233.62:9092,192.168.233.63:9092 --replication-factor 2 --partitions 3 --topic test1
--replication-factor 2 创建分区的副本数 (最少2个)
--partitions 3 分区数
--topic test1 指定主题的名称
生产者发布信息
kafka-console-producer.sh --broker-list 192.168.233.61:9092,192.168.233.62:9092,192.168.233.63:9092 --topic test1
消费者订阅,消费信息
kafka-console-consumer.sh --bootstrap-server 192.168.233.61:9092,192.168.233.62:9092,192.168.233.63:9092 --topic test1 --from-beginning
查看topic当中的主题
kafka-topics.sh --list --bootstrap-server 192.168.233.61:9092,192.168.233.62:9092,192.168.233.63:9092
查看主题的详细信息
kafka-topics.sh --describe --bootstrap-server 192.168.233.61:9092,192.168.233.62:9092,192.168.233.63:9092
topic:主题名称
partition:分区、偏移量
leader:分区的领导者,用来处理分区的读写操作,只有在指定写分区和写分区时才工作,如果不是指定,全量展示,无意义
replicas:副本 0,1,2 对应broker.id
lsr:表示当前与领导者同步的副本
删除主题
kafka-topics.sh --delete --bootstrap-server 192.168.233.61:9092,192.168.233.62:9092,192.168.233.63:9092 --topic test1
修改分区数
kafka-topics.sh --bootstrap-server 192.168.233.61:9092,192.168.233.62:9092,192.168.233.63:9092 --alter --topic test1 --partitions 6
总结
zookeeper就是保存集群的元数据
面试题:kafka的消息堆积如何解决?
原因:消费者出现了延迟或者处理能力太差,导致消息堆积。
解决办法:1.减少kafka持久化的保存时间
2.修改主题的分区数,扩大分区的数量,提高消费者获取的通道
kafka-topics.sh --bootstrap-server 192.168.233.61:9092,192.168.233.62:9092,192.168.233.63:9092 --alter --topic test1 --partitions 6
3.可以指定多个消费者共同工作,处理消息的积压(代码完成)
标签:--,分区,zookeeper,9092,kafka,集群,192.168 From: https://blog.csdn.net/Hai990218/article/details/140918790