消息队列后端流处理【免费下载链接】pulsarApache Pulsar - distributed pub-sub messaging system项目地址https://gitcode.com/gh_mirrors/pulsar28/pulsar点击查看免费下载本文基于 Apache Pulsar 2.3.2 版本管理文档结合仓库源码系统讲解 Pulsar 集群Cluster的创建、初始化、查询、更新、删除、列举以及 peer-cluster 配置等完整生命周期操作。你将掌握通过pulsar-admin命令行、/admin/v2/clustersREST 接口和PulsarAdminJava 客户端三种方式管理集群的完整技能并理解底层元数据初始化原理。理解 Pulsar 集群的基本构成在进入具体操作之前先明确 Pulsar 集群由哪些组件组成。一个 Pulsar 集群 包含一个或多个 Pulsar broker负责消息的生产与消费服务一个或多个 BookKeeper 服务器又称 bookie负责消息数据的持久化存储一个 ZooKeeper 集群负责提供配置和协调管理即元数据存储详见元数据存储。在一个 Pulsar 实例Instance中可以包含多个集群Cluster。所有集群共享同一个全局 ZooKeeper即配置存储 Configuration Store而每个集群又有自己独立的本地 ZooKeeper。Pulsar 提供了三种管理集群的途径三者功能等价管理方式说明pulsar-admin工具的clusters子命令命令行方式适合脚本化运维/admin/v2/clustersREST API 端点HTTP 方式适合与外部系统集成Java API 中PulsarAdmin对象的clusters()方法详见 Java API 客户端编程方式适合嵌入应用代码在仓库中这三种方式的对应实现为CLIpulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdClusters.java其CmdClusters类注册了get、create、update、delete、list、update-peer-clusters、get-peer-clusters以及 failure-domain 相关子命令RESTpulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/ClustersBase.java定义了所有集群 REST 端点的服务端实现Java 客户端接口pulsar-client-admin-api/src/main/java/org/apache/pulsar/client/admin/Clusters.java定义了同步与异步*Async两套方法。创建集群Provision新集群可以通过管理接口创建。请注意该操作需要超级用户superuser权限。方式一pulsar-admin CLI使用create子命令创建集群$ pulsar-admin clusters create cluster-1 \ --url http://my-cluster.org.com:8080 \ --broker-url pulsar://my-cluster.org.com:6650这里--url指定 HTTP 管理服务地址默认端口 8080--broker-url指定 broker 服务地址默认端口 6650。从CmdClusters.java的ClusterDetailsCommand类源码可以看到create/update子命令实际支持的完整参数远不止这两个包括参数说明--urlservice-urlHTTP REST 管理服务地址--url-secure启用 TLS 时的安全管理服务地址HTTPS--broker-urlbroker-service-url供生产/消费使用的普通协议地址--broker-url-secure启用 TLS 时的 broker 安全地址pulsarssl--proxy-url客户端希望通过 Proxy 连接 broker 时的代理服务地址--auth-plugin/--auth-parameters客户端连接集群时使用的认证插件及参数--proxy-protocol代理路由协议类型例如SNI--tls-enable、--tls-allow-insecure、--tls-enable-keystore、--tls-trust-store-type、--tls-trust-store、--tls-trust-store-pwd、--tls-trust-certs-filepath集群内部客户端broker 之间通信的 TLS 相关配置--listener-name客户端连接集群时使用的 listener 名称--cluster-config-file指向一个 YAML 配置文件一次性指定集群的全部配置项尤其值得注意的是--cluster-config-file源码中processArguments()会先检查该参数若提供则优先从 YAML 文件加载配置CmdUtils.loadConfig再以命令行参数覆盖其中个别字段方便将集群配置纳入版本管理。方式二REST APIPUT /admin/v2/clusters/:cluster请求体为集群配置 JSON例如{ serviceUrl: http://pulsar.example.com:8080, brokerServiceUrl: pulsar://pulsar.example.com:6651 }从ClustersBase.createCluster()实现pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/ClustersBase.java可以看到服务端会依次校验超级用户权限、策略是否只读、集群名合法性NamedEntity.checkName集群名不能包含/字符、集群是否已存在已存在则返回 409 Conflict。成功创建返回 204。方式三Java Admin APIClusterData clusterData new ClusterData( serviceUrl, serviceUrlTls, brokerServiceUrl, brokerServiceUrlTls ); admin.clusters().createCluster(clusterName, clusterData);在仓库中ClusterData由pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/ClusterDataImpl.java实现完整字段包括serviceUrl、serviceUrlTls、brokerServiceUrl、brokerServiceUrlTls、proxyServiceUrl、authenticationPlugin、authenticationParameters、proxyProtocol、peerClusterNames、brokerClientTlsEnabled、tlsAllowInsecureConnection、brokerClientTlsEnabledWithKeyStore、brokerClientTlsTrustStoreType、brokerClientTlsTrustStore、brokerClientTlsTrustStorePassword、brokerClientTrustCertsFilePath、listenerName。建议使用其 BuilderClusterData.builder()...build()构建对象这也是仓库测试代码如pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApi2Test.java中的通行写法。另外Clusters接口中每个同步方法都配有对应的异步版本createClusterAsync、getClusterAsync等返回CompletableFuture便于在响应式/异步编程模型中使用。初始化集群元数据Initialize cluster metadata创建集群后还需要初始化该集群的元数据。初始化时需要指定以下全部信息集群名称集群的本地 ZooKeeper 连接串整个实例的配置存储Configuration Store连接串集群的 Web 服务 URL供集群内 broker 交互使用的 broker 服务 URL。必须在启动任何属于该集群的 broker 之前完成元数据初始化。为什么不能用 REST / Java Admin API 做元数据初始化与 Pulsar 绝大多数管理功能不同集群元数据初始化不能通过 admin REST API 或 admin Java 客户端完成因为元数据初始化需要直接与 ZooKeeper 通信而非经由 broker 的 HTTP 管理接口。此时应使用pulsarCLI 工具中的initialize-cluster-metadata命令。该命令的服务端入口位于pulsar-broker/src/main/java/org/apache/pulsar/PulsarClusterMetadataSetup.java。元数据初始化命令示例bin/pulsar initialize-cluster-metadata \ --cluster us-west \ --zookeeper zk1.us-west.example.com:2181 \ --configuration-store zk1.us-west.example.com:2184 \ --web-service-url http://pulsar.us-west.example.com:8080/ \ --web-service-url-tls https://pulsar.us-west.example.com:8443/ \ --broker-service-url pulsar://pulsar.us-west.example.com:6650/ \ --broker-service-url-tls pulsarssl://pulsar.us-west.example.com:6651/各参数含义与reference-cli-tools.md中的选项表及PulsarClusterMetadataSetup.java源码一致Flag说明必填-c,--cluster集群名称是-zk,--zookeeper本地 ZooKeeper quorum 连接串是--configuration-store整个实例的配置存储连接串在 2.3.2 中该参数标记为 hidden由新参数--configuration-metadata-store取代是-uw,--web-service-url新集群的 Web 服务 URL是-tw,--web-service-url-tls启用 TLS 的新集群 Web 服务 URL否-ub,--broker-service-url新集群的 broker 服务 URL否-tb,--broker-service-url-tls启用 TLS 的新集群 broker 服务 URL否注意--*-tls系列参数--web-service-url-tls、--broker-service-url-tls只有在实例启用了 TLS 认证时才需要。获取集群配置Get configuration你可以随时获取现有集群的配置。pulsar-admin CLI使用get子命令并指定集群名$ pulsar-admin clusters get cluster-1 { serviceUrl: http://my-cluster.org.com:8080/, serviceUrlTls: null, brokerServiceUrl: pulsar://my-cluster.org.com:6650/, brokerServiceUrlTls: null, peerClusterNames: null }从ClustersBase.getCluster()源码看该接口同样要求超级用户权限若集群不存在服务端返回 404Cluster does not exist。另外pulsar-admin clusters get的输出实际来自ClusterDataImpl的全部字段当字段未设置时显示为null。REST APIGET /admin/v2/clusters/:clusterJava Admin APIadmin.clusters().getCluster(clusterName);对应异步版本为admin.clusters().getClusterAsync(clusterName)返回CompletableFutureClusterData。更新集群配置Update你可以随时更新现有集群的配置。pulsar-admin CLI使用update子命令通过 flag 指定新的配置值$ pulsar-admin clusters update cluster-1 \ --url http://my-cluster.org.com:4081 \ --broker-url pulsar://my-cluster.org.com:3350update与create使用同一套ClusterDetailsCommand参数体系因此前面表格中列出的全部参数TLS、proxy、auth、cluster-config-file 等在更新时同样可用。从ClustersBase.updateCluster()源码看服务端会校验超级用户权限与策略只读状态若集群不存在则返回 404。REST APIPOST /admin/v2/clusters/:cluster请求体与创建时相同为ClusterDataImplJSON{ serviceUrl: http://pulsar.example.com:8080, brokerServiceUrl: pulsar://pulsar.example.com:6651 }Java Admin APIClusterData clusterData new ClusterData( serviceUrl, serviceUrlTls, brokerServiceUrl, brokerServiceUrlTls ); admin.clusters().updateCluster(clusterName, clusterData);删除集群Delete集群可以从 Pulsar 实例中删除。pulsar-admin CLI使用delete子命令并指定集群名$ pulsar-admin clusters delete cluster-1CmdClusters.Delete子命令还额外支持-a/--all参数当指定时会先级联删除该集群下所有 tenant、namespace、分区与非分区 topic并强制删除 schema最后再删除集群本身。删除前提从ClustersBase.deleteCluster()源码看若该集群仍被任何 tenant 使用即仍有 namespace 部署在上面或存在非空的 namespace 隔离策略服务端将返回 412Cluster not empty。因此删除前务必先清理该集群上的所有业务数据。REST APIDELETE /admin/v2/clusters/:clusterJava Admin APIadmin.clusters().deleteCluster(clusterName);列举集群List你可以获取 Pulsar 实例中所有集群的列表。pulsar-admin CLI使用list子命令$ pulsar-admin clusters list cluster-1 cluster-2注意服务端实现ClustersBase.getClusters()会从返回列表中过滤掉内部保留的global集群。REST APIGET /admin/v2/clustersJava Admin APIadmin.clusters().getClusters();返回类型为ListString异步版本getClustersAsync()返回CompletableFutureListString。配置 peer-cluster 数据Update peer-cluster dataPeer clusters对等集群用于跨集群的复制replication与故障切换场景。在 [2.3.2 版本中peer-cluster 采用单向配置语义如果集群 A 把集群 B 加入其 peer 列表则 A 可以向 B 复制数据而 B 若要向 A 复制需要单独将 A 加入 B 的 peer 列表。pulsar-admin CLI使用update-peer-clusters子命令并指定 peer 集群名列表$ pulsar-admin update-peer-clusters cluster-1 --peer-clusters cluster-2--peer-clusters接受逗号分隔的集群名列表。从CmdClusters.UpdatePeerClusters源码看传入空字符串时会解析为null即清空该集群的 peer 列表传null同样表示清空。REST APIPOST /admin/v2/clusters/:cluster/peers请求体为 peer 集群名数组[ cluster-a, cluster-b ]从ClustersBase.setPeerClusterNames()源码可以看到服务端会做两件事一是校验每个 peer 集群确实存在不存在返回 412二是禁止把集群自身加入自己的 peer 列表同样返回 412提示itself cant be part of peer-list。Java Admin APIadmin.clusters().updatePeerClusterNames(clusterName, peerClusterList);peerClusterList类型为LinkedHashSetString保证顺序且去重。仓库测试pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApi2Test.java中验证了完整行为设置 peer 列表、校验getCluster().getPeerClusterNames()返回值、将自身加入 peer 列表时抛出异常、传null清空列表等场景。相关子命令速查pulsar-admin clusters在仓库中还提供了上述核心操作之外的能力同见CmdClusters.java一并列出便于查阅子命令说明get-peer-clusters获取指定集群的 peer 集群列表create-failure-domain/update-failure-domain为集群创建/更新 failure-domain可用--domain-name与--broker-list参数get-failure-domain获取某个 failure-domain 的 broker 配置delete-failure-domain删除 failure-domainlist-failure-domains列出集群中所有 failure-domain小结与最佳实践综合本指南管理 Pulsar 集群时应记住以下几点三类管理入口等价pulsar-admin clusters命令、/admin/v2/clustersREST 端点与PulsarAdmin.clusters()Java 方法覆盖相同的集群生命周期操作可按场景选用元数据初始化走 CLI创建集群后必须用bin/pulsar initialize-cluster-metadata初始化元数据需本地 ZooKeeper 与配置存储连接串且必须在启动 broker 之前完成此步骤无法通过 REST/Java API 代替权限要求严格创建、查询、更新、删除、peer 配置等集群管理操作均要求超级用户权限源码中统一调用validateSuperUserAccess()删除前先清理集群仍被 tenant 使用时会拒绝删除412命令行可用--all级联清理REST/Java 需先手工清理 namespace 与 topicTLS 参数按需提供--*-tls系列与 keystore 相关参数仅在启用 TLS 认证时需要且启用 keystore 时必须同时提供 trust-store 类型、路径与密码否则 CLI 会直接报错CmdClusters.validateClusterData。通过掌握以上操作你可以独立完成 Pulsar 多集群实例的初始化、日常变更与下线清理为跨地域复制和多集群容灾打下基础。赞分享消息队列后端流处理【免费下载链接】pulsarApache Pulsar - distributed pub-sub messaging system项目地址https://gitcode.com/gh_mirrors/pulsar28/pulsar点击查看免费下载相关推荐Apache Pulsar 集群管理完全指南pulsar-admin、REST API 与 Java Admin API 实战Apache Pulsar 集群管理完全指南pulsar admin、REST API 与 Java Admin API 实战 本指南以 admin api消息队列后端流处理Apache Pulsar 集群管理完全指南基于 pulsar-admin、REST API 与 Java Admin API 的实战操作Apache Pulsar 集群管理完全指南基于 pulsar admin、REST API 与 Java Admin API 的实战操作 本文以 Apach消息队列后端流处理Apache Pulsar 集群管理实战pulsar-admin、REST API 与 Java Admin API 全指南Apache Pulsar 集群管理实战pulsar admin、REST API 与 Java Admin API 全指南 本文以 Apache Pulsa消息队列后端流处理上一篇Ginkgo Cloud Lab 无细胞蛋白表达验证CFPS Validation协议实战从序列到定量 go/no-go 读出的自动化流程下一篇7天入门Go语言极客时间《Go语言核心36讲》高效学习指南创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
