当前位置： 首页 > news >正文

管理流创建schema流程源码解析

news 2025/7/1 9:32:44

一、简析

schema是pulsar重要的功能之一，现在就一起从源码的视角看下管理流创建schema时客户端和服务端的表现
在这里插入图片描述

客户端

客户端主要经历以下四个步骤

  1. 创建Schema实例

    根据数据类型创建相对应的实例，例如Avro创建AvroSchema、JSON创建JSONSchema等

  2. 获取处理Schema的对象

    管理流PulsarAdmin对象获取SchemasImpl对象，这个对象是专门处理所有schema相关的操作。除此之外PulsarAdmin对象还维护着Clusters、Brokers、Tenants等等管理维护集群的重要对象，通过这些对象可以很好的管理维护Pulsar集群

  3. 构造SchemaInfo

    通过Schema实例创建其对应的SchemaInfo信息，里面就包括这个schema的名字、schema的结构化信息、schema类型等等，最后SchemaInfo这个对象会转成字符串发到服务端

  4. 发送HTTP请求

    通过post请求将数据发到服务端，这里是通过Java Rest库javax.ws.rs-api进行处理的

服务端

服务端主要经历以下四个步骤

  1. 参数格式校验

    • 校验租户、命名空间的是否有效(判空、是否有特殊字符)
    • 从缓存中根据Topic获取其对应的TopicName对象

  2. 权限校验

    • 判断是否是Topic的owner
    • 判断当前用户是否有操作当前Topic的权限

  3. SchemaRegistry注册schema

    SchemaRegistryService是服务端处理所有schema相关的对象，而schema相关的读写操作是依赖它的成员SchemaStorage进行处理的，SchemaStorage的最终是通过Bookkeeper客户端对象发送写请求

  4. 写Bookkeeper

    通过LedgerHandle对象向Bookkeeper服务端发送写请求

小结

schame新建流程概括起来就是，客户端构造schema信息，服务端负责schema校验，bookkeeper负责schema的存储

二、客户端源码解析

源码跟踪

下面是通过管理流创建schema的样例代码，核心就是通过PulsarAdmin.schemas获取schema对象，这个schema对象负责所有客户端跟schema相关的操作，包括schema的增删改查等。通过方法的第二个参数可以看到是通过Schema接口提供的静态方法AVRO来构造Avro格式的schema对象，除此之外Schema接口还提供了诸如JSON、KeyValue、PROTOBUF等静态方法提供对应数据格式的schema对象，这里如果将这块构造schema对象逻辑抽成简单工厂模式可能会更合适些
在这里插入图片描述

接下来就进入createSchema方法，顾名思义可以知道这个方法就是用于创建schema的，第一个参数是topic，第二个参数是SchemaInfo对象，这个对象包含了所有要新建的schema信息，这里会将它转换为PostSchemaPayload对象传递给下一个方法。PostSchemaPayload是用来请求到服务端的参数
在这里插入图片描述

这个方法并不会有返回值，sync方法是处理异步结果对象，它在正常写成功情况下不会做任何操作，但如果有什么错误会往外抛出异常。这里核心逻辑是在createSchemaAsync方法
在这里插入图片描述

可以看到这个方法的返回值是个异步对象，146行这里会获取当前topic对应的TopicName对象，并通过schemaPath方法构造WebTarget对象，这个对象中就包含着要请求的HTTP地址，主要是根据当前Topic的版本来决定请求服务端哪个版本的处理方法。除此之外还可以看到有通过Entity.json方法将PostSchemaPayload对象转换为HTTP请求的参数对象，转换逻辑是javax.ws.rs-api这个网络库封装的，就不进行跟踪了
在这里插入图片描述

这里就是客户端最后发送的地方，request方法中还会发送前的安全相关检查，async方法基本上就说明本次HTTP请求是异步的，而post方法也能看得出，这是一个POST类型的HTTP请求，再往后就是将请求发送出去了
在这里插入图片描述

不知是否有人好奇参数WebTarget长什么样子，通过通过调试可以看到值为

/admin/v2/schemas/public/test-namespace-jytixthzgatgirem/test-multi-version-schema-one/schema

此值仅供学习参考，具体这个值的构造逻辑如下
在这里插入图片描述

小结

简单归纳如下

  • 通过Schema接口构造对应数据格式的schema对象，由此对象可得到schema相关的元信息SchemaInfo
  • 构造请求目标的HTTP地址
  • 通过javax.ws.rs-api提供的库发送异步HTTP请求到服务端

