架构师系列- 消息中间件(12)-kafka基础

news/2024/5/20 22:04:30

1、应用场景

1.1 kafka场景

Kafka最初是由LinkedIn公司采用Scala语言开发,基于ZooKeeper,现在已经捐献给了Apache基金会。目前Kafka已经定位为一个分布式流式处理平台,它以 高吞吐、可持久化、可水平扩展、支持流处理等多种特性而被广泛应用。

Apache Kafka能够支撑海量数据的数据传递。在离线和实时的消息处理业务系统中,Kafka都有广泛的应用。

(1)日志收集:收集各种服务的log,通过kafka以统一接口服务的方式开放 给各种consumer,例如Hadoop、Hbase、Solr等;

(2)消息系统:解耦和生产者和消费者、缓存消息等;

(3)用户活动跟踪:Kafka经常被用来记录web用户或者app用户的各种活动,如浏览网页、搜索、点击等活动,这些活动信息被各个服务器发布到kafka的topic中,然后订阅者通过订阅这些topic来做实时 的监控分析,或者装载到Hadoop、数据仓库中做离线分析和挖掘;

(4)运营指标:Kafka也经常用来记录运营监控数据。包括收集各种分布式应用的数据,生产各种操作的集中反馈,比如报警和报告;

(5)流式处理:比如spark streaming和storm;

1.2 kafka特性

kafka以高吞吐量著称,主要有以下特性:

(1)高吞吐量、低延迟:kafka每秒可以处理几十万条消息,它的延迟最低只有几毫秒;

(2)可扩展性:kafka集群支持热扩展;

(3)持久性、可靠性:消息被持久化到本地磁盘,并且支持数据备份防止数据丢失;

(4)容错性:允许集群中节点失败(若副本数量为n,则允许n-1个节点失败);

(5)高并发:支持数千个客户端同时读写;

1.3 消息对比

  • 如果普通的业务消息解耦,消息传输,rabbitMq是首选,它足够简单,管理方便,性能够用。
  • 如果在上述,日志、消息收集、访问记录等高吞吐,实时性场景下,推荐kafka,它基于分布式,扩容便捷
  • 如果很重的业务,要做到极高的可靠性,考虑rocketMq,但是它太重。需要你有足够的了解

1.4 大厂应用

  • 京东通过kafka搭建数据平台,用于用户购买、浏览等行为的分析。成功抗住6.18的流量洪峰
  • 阿里借鉴kafka的理念,推出自己的rocketmq。在设计上参考了kafka的架构体系

2、基础组件

2.1 角色

  • broker:节点,就是你看到的机器
  • provider:生产者,发消息的
  • consumer:消费者,读消息的
  • zookeeper:信息中心,记录kafka的各种信息的地方
  • controller:其中的一个broker,作为leader身份来负责管理整个集群。如果挂掉,借助zk重新选主

2.2 逻辑组件

  • topic:主题,一个消息的通道,收发总得知道消息往哪投
  • partition:分区,每个主题可以有多个分区分担数据的传递,多条路并行,吞吐量大
  • Replicas:副本,每个分区可以设置多个副本,副本之间数据一致。相当于备份,有备胎更可靠
  • leader & follower:主从,上面的这些副本里有1个身份为leader,其他的为follower。leader处理partition的所有读写请求

2.3 副本集合

  • AR(assigned replica):所有副本的统称,AR=ISR+OSR
  • ISR(In-sync Replica):同步中的副本,可以参与leader选主。一旦落后太多(数量滞后和时间滞后两个维度)会被踢到OSR。
  • OSR(Out-Sync Relipcas):踢出同步的副本,一直追赶leader,追上后会进入ISR

2.4 消息标记

  • offset:偏移量,消息消费到哪一条了?每个消费者都有自己的偏移量
  • HW:(high watermark):副本的高水印值,客户端最多能消费到的位置,HW值为8,代表offset为[0,8]的9条消息都可以被消费到,它们是对消费者可见的,而[9,12]这4条消息由于未提交,对消费者是不可见的。
  • LEO:(log end offset):日志末端位移,代表日志文件中下一条待写入消息的offset,这个offset上实际是没有消息的。不管是leader副本还是follower副本,都有这个值。

