消息队列后端流处理【免费下载链接】pulsarApache Pulsar - distributed pub-sub messaging system项目地址https://gitcode.com/gh_mirrors/pulsar28/pulsar点击查看免费下载本指南以 Apache Pulsar 管理 API 中包管理Package Management为主题系统讲解包Package的概念、命名与元数据模型并完整演示如何通过pulsar-admin命令行、REST API 与 Java Admin API 三种方式上传、下载、删除包、读写包元数据、列出版本与按类型列举包。读完本文你将掌握 Pulsar Functions 与 Pulsar IO 连接器Source/Sink跨租户、跨命名空间复用的标准流程并理解底层包管理服务的实现机制。什么是包Package在 Apache Pulsar 中包是一组用户希望在后续操作中复用的元素的集合。它既可以是一个函数Function也可以是一个 Sink 或 Source 连接器具体由用户按需定义。包管理系统Package Management System负责以一致的方式自动管理包简化安装、升级与删除操作并把包作为包管理器处理的最小单元。在 Pulsar 中包按照租户tenant和命名空间namespace维度进行组织用于统一管理 Pulsar Functions 与 Pulsar IO 连接器。包的三要素类型、归属与版本创建包时需要提供以下五类信息| 组件 | 说明 | | - | - | | type | 指定受支持的包类型之一function、sink、source| | tenant | 指定希望创建包的租户 | | namespace | 指定希望创建包的命名空间 | | name | 指定包的完整名称格式为tenant/namespace/package name| | version | 以数字形式指定包的版本格式为MajorVersion.MinorVersion|这五类信息最终会拼接为一个包的完整 URL格式为type://tenant/namespace/package name/version在实际的包名表示中版本与包名之间使用分隔例如本文后续会反复出现的三个典型示例sink://public/default/mysql-sink1.0 function://my-tenant/my-ns/my-function0.1 source://my-tenant/my-ns/mysql-cdc-source2.3从源码实现来看包名在服务端被解析为五个组成部分type、tenant、namespace、name、version。以 PackageName.java 的实现为例包名必须包含://分隔符否则会被判定为非法包的完整名称不含版本必须严格由tenant/namespace/name三段组成若 URL 中省略了version部分系统会自动将版本视为latest即默认指向最新版本类型字符串会经过 PackageType 枚举校验目前只支持FUNCTION(function)、SINK(sink)、SOURCE(source)三种其他取值会抛出IllegalArgumentException。包的元数据包管理系统会存储每个包的数据与元数据并跟踪包的版本变化。包元数据的字段如下| 元数据 | 说明 | | -- | -- | | description | 包的描述信息 | | contact | 包的联系信息例如开发团队的邮箱地址 | | create_time | 包的创建时间 | | modification_time | 包的最后修改时间 | | properties | 用户自定义的键值对映射用于存储其他信息 |这些字段在 Java 侧的实体类 PackageMetadata.java 中一一对应description、contact、createTime、modificationTime与propertiesMapString, String。该类通过 Lombok 的Builder提供了链式构造能力这也是后续pulsar-admin与 Java Admin API 组装元数据的基础。包的使用流程跨命名空间复用包的核心价值在于让同一组函数与 IO 连接器可以被高效复用。例如你可以在多个命名空间中使用同一个函数、Source 或 Sink。主要步骤如下创建包在包管理器中创建包提供上述 type、tenant、namespace、name、version 信息生成形如type://tenant/namespace/packageNameversion的包 URL。上传元素将希望跨命名空间复用的函数、Source、Sink 文件上传到该包。应用权限为来自各个命名空间的调用方对该包授予相应权限。完成上述步骤后就可以通过包管理器以 URL 定位并调用包中定义的元素。Pulsar 中的包管理方式Pulsar 提供三种方式管理包资源命令行工具pulsar-admin、REST API 与 Java Admin API。三者在功能上一一对应可完成以下七类操作上传包upload下载包download删除包delete获取包元数据get metadata更新包元数据update metadata列出包的所有版本list versions列出某命名空间下指定类型的所有包list上传一个包使用 pulsar-admin 上传bin/pulsar-admin packages upload function://public/default/examplev0.1 --path package-file --description package-description使用 REST API 上传POST /admin/v3/packages/:type/:tenant/:namespace/:packageName/:version使用 Java Admin API 上传同步上传到包管理服务void upload(PackageMetadata metadata, String packageName, String path) throws PulsarAdminException;异步上传CompletableFutureVoid uploadAsync(PackageMetadata metadata, String packageName, String path);以pulsar-admin的 CmdPackages.java 实现为准upload子命令支持以下参数| 参数 | 是否必填 | 说明 | | - | - | - | | packageName | 是 | 包名格式type://tenant/namespace/packageNameversion| |--description| 是 | 包的描述 | |--contact| 否 | 包的联系信息 | |--properties/-P| 否 | 自定义键值对可重复指定多个 | |--path| 是 | 待上传的包文件路径 |其执行逻辑是先通过PackageMetadata.builder()组装元数据再调用Packages.upload(metadata, packageName, path)完成上传。注意上传时的--description是必填项这一点与原文档中的示例一致。下载一个包使用 pulsar-admin 下载bin/pulsar-admin packages download function://public/default/examplev0.1 --path package-file使用 REST API 下载GET /admin/v3/packages/:type/:tenant/:namespace/:packageName/:version使用 Java Admin API 下载同步从包管理服务下载void download(String packageName, String path) throws PulsarAdminException;异步下载CompletableFutureVoid downloadAsync(String packageName, String path);在 CmdPackages.java 中download同样通过--path指定下载目标路径并将包内容写入该路径。删除一个包使用 pulsar-admin 删除以下命令删除版本为 0.1 的包bin/pulsar-admin packages delete functions://public/default/examplev0.1使用 REST API 删除DELETE /admin/v3/packages/:type/:tenant/:namespace/:packageName/:version使用 Java Admin API 删除同步删除指定包void delete(String packageName) throws PulsarAdminException;异步删除CompletableFutureVoid deleteAsync(String packageName);获取包的元数据使用 pulsar-admin 获取bin/pulsar-admin packages get-metadata function://public/default/testv1使用 REST API 获取GET /admin/v3/packages/:type/:tenant/:namespace/:packageName/:version/metadata使用 Java Admin API 获取同步获取PackageMetadata getMetadata(String packageName) throws PulsarAdminException;异步获取CompletableFuturePackageMetadata getMetadataAsync(String packageName);从 Broker 侧的实现看Packages.java 是Path(/packages)下的 REST 端点getMeta方法接收type、tenant、namespace、packageName、version五个路径参数成功时返回PackageMetadata对象HTTP 200包不存在时返回 404包名非法时返回 412未启用包管理服务时返回 503。这一响应语义同样适用于其他 REST 端点。更新包的元数据使用 pulsar-admin 更新bin/pulsar-admin packages update-metadata function://public/default/examplev0.1 --description update-description使用 REST API 更新PUT /admin/v3/packages/:type/:tenant/:namespace/:packageName/:version/metadata使用 Java Admin API 更新同步更新void updateMetadata(String packageName, PackageMetadata metadata) throws PulsarAdminException;异步更新CompletableFutureVoid updateMetadataAsync(String packageName, PackageMetadata metadata);在 CmdPackages.java 中update-metadata要求--description为必填同时可选--contact与--properties最终通过PackageMetadata.builder()重建完整元数据并调用updateMetadata覆盖写入。注意它与upload不同update-metadata不会修改包的数据内容只更新元数据。列出包的所有版本使用 pulsar-admin 列出bin/pulsar-admin packages list-versions type://tenant/namespace/packageName使用 REST API 列出GET /admin/v3/packages/:type/:tenant/:namespace/:packageName使用 Java Admin API 列出同步列出ListString listPackageVersions(String packageName) throws PulsarAdminException;异步列出CompletableFutureListString listPackageVersionsAsync(String packageName);注意list-versions的包名参数不需要携带版本号只传type://tenant/namespace/packageName即可系统会返回该包的全部版本号列表。列出命名空间下指定类型的所有包使用 pulsar-admin 列出bin/pulsar-admin packages list --type function public/default使用 REST API 列出PUT /admin/v3/packages/:type/:tenant/:namespace使用 Java Admin API 列出同步列出ListString listPackages(String type, String namespace) throws PulsarAdminException;异步列出CompletableFutureListString listPackagesAsync(String type, String namespace);在 CmdPackages.java 中list子命令需要--type必填与命名空间位置参数必填格式为tenant/namespace。值得注意的是原文档给出的 REST 端点为PUT /admin/v3/packages/:type/:tenant/:namespace属于文档中的既有描述实际使用时请以当前版本生成的 OpenAPI 文档为准。底层实现与源码指引包管理功能在仓库中由多个模块协同实现你可以沿以下路径深入阅读Java Admin 客户端接口Packages.java 定义了全部七个操作含同步与异步共 14 个方法是pulsar-admin与 Java 程序共用的能力入口。命令行实现CmdPackages.java 注册了get-metadata、update-metadata、upload、download、list、list-versions、delete七个pulsar-admin packages子命令。包名解析PackageName.java 负责将type://tenant/namespace/nameversion拆解为结构化字段并对非法命名抛出异常同时内置了容量 10 万、30 分钟过期的本地缓存以提升解析性能。包类型定义PackageType.java 定义了function、sink、source三种受支持类型。元数据模型PackageMetadata.java 定义了 description、contact、createTime、modificationTime、properties 五个字段。Broker REST 端点Packages.java 是/admin/v3/packages路径下的服务端实现入口实际业务逻辑继承自PackagesBase。此外包管理服务本身位于 pulsar-package-management 模块包含核心抽象core、BookKeeper 存储实现bookkeeper-storage与文件系统存储实现filesystem-storage其元数据序列化与解析逻辑可参考 PackageMetadataUtil.java。总结包管理是 Pulsar 复用函数与 IO 连接器的基础设施通过type://tenant/namespace/packageNameversion的统一命名将函数、Source、Sink 及其元数据集中托管在租户与命名空间维度下并借助pulsar-admin、REST API 与 Java Admin API 完成上传、下载、删除、元数据读写与版本列举等全部管理操作。结合本仓库源码阅读上述关键类可以进一步理解包名校验、版本回退latest默认版本、元数据序列化等底层细节从而在生产环境中设计出规范、可审计的包管理流程。赞分享消息队列后端流处理【免费下载链接】pulsarApache Pulsar - distributed pub-sub messaging system项目地址https://gitcode.com/gh_mirrors/pulsar28/pulsar点击查看免费下载相关推荐Apache Pulsar 权限管理实战基于 admin-api-permissions 的命名空间与通配符授权完全指南Apache Pulsar 权限管理实战基于 admin api permissions 的命名空间与通配符授权完全指南 Apache Pulsar 提供细粒消息队列后端流处理Apache Pulsar命名空间管理多租户资源隔离与配额控制Apache Pulsar命名空间管理多租户资源隔离与配额控制 在分布式系统中多租户资源隔离是企业级消息队列的核心需求。Apache Pulsar作为云原生消息队列后端Scrcpy GUI多设备管理同时控制多台Android手机的完整指南Scrcpy GUI多设备管理同时控制多台Android手机的完整指南 Scrcpy GUI是一款简单而美观的应用程序专为scrcpy打造它允许用户通过电桌面应用移动开发开发工具创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考