三、服务端源码解析

源码跟踪

服务端的接收逻辑在SchemasResource类，这个类在org.apache.pulsar.broker.admin包下，这个包下全是处理管理流相关的操作，如果有做pulsar平台化需求的，这个包下的相关逻辑值得一读。

再来看看postSchema方法，首先是validateTopicName方法，这个方法就是对入参进行判空、是否有特殊字符做检查；接下来就是核心方法postSchemaAsync，通过方法名可以推断出这是个异步处理schema写请求的方法
在这里插入图片描述

postSchemaAsync方法看似复杂，实际上核心的就是133行，其余的方法大概说一下，validateOwnershipAndOperationAsync方法主要检查当前用户是否有新建schema的操作权限，getSchemaCompatibilityStrategyAsyncWithoutAuth方法相对复杂一些，放到后面详细讲解。那么再看回133行，其中getSchemaRegistryService方法获取的是SchemaRegistryServiceImpl对象，顾名思义可以知道Pulsar的SchemaResistry相关的功能都是由它进行处理，现在先看它的putSchemaIfAbsent方法
在这里插入图片描述

SchemaStorage对象是SchemaRegistryServiceImpl的核心成员，负责schema存储相关的操作。在新建schema时会调用它的put方法进行创建；这里有个trimDeletedSchemaAndGetList方法，如果put方法在创建schema时有任何异常，则此方法会去删除该新建的schema，避免写"一半"的情况发生，某种意义上这也是一种回滚的设计。
在这里插入图片描述

这里的getAll方法很重要，会根据schema的id来查询是否已经存在当前schema，有的话则将版本号加1。处理完之后就调用put方法
在这里插入图片描述

这里没什么逻辑，继续往下跟踪
在这里插入图片描述

getSchemaLocator方法会构造LocatorEntry对象，调用putSchema
在这里插入图片描述

由于是初次创建schema，因此直接走到337行；如果这个topic已经创建过schema则会读取之前的schema信息再新增，同时把版本号自增

在这里插入图片描述

在这里可以看得到构造IndexEntry对象，这是消息的索引对象，后续用来加速查询schema
在这里插入图片描述

这个方法的内容就很眼熟了(bookkeeper相关内容)，createLedger方法会先创建这个Ledger
在这里插入图片描述

在576行可以看到最终调用bookkeeper创建这个Ledger
在这里插入图片描述

再来看看addEntry方法，这里核心也是调用bookkeeper的ledgerHandle进行数据写入
在这里插入图片描述

这个方法是属于Bookkeeper客户端的逻辑了，通过方法注释可以看到，这个方法负责将数据异步写入到一个打开的Ledger。Bookkeeper相关的逻辑后续在单独写post进行讲解
在这里插入图片描述

小结

简单归纳如下

  1. 参数格式校验、操作权限校验
  2. 查询当前Topic是否已经创建过schema，有则以插入时版本号自增
  3. 如果是初次创建Schema，则调用bookkeeper创建Ledger
  4. 往这个Schema对应的Ledger内插入schema元数据信息

四、其他

序列化对象创建流程

现在再专门来看看序列化对象的创建过程，回到开头管理流创建schema的地方，Schema.AVRO方法是咱们本次要看的
在这里插入图片描述

通过注释可以看到，此静态方法是创建一个Avro类型的schema对象，getDefaultImplementation方法是获取实现类(饿汉单例设计模式)，而newAvroSchema方法才是本次要看的
在这里插入图片描述

继续往下跟踪
在这里插入图片描述

获取对应处理的类加载器，并通过对应的类加载器创建AvroSchema实例
在这里插入图片描述

54行是核心，其他的都是赋值操作
在这里插入图片描述

super调用父类构造函数做赋值操作，还是继续看
在这里插入图片描述

继续跟踪parse逻辑
在这里插入图片描述

FACTORY.createParser方法是jackson的方法，用于创建JsonParser对象的；因此继续跟踪parse方法
在这里插入图片描述

1471行可以看到返回了我们想要的Schema对象，那么Schema.parse方法就是重中之重
在这里插入图片描述

这个方法是核心，本身会递归的进行解析赋值给schema对象
在这里插入图片描述

相信读者读到这里也好奇schema长什么样，因此提供下图让读者感受下，能大概推测得出来这里已经涵盖了schema的结构信息了
在这里插入图片描述

getSchemaCompatibilityStrategyAsyncWithoutAuth方法