那么这三者有什么关系呢?

比如在副本数等于3的情况下,消息发送到Leader A之后会更新LEO的值,Follower B和Follower C也会实时拉取Leader A中的消息来更新自己,HW就表示A、B、C三者同时达到的日志位移,也就是A、B、C三者中LEO最小的那个值。由于B、C拉取A消息之间延时问题,所以HW一般会小于LEO,即LEO>=HW。

具体的同步原理,下面章节会详细讲到

3、架构探索

3.1 发展历程

Apache Kafka

3.1.1 版本命名

Kafka在1.0.0版本前的命名规则是4位,比如0.8.2.1,0.8是大版本号,2是小版本号,1表示打过1个补丁

现在的版本号命名规则是3位,格式是“大版本号”+“小版本号”+“修订补丁数”,比如2.5.0,前面的2代表的是大版本号,中间的5代表的是小版本号,0表示没有打过补丁

我们所看到的下载包,前面是scala编译器的版本,后面才是真正的kafka版本。

3.1.2 演进历史

0.7版本 只提供了最基础的消息队列功能。

0.8版本 引入了副本机制,至此Kafka成为了一个真正意义上完备的分布式高可靠消息队列解决方案。

0.9版本 增加权限和认证,使用Java重写了新的consumer API,Kafka Connect功能;不建议使用consumer API;

0.10版本 引入Kafka Streams功能,正式升级成分布式流处理平台;建议版本0.10.2.2;建议使用新版consumer API

0.11版本 producer API幂等,事务API,消息格式重构;建议版本0.11.0.3;谨慎对待消息格式变化

1.0和2.0版本 Kafka Streams改进;建议版本2.0;

3.2 集群搭建

1)原生启动

kafka启动需要zookeeper,第一步启动zk:

docker run --name zookeeper-1 -d -p 2181 zookeeper:3.4.13

原生安装:下载后解压启动即可 Apache Kafka

bin/kafka-server-start.sh config/server.properties

#server.properties配置说明
#表示broker的编号,如果集群中有多个broker,则每个broker的编号需要设置的不同
broker.id=0 
#brokder对外提供的服务入口地址,默认9092
listeners=PLAINTEXT://:9092 
#设置存放消息日志文件的地址
log.dirs=/tmp/kafka/log 
#Kafka所需Zookeeper集群地址,这里是关键!加入同一个zk的kafka为同一集群
zookeeper.connect=zookeeper:2181 

2)推荐docker-compose 一键启动

#参考资料中的kafka.yml
#注意hostname问题,ip地址:192.168.10.30,换成你自己服务器的
#docker-compose -f kafka.yml up -d 启动
version: '3'
services:zookeeper:image: zookeeper:3.4.13kafka-1:container_name: kafka-1image: wurstmeister/kafka:2.12-2.2.2ports:- 10903:9092environment:KAFKA_BROKER_ID: 1 HOST_IP: 192.168.10.30KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181#docker部署必须设置外部可访问ip和端口,否则注册进zk的地址将不可达造成外部无法连接KAFKA_ADVERTISED_HOST_NAME: 192.168.10.30KAFKA_ADVERTISED_PORT: 10903 volumes:- /etc/localtime:/etc/localtimedepends_on:- zookeeper           kafka-2:container_name: kafka-2image: wurstmeister/kafka:2.12-2.2.2ports:- 10904:9092environment:KAFKA_BROKER_ID: 2 HOST_IP: 192.168.10.30KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181KAFKA_ADVERTISED_HOST_NAME: 192.168.10.30KAFKA_ADVERTISED_PORT: 10904 volumes:- /etc/localtime:/etc/localtimedepends_on:- zookeeper 

3.3 组件探秘

命令行工具是管理kafka集群最直接的工具。官方自带,不需要额外安装。

3.2.1 主题创建

#进入容器
docker exec -it kafka-1 sh
#进入bin目录
cd /opt/kafka/bin
#创建
kafka-topics.sh --zookeeper zookeeper:2181 --create --topic test --partitions 2 --replication-factor 1

