消息队列后端流处理【免费下载链接】pulsarApache Pulsar - distributed pub-sub messaging system项目地址https://gitcode.com/gh_mirrors/pulsar28/pulsar点击查看免费下载Namespace命名空间是 Apache Pulsar 中连接租户Tenant与主题Topic的关键逻辑分组也是配额、复制、保留、限流等全部策略的生效单元。本篇基于 Apache Pulsar 2.3.0 管理文档与仓库源码系统讲解如何通过pulsar-admin命令行、/admin/v2/namespacesREST 接口和 JavaPulsarAdmin对象完成 Namespace 的创建、查询、删除以及全部策略配置帮助你直接在生产与测试环境中落地 Namespace 治理方案。Namespace 是什么Pulsar 的逻辑分组与策略承载单元在 Apache Pulsar 中Namespace 是对相关 Topic 的逻辑分组其完整命名层级为tenant/namespace。从概念上理解Tenant租户容量分配与认证/授权方案的管理单元Namespace归属于某个 Tenant 的主题分组Topic生产者与消费者传递消息的命名通道。除了分组功能Namespace 还是策略下发的最小粒度——跨地域复制集群、Backlog 配额、持久化参数、消息 TTL、保留策略、分发限流等配置都挂在 Namespace 上作用于其下所有主题。从源码结构看仓库中存在 v1 与 v2 两套管理实现pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v1/Namespaces.java 与 pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/Namespaces.java本文涉及的 REST 接口均为 v2 路径/admin/v2/namespaces。三大管理入口Namespace 可以通过以下三种方式管理pulsar-admin命令行工具使用namespaces子命令全部子命令定义在 pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdNamespaces.java命令描述为Operations about namespacesREST API访问/admin/v2/namespaces端点由 broker 侧的 v2 Namespaces.java 提供实现Java API调用PulsarAdmin对象的namespaces()方法方法签名定义在 pulsar-client-admin-api/src/main/java/org/apache/pulsar/client/admin/Namespaces.java。pulsar-admin中可用的 namespace 子命令完整列表见 reference-pulsar-admin.md包括create、policies、list、delete、set-clusters、set-backlog-quota、set-persistence、set-message-ttl、set-retention、split-bundle、set-dispatch-rate、clear-backlog、unload等。Namespace 基础资源操作创建 Namespace在指定 Tenant 下创建新的 Namespace$ pulsar-admin namespaces create test-tenant/test-namespaceREST 方式为PUT /admin/v2/namespaces/:tenant/:namespace对应 broker 源码中PUT Path(/{tenant}/{namespace})的createNamespace操作。Java 方式admin.namespaces().createNamespace(namespace);从 CmdNamespaces.java 的Create命令实现可以看到创建时还支持两个可选参数--clusters, -c指定该 Namespace 归属的集群列表--bundles, -b指定要激活的 bundle 数量其取值范围被校验为(0, 2^32]超出会抛出Invalid number of bundles参数异常。对于 v2 命名空间tenant/namespace形式create会直接把上述参数组装成Policies对象一次调用createNamespace(namespace, policies)完成创建与初始策略设置。查看 Namespace 当前策略随时可以获取 Namespace 当前绑定的完整策略$ pulsar-admin namespaces policies test-tenant/test-namespace { auth_policies: { namespace_auth: {}, destination_auth: {} }, replication_clusters: [], bundles_activated: true, bundles: { boundaries: [ 0x00000000, 0xffffffff ], numBundles: 1 }, backlog_quota_map: {}, persistence: null, latency_stats_sample_rate: {}, message_ttl_in_seconds: 0, retention_policies: null, deleted: false }响应字段直观对应上文提到的各类策略replication_clusters复制集群、bundlesbundle 边界与数量、backlog_quota_mapBacklog 配额、persistence持久化、message_ttl_in_seconds消息 TTL、retention_policies保留策略等。REST 方式为GET /admin/v2/namespaces/:tenant/:namespaceJava 方式为admin.namespaces().getPolicies(namespace)。列出 Tenant 下的所有 Namespace$ pulsar-admin namespaces list test-tenant test-tenant/ns1 test-tenant/ns2RESTGET /admin/v2/namespaces/:tenant对应getTenantNamespacesJavaadmin.namespaces().getNamespaces(tenant);删除 Namespace$ pulsar-admin namespaces delete test-tenant/ns1RESTDELETE /admin/v2/namespaces/:tenant/:namespaceJavaadmin.namespaces().deleteNamespace(namespace);注意delete子命令还支持-f, --force参数用于强制删除 Namespace 及其下所有主题见Delete命令源码deleteNamespace(namespace, force)。配置跨地域复制集群Pulsar 支持将消息从一个集群colo内部复制到另一个集群这就是 geo-replication。通过设置 Namespace 的复制集群来启用$ pulsar-admin namespaces set-clusters test-tenant/ns1 \ --clusters cl1RESTPOST /admin/v2/namespaces/:tenant/:namespace/replicationbroker 源码中POST Path(/{tenant}/{namespace}/replication)其ApiResponse标注了 409/412 两种失败场景peer 集群不能同时出现在复制集群列表中、Namespace 必须为 global 或集群 ID 无效Javaadmin.namespaces().setNamespaceReplicationClusters(namespace, clusters);查询已配置的复制集群$ pulsar-admin namespaces get-clusters test-tenant/cl1/ns1 cl2RESTGET /admin/v2/namespaces/:tenant/:namespace/replicationJavaadmin.namespaces().getNamespaceReplicationClusters(namespace)从 CmdNamespaces.java 的SetReplicationClusters实现可以看到--clusters为必填项支持逗号分隔的多个集群 ID。Backlog 配额管理Backlog 配额用于限制 Namespace 的存储/带宽资源避免消息积压失控。一旦达到阈值管理员可以预先设定以下三种处置策略之一producer_request_holdbroker 暂停接收并持久化生产者的请求负载producer_exceptionbroker 通过抛出异常断开与生产者的连接consumer_backlog_evictionbroker 开始丢弃 backlog 中的消息。Backlog 配额限制通过指定backlog-quota-type来界定默认类型为destination_storage按字节大小限制。设置 Backlog 配额$ pulsar-admin namespaces set-backlog-quota --limit 10 --policy producer_request_hold test-tenant/ns1RESTPOST /admin/v2/namespaces/:tenant/:namespace/backlogQuotaJavaadmin.namespaces().setBacklogQuota(namespace, new BacklogQuota(limit, policy))从源码看SetBacklogQuota命令该命令还支持更多参数-l, --limit大小限制如10M、16G-lt, --limitTime时间限制秒非正数表示禁用时间限制-p, --policy达到限制后的处置策略可选producer_request_hold、producer_exception、consumer_backlog_eviction-t, --type配额类型可选destination_storage按字节与message_age按消息时间戳两者可组合使用。查看 Backlog 配额$ pulsar-admin namespaces get-backlog-quotas test-tenant/ns1 { destination_storage: { limit: 10, policy: producer_request_hold } }RESTGET /admin/v2/namespaces/:tenant/:namespace/backlogQuotaMapJavaadmin.namespaces().getBacklogQuotaMap(namespace)。移除 Backlog 配额$ pulsar-admin namespaces remove-backlog-quota test-tenant/ns1RESTDELETE /admin/v2/namespaces/:tenant/:namespace/backlogQuotaJavaadmin.namespaces().removeBacklogQuota(namespace, backlogQuotaType)。持久化策略配置持久化策略用于配置 Namespace 下所有主题消息的持久化级别关键参数如下参数含义文档默认值命令行实际默认值源码--bookkeeper-ack-quorum每条 entry 需要等待的 ack 数保证的副本数02--bookkeeper-ensemble一个主题使用的 bookie 数量02--bookkeeper-write-quorum每条 entry 写入的次数02--ml-mark-delete-max-ratemark-delete 操作的节流速率0 表示不限速0.00说明文档中标注的默认值为 0而从 CmdNamespaces.java 的SetPersistence实现看命令行参数实际默认值为ensemble2, writeQuorum2, ackQuorum2, markDeleteRate0且三者必须大于 0、mark-delete 速率不能为负数否则抛出参数异常。设置持久化策略$ pulsar-admin namespaces set-persistence --bookkeeper-ack-quorum 2 --bookkeeper-ensemble 3 --bookkeeper-write-quorum 2 --ml-mark-delete-max-rate 0 test-tenant/ns1RESTPOST /admin/v2/namespaces/:tenant/:namespace/persistenceJavaadmin.namespaces().setPersistence(namespace,new PersistencePolicies(bookkeeperEnsemble, bookkeeperWriteQuorum,bookkeeperAckQuorum,managedLedgerMaxMarkDeleteRate))查看持久化策略$ pulsar-admin namespaces get-persistence test-tenant/ns1 { bookkeeperEnsemble: 3, bookkeeperWriteQuorum: 2, bookkeeperAckQuorum: 2, managedLedgerMaxMarkDeleteRate: 0 }RESTGET /admin/v2/namespaces/:tenant/:namespace/persistenceJavaadmin.namespaces().getPersistence(namespace)。Namespace Bundle 与负载均衡卸载与拆分什么是 Namespace BundleNamespace bundle是同一 Namespace 下主题的虚拟分组用一个 32 位哈希区间定义例如0x00000000到0xffffffff。每个 bundle 只能由一个 broker 服务。当某个 broker 上负载的 bundle 过多时可以把过重的 bundle卸载unload交给负载更低的 broker 接管当 bundle 中活跃主题过多导致单一 broker 压力过大时可以拆分splitbundle 来分散负载。卸载 Namespace Bundle$ pulsar-admin namespaces unload --bundle 0x00000000_0xffffffff test-tenant/ns1RESTPUT /admin/v2/namespaces/:tenant/:namespace/{bundle}/unloadJavaadmin.namespaces().unloadNamespaceBundle(namespace, bundle)从Unload命令源码可以看出--bundle为可选参数格式{start-boundary}_{end-boundary}不传 bundle 时卸载整个 Namespace调用unload(namespace)传了则按 bundle 卸载。拆分 Namespace Bundle$ pulsar-admin namespaces split-bundle --bundle 0x00000000_0xffffffff test-tenant/ns1RESTPUT /admin/v2/namespaces/:tenant/:namespace/{bundle}/splitJavaadmin.namespaces().splitNamespaceBundle(namespace, bundle)SplitBundle命令还支持更多精细化参数源码中均有定义--bundle-type, -bt按类型拆分与--bundle互斥--unload, -u拆分后是否立即卸载新生成的 bundles--split-algorithm-name, -san拆分算法可选range_equally_divide按哈希区间均分与topic_count_equally_divide按主题数均分缺省时使用 broker 端配置。消息 TTL 配置TTLTime To Live配置消息在未消费状态下的存活时长单位秒达到 TTL 后未消费的消息可被清除$ pulsar-admin namespaces set-message-ttl --messageTTL 100 test-tenant/ns1RESTPOST /admin/v2/namespaces/:tenant/:namespace/messageTTLbroker 源码标注可能返回 404“租户/集群/Namespace 不存在”与 412“无效 TTL”Javaadmin.namespaces().setNamespaceMessageTTL(namespace, messageTTL)查询当前 TTL$ pulsar-admin namespaces get-message-ttl test-tenant/ns1 100RESTGET /admin/v2/namespaces/:tenant/:namespace/messageTTLJavaadmin.namespaces().getNamespaceMessageTTL(namespace)。另外set-message-ttl支持-ttl简写值为 0 时表示禁用 TTL。清理 Backlog清理整个 Namespace 的 Backlog清理指定 Namespace 下所有主题的全部消息积压也可以只清理某个订阅subscription的积压$ pulsar-admin namespaces clear-backlog --sub my-subscription test-tenant/ns1RESTPOST /admin/v2/namespaces/:tenant/:namespace/clearBacklog针对订阅即clearNamespaceBacklogForSubscriptionJavaadmin.namespaces().clearNamespaceBacklogForSubscription(namespace, subscription)清理指定 Bundle 的 Backlog清理指定 NamespaceBundle 下所有主题的积压同样可限定订阅$ pulsar-admin namespaces clear-backlog --bundle 0x00000000_0xffffffff --sub my-subscription test-tenant/ns1RESTPOST /admin/v2/namespaces/:tenant/:namespace/{bundle}/clearBacklogJavaadmin.namespaces().clearNamespaceBundleBacklogForSubscription(namespace, bundle, subscription)从ClearBacklog命令源码CmdNamespaces.java可以看到其完整逻辑--sub与--bundle可组合使用分别路由到clearNamespaceBundleBacklogForSubscription、clearNamespaceBacklogForSubscription、clearNamespaceBundleBacklog、clearNamespaceBacklog四种调用同时支持--force参数跳过“确认清除 backlog”的交互提示。消息保留策略 Retention每个 Namespace 包含多个主题为避免主题存储无限制增长可以同时按大小与时长设定保留上限$ pulsar-admin set-retention --size 10 --time 100 test-tenant/ns1RESTPOST /admin/v2/namespaces/:tenant/:namespace/retentionJavaadmin.namespaces().setRetention(namespace, new RetentionPolicies(retentionTimeInMin, retentionSizeInMB))查看保留策略$ pulsar-admin namespaces get-retention test-tenant/ns1 { retentionTimeInMinutes: 10, retentionSizeInMB: 100 }RESTGET /admin/v2/namespaces/:tenant/:namespace/retentionJavaadmin.namespaces().getRetention(namespace)。需要特别说明的是set-retention的参数语义见SetRetention源码--time, -t保留时长可写为100m、3h、2d、5w等单位内部通过RelativeTimeUtil解析为秒再折算为分钟0表示不保留-1表示无限期保留--size, -s保留大小上限如10M、16G、3T0或小于 1MB 表示不按大小保留-1表示大小无限。对应 REST 响应中的retentionTimeInMinutes与retentionSizeInMB字段。分发限流 Dispatch Throttling为 Namespace 下所有主题设置消息分发速率可从每秒消息数msg-dispatch-rate与每秒字节数byte-dispatch-rate两个维度限制。速率周期通过dispatch-rate-period配置单位秒msg-dispatch-rate与byte-dispatch-rate的默认值为-1表示不启用限流。$ pulsar-admin namespaces set-dispatch-rate test-tenant/ns1 \ --msg-dispatch-rate 1000 \ --byte-dispatch-rate 1048576 \ --dispatch-rate-period 1RESTPOST /admin/v2/namespaces/:tenant/:namespace/dispatchRateJavaadmin.namespaces().setDispatchRate(namespace, 1000, 1048576, 1)查看已配置的分发速率$ pulsar-admin namespaces get-dispatch-rate test-tenant/ns1 { dispatchThrottlingRatePerTopicInMsg : 1000, dispatchThrottlingRatePerTopicInByte : 1048576, ratePeriodInSecond : 1 }RESTGET /admin/v2/namespaces/:tenant/:namespace/dispatchRateJavaadmin.namespaces().getDispatchRate(namespace)。从 CmdNamespaces.java 的SetDispatchRate实现看命令还支持--relative-to-publish-rate, -rp参数开启后 broker 按“发布速率 分发速率”之和进行节流且该命令支持-md、-bd、-dt简写。从 Broker 卸载 Namespace可以将整个 Namespace或其中某个 bundle从当前负责服务的 broker 上卸载$ pulsar-admin namespaces unload my-tenant/my-nsRESTPUT /admin/v2/namespaces/:tenant/:namespace/unloadJavaadmin.namespaces().unload(namespace)这与上文“卸载 Namespace Bundle”本质相同——不带--bundle参数时卸载整个 Namespace带参数则只卸载指定 bundle。Namespace Isolation原文档中Namespace isolationNamespace 隔离一节标注为Coming soon即将推出。从仓库现状看此功能在 2.3.0 版本文档中尚未给出具体操作说明实际使用时请以当前版本 reference-pulsar-admin.md 中列出的子命令为准。小结与最佳实践建议综合本文涉及的十余类 Namespace 操作可以沉淀出如下实践建议先规划后创建创建 Namespace 时通过--clusters一次性指定归属集群、通过--bundles规划 bundle 数量合法区间为(0, 2^32]减少后续变更策略组合使用Backlog 配额producer_request_hold/producer_exception/consumer_backlog_eviction三选一、TTL、Retention 三者协同分别解决积压失控、未消费消息过期、已消费消息保留的问题负载均衡三件套get-clusters/unload/split-bundle配合使用在 broker 过载时先卸载重 bundle再按需拆分以彻底分散热点限流分级set-dispatch-rate的-1默认值意味着“不启用”设置前务必确认业务流量模型--relative-to-publish-rate可用于压制突发发布变更留痕policies子命令返回的 JSON 是排查问题的第一入口任何策略变更前后都建议先执行一次快照对比。所有上述操作均可在 conf/standalone.conf 与 conf/broker.conf 提供的单机/集群环境中直接验证命令的完整参数语义可在 CmdNamespaces.java 中逐条核对REST 端点定义可在 v2/Namespaces.java 中溯源。赞分享消息队列后端流处理【免费下载链接】pulsarApache Pulsar - distributed pub-sub messaging system项目地址https://gitcode.com/gh_mirrors/pulsar28/pulsar点击查看免费下载相关推荐Apache Pulsar Namespace 管理实战pulsar-admin、REST API 与 Java Admin API 全指南Apache Pulsar Namespace 管理实战pulsar admin、REST API 与 Java Admin API 全指南 导读 命名空间消息队列后端流处理Apache Pulsar Namespace 管理实战pulsar-admin、REST API 与 Java Admin API 全面指南Apache Pulsar Namespace 管理实战pulsar admin、REST API 与 Java Admin API 全面指南 本指南以 Ap消息队列后端流处理Apache Pulsar Namespace 管理完全指南pulsar-admin / REST API / Java Admin API 三合一实战Apache Pulsar Namespace 管理完全指南pulsar admin / REST API / Java Admin API 三合一实战 Na消息队列后端流处理创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考