AdminResource#getSchemaCompatibilityStrategyAsyncWithoutAuth方法是在服务端处理schema创建请求阶段会调用的方法，现在就一起跟踪看看

731行和739行分别是获取Topic级别和Namespace级别的schema兼容策略，如果没有定义则默认自动更新。例如Topic A之前已经创建过schema1，那么如果此时再发起schema2创建请求，则服务端会继续保存并且生效schema2，只不过它的版本号会进行累加，当然，也可以配置为不支持schema策略不支持更新，一旦确定了后就不允许再变更
在这里插入图片描述

五、总结

相信大家对schema创建的流程已经很清楚了，再次简单归纳下

  1. 客户端根据用户定义的结构信息创建对应的Schema对象，并将结构信息以HTTP请求发给服务端
  2. 服务端检测并根据Schema兼容策略做相对应的处理，一般情况下会调用Bookkeeper创建Ledger以及Entry
  3. Bookkeeper将此Schema数据持久化到磁盘，相当于Schema信息会被Bookkeeper当作一条消息进行存储

这基本上就是全部内容，当然细节感兴趣的小伙伴可以自行跟踪代码，相信你会有更多收获～

相关文章：

管理流创建schema流程源码解析

一、简析 schema是pulsar重要的功能之一，现在就一起从源码的视角看下管理流创建schema时客户端和服务端的表现 客户端 客户端主要经历以下四个步骤 创建Schema实例 根据数据类型创建相对应的实例，例如Avro创建AvroSchema、JSON创建JSONSchema等 获取…...

编程日记 2024/8/2 3:26:19

【iOS】iOS内存五大分区

iOS内存五大分区 总揽 iOS中，内存主要分为五大区域：栈区，堆区，全局区/静态区，常量区和代码区。总览图如下。 这个图我觉得更好记，因为下面是低地址，上面是高地址，是比较符合日常…...

编程日记 2024/8/2 3:25:18

【项目实战】—— 高并发内存池

文章目录 什么是高并发内存池？项目介绍一、项目背景二、项目目标三、核心组件四、关键技术五、应用场景六、项目优势 什么是高并发内存池？ 高并发内存池是一种专门设计用于高并发环境下的内存管理机制。它的原型是Google的一个开源项目tcmalloc&#xff…...

编程日记 2024/8/2 3:24:17

二叉搜索树的第 k 大的节点

题目描述 给定一棵二叉搜索树，请找出其中第 k 大的节点。 解题基本知识 二叉搜索树（Binary Search Tree）又名二叉查找树、二叉排序树。它是一棵空树，或者是具有下列性质的二叉树： 若它的左子树不空，则左子…...

编程日记 2024/8/2 3:23:16

利用langchain 做大模型 Few-shot Learning 提示，包括固定和向量相似的动态样本筛选

文章目录 few-shotFixed Examples 固定样本Dynamic few-shot prompting 动态样本提示辅助参考资料 few-shot 相比大模型微调，在有些情况下，我们更想使用 Few-shot Learning 通过给模型喂相关样本示例，让模型能够提升相应任务的能力。 固定样…...

编程日记 2024/8/2 3:22:14

基于python的百度迁徙迁入、迁出数据分析（五）

终于在第五篇文章我们进入了这个系列的正题：数据分析 这里我选择上海2024年5月1日——5月5日的迁入、迁出数据作为分析的基础，首先选择节假日的数据作为分析的原因呢，主要是节假日人们出行目的比较单一（出游、探亲）&a…...

编程日记 2024/8/2 3:21:12

SpringBoot 如何处理跨域请求

SpringBoot 处理跨域请求，通常是通过配置全局的 CORS（跨源资源共享）策略来实现的。CORS 是一种机制，它使用额外的 HTTP 头部来告诉浏览器，让运行在一个 origin (domain) 上的 web 应用被准许访问来自不同源服务器上的指…...

编程日记 2024/8/2 3:20:09

大数据技术基础编程、实验和案例----大数据课程综合实验案例