3.2.2 查看主题

kafka-topics.sh --zookeeper zookeeper:2181 --list

3.2.3 主题详情

kafka-topics.sh --zookeeper zookeeper:2181 --describe --topic test#分析输出:
Topic:test  PartitionCount:2    ReplicationFactor:1 Configs:Topic: test Partition: 0    Leader: 2   Replicas: 2 Isr: 2Topic: test Partition: 1    Leader: 1   Replicas: 1 Isr: 1

3.2.4 消息收发

#使用docker连接任意集群中的一个容器
docker exec -it kafka-1 sh#进入kafka的容器内目录
cd /opt/kafka/bin#客户端监听
./kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test#另起一个终端,验证发送
./kafka-console-producer.sh --broker-list localhost:9092 --topic test

 3.2.5 分组消费

#启动两个consumer时,如果不指定group信息,消息被广播
#指定相同的group,让多个消费者分工消费(画图:group原理)./kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --group aaa#结果:在发送方,连续发送 1-4 ,4条消息,同一group下的两台consumer交替消费,并发执行

注意!!!

这是在消费者和分区数相等(都是2)的情况下。 如果同一group下的 ( 消费者数量 > 分区数量 ) 那么就会有消费者闲置。

验证方式:

可以再多启动几个消费者试一试,会发现,超出2个的时候,有的始终不会消费到消息。 停掉可以消费到的,那么闲置的会被激活,进入工作状态

3.2.6 指定分区 

#指定分区通过参数 --partition,注意!需要去掉上面的group
#指定分区的意义在于,保障消息传输的顺序性(画图:kafka顺序性原理)
./kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --partition 0
./kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --partition 1#结果:发送1-4条消息,交替出现。说明消息被均分到各个分区中投递#默认的发送是没有指定key的
#要指定分区发送,就需要定义key。那么相同的key被路由到同一个分区
./kafka-console-producer.sh --broker-list kafka-1:9092 --topic test --property parse.key=true#携带key再发送,注意key和value之间用tab分割
>1  1111
>1  2222
>2  3333
>2  4444#查看consumer的接收情况
#结果:相同的key被同一个consumer消费掉

 3.2.7 偏移量

#偏移量决定了消息从哪开始消费,支持:开头,还是末尾# earliest: 当各分区下有已提交的offset时,从提交的offset开始消费;无提交的offset时,从头开始消费
# latest: 当各分区下有已提交的offset时,从提交的offset开始消费;无提交的offset时,消费新产生的该分区下的数据
# none: topic各分区都存在已提交的offset时,从offset后开始消费;只要有一个分区不存在已提交的offset,则抛出异常# 注意点!!!有提交偏移量的话,仍然以提交的为主,即便使用earliest,比提交点更早的也不会被提取#--offset [earliest|latest(默认)] , 或者 --from-beginning
#新起一个终端,指定offset位置
./kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --partition 0 --offset earliest./kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --partition 0 --from-beginning#结果:之前发送的消息,从头又消费了一遍!

3.4 zk探秘

前面说过,zk存储了kafka集群的相关信息,本节来探索内部的秘密。

kafka的信息记录在zk中,进入zk容器,查看相关节点和信息

docker exec -it kafka-zookeeper-1 sh>./bin/zkCli.sh>ls /#结果:得到以下配置信息

3.4.1 broker信息 

[zk: localhost:2181(CONNECTED) 0] ls /brokers
[ids, topics, seqid]
[zk: localhost:2181(CONNECTED) 1] ls /brokers/ids
[1, 2]#机器broker信息
[zk: localhost:2181(CONNECTED) 4] get /brokers/ids/1
{"listener_security_protocol_map":{"PLAINTEXT":"PLAINTEXT"},"endpoints":["PLAINTEXT://192.168.10.30:10903"],"jmx_port":-1,"host":"192.168.10.30","timestamp":"1609825245500","port":10903,"version":4}
cZxid = 0x27
ctime = Tue Jan 05 05:40:45 GMT 2021
mZxid = 0x27
mtime = Tue Jan 05 05:40:45 GMT 2021
pZxid = 0x27
cversion = 0
dataVersion = 1
aclVersion = 0
ephemeralOwner = 0x105a2db626b0000
dataLength = 196
numChildren = 0

 3.4.2 主题与分区

#分区节点路径
[zk: localhost:2181(CONNECTED) 5] ls /brokers/topics
[test, __consumer_offsets]
[zk: localhost:2181(CONNECTED) 6] ls /brokers/topics/test
[partitions]
[zk: localhost:2181(CONNECTED) 7] ls /brokers/topics/test/partitions
[0, 1]
[zk: localhost:2181(CONNECTED) 8] ls /brokers/topics/test/partitions/0
[state]#分区信息,leader所在的机器id,isr列表等
[zk: localhost:2181(CONNECTED) 18] get /brokers/topics/test/partitions/0/state
{"controller_epoch":1,"leader":1,"version":1,"leader_epoch":0,"isr":[1]}
cZxid = 0xb0
ctime = Tue Jan 05 05:56:06 GMT 2021
mZxid = 0xb0
mtime = Tue Jan 05 05:56:06 GMT 2021
pZxid = 0xb0
cversion = 0
dataVersion = 0
aclVersion = 0
ephemeralOwner = 0x0
dataLength = 72
numChildren = 0

 3.4.3 消费者与偏移量

[zk: localhost:2181(CONNECTED) 15] ls /consumers
[]
#空的???
#那么,消费者以及它的偏移记在哪里呢???

kafka 消费者记录 group 的消费 偏移量 有两种方式 :

1)kafka 自维护 (新)

2)zookpeer 维护 (旧) ,已经逐渐被废弃

查看方式:

上面的消费用的是控制台工具,这个工具使用--bootstrap-server,不经过zk,也就不会记录到/consumers下。

其消费者的offset会更新到一个kafka自带的topic【__consumer_offsets】下面

#先起一个消费端,指定group
./kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --group aaa#使用控制台工具查看消费者及偏移量情况
./kafka-consumer-groups.sh --bootstrap-server kafka-1:9092 --list
KMOffsetCache-44acff134cad
aaa#查看偏移量详情
./kafka-consumer-groups.sh --bootstrap-server kafka-1:9092 --describe --group aaa

当前与LEO保持一致,说明消息都完整的被消费过

 停掉consumer后,往provider中再发几条记录,offset开始滞后:

重新启动consumer,消费到最新的消息,同时再返回看偏移量,消息得到同步:

 3.4.4 controller

#当前集群中的主控节点是谁
[zk: localhost:2181(CONNECTED) 17] get /controller
{"version":1,"brokerid":1,"timestamp":"1609825245694"}
cZxid = 0x2a
ctime = Tue Jan 05 05:40:45 GMT 2021
mZxid = 0x2a
mtime = Tue Jan 05 05:40:45 GMT 2021
pZxid = 0x2a
cversion = 0
dataVersion = 0
aclVersion = 0
ephemeralOwner = 0x105a2db626b0000
dataLength = 54
numChildren = 0

3.5 km

3.5.1 启动

kafka-manager是目前最受欢迎的kafka集群管理工具,最早由雅虎开源。提供可视化kafka集群操作

官网:Releases · yahoo/CMAK · GitHub

注意它的版本,docker社区的镜像版本滞后于kafka,我们自己来打镜像。