一、实验目的 (1）熟悉Linux系统、MySQL、Hadoop、HBase、Hive、Sqoop、R、Eclipse等系统和软件的安装和使用； (2）了解大数据处理的基本流程； (3）熟悉数据预处理方法； (4）熟悉在不同类型数据库之…...

编程日记 2024/8/2 3:17:05

微信小程序-获取手机号：HttpClientErrorException: 412 Precondition Failed: [no body]

问题： 412 异常就是你的请求参数获取请求头与服务器的不符，缺少请求体！ 我的问题： 我这里获取微信手机号的时候突然给我报错142，但是代码用的是原来的代码，换了一个框架就噶了！ 排查问题&am…...

编程日记 2024/8/2 3:16:04

大数据核心概念与技术架构简介

大数据基本概念 大数据是指无法在一定时间范围内用常规软件工具进行捕捉、管理和处理的数据集合，是具有更强的决策力、洞察发现力和流程优化能力的海量、高增长率和多样化的信息资产。 大数据特征： 数据量大：一般以P（1000个TB&a…...

编程日记 2024/8/2 3:15:03

快排 谁在中间

原题 Whos in the Middle FJ is surveying his herd to find the most average cow. He wants to know how much milk this median cow gives: half of the cows give as much or more than the median; half give as much or less. FJ正在调查他的牛群，以找到最…...

编程日记 2024/8/2 3:14:02

ORA-00911: invalid character

场景： 调用接口查询oracle的数据库数据时报错ORA-00911: invalid character，但是sql语句没有问题放在navicat控制台中运行也没有问题，但是代码中跑就会报无效字符集 分析： 代码中Oracle的语法解析器比较严格，比如句…...

编程日记 2024/8/2 3:12:00

Pytorch实现线性回归Linear Regression

借助 PyTorch 实现深度神经网络 - 线性回归 - 第 2 周 | Coursera 线性回归预测 用PyTorch实现线性回归模块 创建自定义模块（内含一个线性回归） 训练线性回归模型 对于线性回归，特定类型的噪声是高斯噪声 平均损失均方误差函数&#xff1a…...

编程日记 2024/8/2 3:10:59

十八次（虚拟主机与vue项目、samba磁盘映射、nfs共享）

1、虚拟主机搭建环境准备 将原有的nginx.conf文件备份 [rootserver ~]# cp /usr/local/nginx/conf/nginx.conf /usr/local/nginx/conf/nginx.conf.bak[rootserver ~]# grep -Ev "#|^$" /usr/local/nginx/conf/nginx.conf[rootserver ~]# grep -Ev "#|^$"…...

编程日记 2024/8/2 3:08:57

P1340 兽径管理 题解|最小生成树

题目大意 洛谷中链接 推荐文章：并查集入门 原文 约翰农场的牛群希望能够在 N N N 个草地之间任意移动。草地的编号由 1 1 1 到 N N N。草地之间有树林隔开。牛群希望能够选择草地间的路径，使牛群能够从任一 片草地移动到任一片其它草地。 牛群可在…...

编程日记 2024/8/2 3:02:52

Python，Maskrcnn训练，cannot import name ‘saving‘ from ‘keras.engine‘ ，等问题集合

Python版本3.9&#xff0c;tensorflow2.11.0&#xff0c;keras2.11.0 问题一、module keras.engine has no attribute Layer Traceback (most recent call last):File "C:\Users\Administrator\Desktop\20240801\代码\test.py", line 16, in <module>from mrc…...

编程日记 2024/8/2 3:00:50

Linux常用工具

文章目录 tar打包命令详解unzip命令&#xff1a;解压zip文件vim操作详解netstat详解df命令详解ps命令详解find命令详解 tar打包命令详解 tar命令做打包操作 当 tar 命令用于打包操作时&#xff0c;该命令的基本格式为&#xff1a; tar [选项] 源文件或目录此命令常用的选项及…...

编程日记 2024/8/2 2:59:49

AI未来的发展如何

AI&#xff08;人工智能&#xff09;的发展前景非常广阔&#xff0c;随着技术的不断进步和应用场景的不断拓展&#xff0c;AI将在多个领域发挥重要作用。以下是对AI发展前景的详细分析&#xff1a; 一、技术突破与创新 生成式AI的兴起&#xff1a;以ChatGPT为代表的生成式AI技…...

编程日记 2024/8/2 2:58:48

若依替换首页上的logo

...

编程日记 2024/8/2 2:54:43

sed的使用示例

场景:使用sed将多个空格变成单空格,再使用cut来切分得到需要的结果 得到后面这个文件名: ls ./ drwxr-x— 2 root root 6 Jul 18 9:00 7b40f1412d83c1524af7977593607f15 drwxr-x— 2 root root 6 Jul 18 14:00 50af29cef2c65a9d28905a3ce831bcb7 drwxr-x— 2 root root 6 Jul…...

编程日记 2024/8/2 2:53:42

Python｜GIF 解析与构建（5）：手搓截屏和帧率控制

目录 Python&#xff5c;GIF 解析与构建&#xff08;5&#xff09;&#xff1a;手搓截屏和帧率控制 一、引言 二、技术实现&#xff1a;手搓截屏模块 2.1 核心原理 2.2 代码解析&#xff1a;ScreenshotData类 2.2.1 截图函数&#xff1a;capture_screen 三、技术实现&…...

编程新知 2025/6/27 2:57:24

Mybatis逆向工程，动态创建实体类、条件扩展类、Mapper接口、Mapper.xml映射文件

今天呢&#xff0c;博主的学习进度也是步入了Java Mybatis 框架&#xff0c;目前正在逐步杨帆旗航。 那么接下来就给大家出一期有关 Mybatis 逆向工程的教学&#xff0c;希望能对大家有所帮助&#xff0c;也特别欢迎大家指点不足之处&#xff0c;小生很乐意接受正确的建议&…...

编程新知 2025/7/1 5:43:01

微信小程序 - 手机震动

一、界面 <button type"primary" bindtap"shortVibrate">短震动</button> <button type"primary" bindtap"longVibrate">长震动</button> 二、js逻辑代码 注&#xff1a;文档 https://developers.weixin.qq…...

编程新知 2025/6/26 10:12:53

【Web 进阶篇】优雅的接口设计：统一响应、全局异常处理与参数校验

系列回顾&#xff1a; 在上一篇中&#xff0c;我们成功地为应用集成了数据库&#xff0c;并使用 Spring Data JPA 实现了基本的 CRUD API。我们的应用现在能“记忆”数据了&#xff01;但是&#xff0c;如果你仔细审视那些 API&#xff0c;会发现它们还很“粗糙”&#xff1a;有…...

编程新知 2025/6/21 12:51:57

从零实现STL哈希容器：unordered_map/unordered_set封装详解

本篇文章是对C学习的STL哈希容器自主实现部分的学习分享 希望也能为你带来些帮助~ 那咱们废话不多说&#xff0c;直接开始吧&#xff01; 一、源码结构分析 1. SGISTL30实现剖析 // hash_set核心结构 template <class Value, class HashFcn, ...> class hash_set {ty…...

编程新知 2025/6/25 23:57:55

【Java_EE】Spring MVC

目录 Spring Web MVC ​编辑注解 RestController RequestMapping RequestParam RequestParam RequestBody PathVariable RequestPart 参数传递 注意事项 ​编辑参数重命名 RequestParam ​编辑​编辑传递集合 RequestParam 传递JSON数据 ​编辑RequestBody ​…...

编程新知 2025/6/23 7:20:00

HDFS分布式存储 zookeeper

hadoop介绍 狭义上hadoop是指apache的一款开源软件 用java语言实现开源框架&#xff0c;允许使用简单的变成模型跨计算机对大型集群进行分布式处理&#xff08;1.海量的数据存储 2.海量数据的计算&#xff09;Hadoop核心组件 hdfs&#xff08;分布式文件存储系统&#xff09;&a…...

编程新知 2025/6/26 0:04:04

七、数据库的完整性

七、数据库的完整性 主要内容 7.1 数据库的完整性概述 7.2 实体完整性 7.3 参照完整性 7.4 用户定义的完整性 7.5 触发器 7.6 SQL Server中数据库完整性的实现 7.7 小结 7.1 数据库的完整性概述 数据库完整性的含义 正确性 指数据的合法性 有效性 指数据是否属于所定…...

编程新知 2025/6/28 14:04:35

AirSim/Cosys-AirSim 游戏开发（四）外部固定位置监控相机

这个博客介绍了如何通过 settings.json 文件添加一个无人机外的 固定位置监控相机&#xff0c;因为在使用过程中发现 Airsim 对外部监控相机的描述模糊&#xff0c;而 Cosys-Airsim 在官方文档中没有提供外部监控相机设置&#xff0c;最后在源码示例中找到了&#xff0c;所以感…...

编程新知 2025/7/1 4:03:24

TSN交换机正在重构工业网络，PROFINET和EtherCAT会被取代吗？

在工业自动化持续演进的今天&#xff0c;通信网络的角色正变得愈发关键。 2025年6月6日&#xff0c;为期三天的华南国际工业博览会在深圳国际会展中心&#xff08;宝安&#xff09;圆满落幕。作为国内工业通信领域的技术型企业&#xff0c;光路科技&#xff08;Fiberroad&…...

编程新知 2025/6/20 21:57:36