#Dockerfile
FROM daocloud.io/library/java:openjdk-8u40-jdk
ADD kafka-manager-2.0.0.2/ /opt/km2002/
CMD ["/opt/km2002/bin/kafka-manager","-Dconfig.file=/opt/km2002/conf/application.conf"]#打包,注意将kafka-manager-2.0.0.2放到同一目录
docker build -t km:2002 .# 还可以直接拉取
docker pull liggdocker/km:2002
# 修改镜像标签为km:2002
docker tag imageId km:2002#启动:在上面的yml里,services节点下加一段
#参考资料:km.yml
#执行: docker-compose -f km.yml up -dkm:image: liggdocker/km:2002ports:- 10906:9000depends_on:- zookeeper#完整的km.yml内容
version: '3'
services:zookeeper:image: zookeeper:3.4.13kafka-1:container_name: kafka-1image: wurstmeister/kafka:2.12-2.2.2ports:- 10903:9092environment:KAFKA_BROKER_ID: 1HOST_IP: 192.168.10.30KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181#docker部署必须设置外部可访问ip和端口,否则注册进zk的地址将不可达造成外部无法连接KAFKA_ADVERTISED_HOST_NAME: 192.168.10.30KAFKA_ADVERTISED_PORT: 10903volumes:- /etc/localtime:/etc/localtimedepends_on:- zookeeperkafka-2:container_name: kafka-2image: wurstmeister/kafka:2.12-2.2.2ports:- 10904:9092environment:KAFKA_BROKER_ID: 2HOST_IP: 192.168.10.30KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181KAFKA_ADVERTISED_HOST_NAME: 192.168.10.30KAFKA_ADVERTISED_PORT: 10904volumes:- /etc/localtime:/etc/localtimedepends_on:- zookeeperkm:image: liggdocker/km:2002ports:- 10906:9000depends_on:- zookeeper

3.5.2 使用

使用km可以方便的查看以下信息:

  • cluster:创建集群,填写zk地址,选中jmx,consumer信息等选项
  • brokers:列表,机器信息
  • topic:主题信息,主题内的分区信息。创建新的主题,增加分区
  • cosumers: 消费者信息,偏移量等

http://www.mrgr.cn/p/31177628

相关文章

ubutu ac68

1、直接去 https://github.com/morrownr/8814au 下载编译,运行 install-driver.sh,重启后 2、 3、 断开再连,然后usb拔掉再插,选择 虚拟机的系统 4 抓 https://blog.csdn.net/weixin_41856150/article/details/102327913

《痞子衡嵌入式半月刊》 第 98 期

痞子衡嵌入式半月刊: 第 98 期这里分享嵌入式领域有用有趣的项目/工具以及一些热点新闻,农历年分二十四节气,希望在每个交节之日准时发布一期。 本期刊是开源项目(GitHub: JayHeng/pzh-mcu-bi-weekly),欢迎提交 issue,投稿或推荐你知道的嵌入式那些事儿。 上期回顾 :《…

刷题训练之前缀和

> 作者:დ旧言~ > 座右铭:松树千年终是朽,槿花一日自为荣。 > 目标:熟练掌握前缀和算法。 > 毒鸡汤:学习,学习,再学习 ! 学,然后知不足。 > 专栏选自:刷题…

瑞士轮——结构体(快速排序 or 归并排序?)

瑞士轮——结构体&&(快速排序 or 归并排序?)题目链接:https://www.luogu.com.cn/problem/P1309 题意应该非常明确了(这里就不细讲了):有2*N个人,首先根据成绩进行排序,相邻的两个人进行比赛,强的人成绩+1,输的人成绩不变,最后又根据成绩进行排序,进行r次操作,…

管理集群工具之LVS

管理集群工具之LVS 集群概念 将很多机器组织在一起,作为一个整体对外提供服务集群在扩展性、性能方面都可以做到很灵活集群分类 负载均衡集群:Load Balance高可用集群:High Availability高性能计算:High Performance Computing …

pytest系列——allure之在测试用例添加标题(@allure.title())

前言 通过使用装饰器allure.title可以为测试用例自定义一个更具有阅读性的易读的标题。 allure.title的三种使用方式: 直接使用allure.title为测试用例自定义标题;allure.title支持通过占位符的方式传递参数,可以实现测试用例标题参数化&a…

C++|stack-queue-priority_queue(适配器+模拟实现+仿函数)

目录 一、容器适配器 1.1容器适配器概念的介绍 1.2stack和queue的底层结构 1.3deque容器的介绍 1.3.1deque的缺陷及为何选择他作为stack和queue的底层默认实现 二、stack的介绍和使用 2.1stack的介绍 2.2stack的使用 2.3stack的模拟实现 三、queue的介绍和使用 …

练习安装Python扩展库

(三)、练习安装Python扩展库 【实验截图】 1、在资源管理器中进入 Python 安装目录的 scripts 子目录,然后按下 Shift 键,在空 白处单击鼠标右键,在弹出来的菜单中选择“在此处打开命令窗口”进入命令提示符环境2.使用 pip 命令在线安装 Python 扩展库 numpy、pandas、sci…

Spark-机器学习(3)回归学习之线性回归

在之前的文章中,我们了解我们的机器学习,了解我们spark机器学习中的特征提取和我们的tf-idf,word2vec算法。想了解的朋友可以查看这篇文章。同时,希望我的文章能帮助到你,如果觉得我的文章写的不错,请留下你…

ZeRO论文阅读

一.前情提要 1.本文理论为主,并且仅为个人理解,能力一般,不喜勿喷 2.本文理论知识较为成体系 3.如有需要,以下是原文,更为完备 Zero 论文精读【论文精读】_哔哩哔哩_bilibili 二.正文 1.前言 ①为什么用该技术&…

4.25日团队开发第六天

今天进行了晨会晨会成员:董茂欣、龚涵彬、刘雪超 主要进行了团队内容完成分析,以及不会点的讨论,调用后端接口完成视频的播放

企业网架构与安全设备部署

在现代网络中,为了满足不同规模和需求的组织和企业的通信需求,网络架构通常会划分为多个层次,其中包括接入层、汇聚层和核心层。目录企业网三层架构常见安全设备网络区域划分网络架构拓扑示例 企业网三层架构 在现代网络中,为了满足不同规模和需求的组织和企业的通信需求,…

Typora for Mac:轻量级Markdown编辑器

Typora for Mac是一款专为Mac用户设计的轻量级Markdown编辑器,它以其简洁的界面和强大的功能,成为了Markdown写作爱好者的首选工具。 Typora for Mac v1.8.10中文激活版下载 Typora的最大特色在于其所见即所得的编辑模式,用户无需关心复杂的M…

Ubuntu部署jmeter与ant

为了整合接口自动化的持续集成工具,我将jmeter与ant都部署在了Jenkins容器中,并配置了build.xml 一、ubuntu部署jdk 1:先下载jdk-8u74-linux-x64.tar.gz,上传到服务器,这里上传文件用到了ubuntu 下的 lrzsz。 ubunt…

【工作】比亚迪工作笔记2——入职两周

1、工作制度 比亚迪每天打卡3次。 》早上弹性打卡上班,可以在8:30~9:30之间打卡。9:30之后算迟到。 》中午打卡时间12:00~13:00。大部分人选择12:01打卡然后去吃饭。 》晚上打卡时间,要求早晚打卡之间不少于9小时(理想情况下)。 实际上虹桥这边很少按点下班,工作到9…

2024第十五届蓝桥杯网络安全赛项WriteUp

欢迎关注公众号【Real返璞归真】回复【蓝桥杯2024】获取完整题目附件。 排名 安全知识 错1个选择题,题目说的不清楚,没搞懂题意。肯定不能用eval。错了理论题有点遗憾。 没想到这题前端是要解析json数据,排除CD选了A,结果发现正…

【项目学习01_2024.04.27_Day01】

学习笔记 项目学习链接第2章 内容管理模块v3.11 模块需求分析1.1 什么是需求分析1.2 模块介绍1.3 业务流程1.4 界面原型 2 创建模块工程2.1 模块工程结构父工程和子工程之间的继承关系以及工程与工程之间的依赖关系,通俗理解:2.2 创建模块工程\pom\含义及…

vue2 学习笔记

视屏地址:https://www.bilibili.com/video/BV1Zy4y1K7SH?p=14&spm_id_from=pageDriver&vd_source=ad97a93a8a42c9559b03a66114d94d18 vue2 学习笔记: 1. 对象里面写方法,不用写 function 关键字。比如:

一个算法工程师的学习

本系列主要记录在算法工作模型训练过程中一些列用到的技术。训练模型需要学习和积累的知识非常多,在学习了之后会做一个总结,并记录在这里。voc数据集转换成coco数据集