消息队列后端流处理【免费下载链接】pulsarApache Pulsar - distributed pub-sub messaging system项目地址https://gitcode.com/gh_mirrors/pulsar28/pulsar点击查看免费下载本指南以 Apache Pulsar 官方 Docker 镜像apachepulsar/pulsar为主线完整演示如何在本地开发与测试环境中以 standalone 模式启动一个单节点 Pulsar 集群并通过 Python 客户端完成主题topic的生产、消费与统计查询。读完本文你将掌握 Docker 启动命令中端口、数据卷等关键参数的语义理解public/default自动命名空间与数据持久化机制并能独立复现启动集群 → 收发消息 → 查看主题统计的完整闭环。本文内容以 getting-started-docker.md 为主体骨架并结合当前仓库中的 Docker 构建文件与 Standalone 启动源码进行深度扩充。为什么用 Docker 跑 Pulsar StandalonePulsar 是一个分布式 pub-sub 消息系统完整集群通常由 ZooKeeper元数据存储、BookKeeper持久化存储与 Broker消息代理三类角色组成。对于本地开发与功能验证场景Pulsar 提供了 standalone 运行模式在单个进程内同时拉起上述全部组件形成最小可用集群。使用 Docker 镜像而非本地安装发行包可以免去 JDK、Python、依赖库等环境准备一条命令即可拉起完整服务且镜像内各组件版本与官方发布包严格一致。当前仓库中 docker/pulsar/Dockerfile 的构建逻辑印证了这一点镜像分为两个阶段第一阶段基于busybox解包 Pulsar 发布 tar 包并放入辅助脚本第二阶段基于ubuntu:20.04安装openjdk-11-jdk-headless与 Python 依赖最后以非 root 用户UID 10000运行默认日志级别为INFO,CONSOLEENV PULSAR_ROOT_LOGGERINFO,CONSOLE。如果你尚未安装 Docker请先安装 Docker Community Edition并按照所在操作系统的官方说明完成配置。用一条 Docker 命令启动 Standalone 集群在终端执行以下命令Linux / macOS$ docker run -it \ -p 6650:6650 \ -p 8080:8080 \ -v $PWD/data:/pulsar/data \ apachepulsar/pulsar:pulsar:version \ bin/pulsar standaloneWindows 用户使用 PowerShell 或 cmd 时需要把数据卷路径转为小写命令形如$ docker run -it \ -p 6650:6650 \ -p 8080:8080 \ -v $PWD/data:/pulsar/data.ToLower() \ apachepulsar/pulsar:pulsar:version \ bin/pulsar standalone说明pulsar:version是文档构建时的版本占位符实际执行时请替换为你需要的具体版本号例如2.3.1。命令参数逐项解析参数含义-it以交互模式附加到容器终端便于实时查看INFO级日志并可用 CtrlC 停止服务-p 6650:6650将容器内 Broker 二进制协议端口 6650 映射到宿主机客户端通过pulsar://协议连接-p 8080:8080将容器内 HTTP 管理端口 8080 映射到宿主机REST Admin API 与统计查询均走该端口-v $PWD/data:/pulsar/data数据卷挂载容器内的数据与元数据写入宿主机当前目录下的data目录避免容器重启后一切从零开始bin/pulsar standalone容器入口执行的启动命令等价于运行 PulsarStandaloneStarter 主类关于参数有三个细节值得展开端口语义6650 对应 Broker 二进制服务端口与 conf/standalone.conf 中的brokerServicePort6650一致8080 对应 HTTP 服务端口对应webServicePort8080。二者在镜像内默认监听容器内部地址-p将其暴露到宿主机localhost因此本地客户端可以直接访问pulsar://localhost:6650与http://localhost:8080。Windows 路径小写$PWD/data是宿主机上的挂载源目录在 Windows 上该路径必须为小写。例如可以显式指定为E:/data。$PWD/data:/pulsar/data.ToLower()的作用是把整个卷映射字符串转为小写以满足 Docker for Windows 的路径校验。数据持久化/pulsar/data是镜像内 Pulsar 各组件写入数据的默认位置。Dockerfile 注释明确列出各组件的写目录约定BookKeeper 与 ZooKeeper 使用/pulsar/dataFunction Worker 与 pulsar-admin 初始化/pulsar/logsFunctions 写入/pulsar/download。挂载宿主机目录后容器删除或重建都不会丢失已存储的消息数据与元数据。启动成功的标志如果启动成功控制台会出现类似下面的INFO级日志2017-08-09 22:34:04,030 - INFO - [main:WebService213] - Web Service started at http://127.0.0.1:8080 2017-08-09 22:34:04,038 - INFO - [main:PulsarService335] - messaging service is ready, bootstrap service on port8080, broker urlpulsar://127.0.0.1:6650, clusterstandalone, configsorg.apache.pulsar.broker.ServiceConfiguration4db60246 ...其中第二行是关键messaging service is ready表示消息服务就绪broker urlpulsar://127.0.0.1:6650给出客户端应连接的地址clusterstandalone表明当前以 standalone 集群身份运行。从源码理解 standalone 启动流程bin/pulsar standalone最终进入 PulsarStandaloneStarter.main。该入口类会依次完成解析命令行参数如--no-broker、--only-broker互斥校验、读取conf/standalone.conf生成ServiceConfiguration、将 ZooKeeper 地址固定为本地127.0.0.1与指定端口并设置config.setRunningStandalone(true)然后启动 Standalone 服务。其父类 PulsarStandalone 内部持有三个核心组件LocalBookkeeperEnsemble bkEnsemble进程内 BookKeeper 集群PulsarService brokerPulsar Broker 服务WorkerService fnWorkerServicePulsar Functions Worker用于本地函数运行。此外启动流程会注册 JVM 关闭钩子在进程退出时依次停止 Functions Worker、Broker 与 BookKeeper保证容器停止时数据落盘完整。自动创建的 public/default 命名空间Standalone 集群启动后系统会自动创建一个public/default命名空间专供开发测试使用。在 Pulsar 中所有主题都隶属于某个命名空间namespace命名空间是资源隔离与策略管理的基本单位。主题的完整名称形如persistent://public/default/my-topic。关于命名空间、主题的详细概念可参阅 concepts-messaging.md 中的 Topics 一节。Standalone 模式下自动创建该命名空间意味着你拿到一个刚启动的集群后无需任何 admin 操作即可直接创建并读写主题这是它适合快速上手的重要原因。启动消息生产与消费Pulsar 提供 Java、Go、Python、C 等多语言客户端库。在本地 standalone 集群场景下客户端只需连接以下两个地址之一pulsar://localhost:6650—— 二进制协议地址客户端推荐使用http://localhost:8080—— HTTP 管理/查找地址。下面以 Python 客户端为例完成一次完整的订阅消费 生产发送闭环。对应语言的完整文档可参考 client-libraries-python.md、client-libraries-java.md、client-libraries-go.md 与 client-libraries-cpp.md。安装 Python 客户端直接从 PyPI 安装官方客户端$ pip install pulsar-client创建消费者并订阅主题新建文件consumer.py内容如下import pulsar client pulsar.Client(pulsar://localhost:6650) consumer client.subscribe(my-topic, subscription_namemy-sub) while True: msg consumer.receive() print(Received message: %s % msg.data()) consumer.acknowledge(msg) client.close()这段代码的要点client.subscribe(my-topic, subscription_namemy-sub)创建订阅名为my-sub的消费者。这里主题名未带persistent://public/default/前缀客户端会自动补全为默认命名空间下的持久化主题。consumer.receive()是阻塞式接收没有消息时线程挂起等待收到消息后打印内容并调用acknowledge(msg)显式确认。消息确认ack语义很重要只有确认后的消息才不会被重新投递这是 Pulsar 保证至少一次at-least-once投递语义的基础。启动生产者发送测试消息另开一个终端新建文件producer.pyimport pulsar client pulsar.Client(pulsar://localhost:6650) producer client.create_producer(my-topic) for i in range(10): producer.send((hello-pulsar-%d % i).encode(utf-8)) client.close()create_producer(my-topic)会在主题不存在时自动创建随后循环发送 10 条形如hello-pulsar-0、hello-pulsar-1……hello-pulsar-9的 UTF-8 编码消息。send为同步发送消息写入后返回实际工程中也可用send_async异步发送以获得更高吞吐。先运行消费者它会一直等待消息再运行生产者$ python consumer.py $ python producer.py消费者终端会依次打印收到的 10 条消息验证了从生产到消费的完整链路。获取主题统计信息Pulsar 支持通过 REST API、Java Admin API 或命令行工具pulsar-admin管理系统各方面。以最轻量的方式可以用curl直接查询某个主题的统计信息$ curl http://localhost:8080/admin/v2/persistent/public/default/my-topic/stats | python -m json.tool该 URL 的路径结构为persistent://public/default/my-topic主题的 REST 统计端点python -m json.tool只是把 JSON 输出做美化排版。返回结果大致如下{ averageMsgSize: 0.0, msgRateIn: 0.0, msgRateOut: 0.0, msgThroughputIn: 0.0, msgThroughputOut: 0.0, publishers: [ { address: /172.17.0.1:35048, averageMsgSize: 0.0, clientVersion: 1.19.0-incubating, connectedSince: 2017-08-09 20:59:34.6210000, msgRateIn: 0.0, msgThroughputIn: 0.0, producerId: 0, producerName: standalone-0-1 } ], replication: {}, storageSize: 16, subscriptions: { my-sub: { blockedSubscriptionOnUnackedMsgs: false, consumers: [ { address: /172.17.0.1:35064, availablePermits: 996, blockedConsumerOnUnackedMsgs: false, clientVersion: 1.19.0-incubating, connectedSince: 2017-08-09 21:05:39.2220000, consumerName: 166111, msgRateOut: 0.0, msgRateRedeliver: 0.0, msgThroughputOut: 0.0, unackedMessages: 0 } ], msgBacklog: 0, msgRateExpired: 0.0, msgRateOut: 0.0, msgRateRedeliver: 0.0, msgThroughputOut: 0.0, type: Exclusive, unackedMessages: 0 } } }对返回字段做简要解读顶层字段msgRateIn/msgRateOut分别表示主题的入站与出站消息速率条/秒msgThroughputIn/msgThroughputOut为对应吞吐字节/秒publishers数组列出当前连接的生产者包含其地址、客户端版本、连接时间与producerNamesubscriptions以订阅名为键列出每个订阅的消费者列表、订阅类型本例为Exclusive、消息积压msgBacklog、未确认消息数unackedMessages等指标storageSize表示主题当前占用的存储字节数。这段统计输出中my-sub订阅的type为Exclusive独占订阅即同一时刻仅允许一个消费者接收该订阅的消息这是 Pulsar 订阅类型中的默认模式。常见问题与实用技巧端口冲突如果宿主机 6650 或 8080 已被占用启动会失败。可改用其他宿主机端口映射如-p 16650:6650 -p 18080:8080此时客户端连接地址相应改为pulsar://localhost:16650、http://localhost:18080。容器重启后数据仍在只要挂载卷不变重启容器不会丢失已存储的数据因为数据实际落在宿主机$PWD/data目录中。容器内配置覆盖Pulsar 镜像提供了通过环境变量覆盖配置文件的机制。镜像内置了 apply-config-from-env.py 等辅助脚本以及 gen-yml-from-env.py、generate-zookeeper-config.sh支持以PULSAR_PREFIX_为前缀的环境变量批量改写 conf 下的 properties 配置适合在编排环境如 docker-compose、Kubernetes中做参数化部署。停止与清理在交互终端中按 CtrlC 触发优雅关闭对应 Standalone 注册的 JVM 关闭钩子若需彻底清理本地数据删除宿主机data目录即可恢复全新集群状态。小结通过 Docker 镜像运行 Pulsar Standalone是本地开发、功能验证与学习 Pulsar 概念的最快捷路径。本文完整覆盖了启动命令的参数语义、public/default自动命名空间、Python 客户端收发消息以及 REST 统计查询四个核心环节并补充了 Dockerfile 构建细节与PulsarStandalone启动源码作为底层印证。在此基础上你已具备继续探索 Admin API 概览 的坚实基础——后者展示了如何用 REST、Java 与命令行工具精细控制 Pulsar 系统的每个方面。赞分享消息队列后端流处理【免费下载链接】pulsarApache Pulsar - distributed pub-sub messaging system项目地址https://gitcode.com/gh_mirrors/pulsar28/pulsar点击查看免费下载相关推荐Apache Pulsar 快速入门使用 Docker 在 Standalone 模式下启动集群并收发消息Apache Pulsar 快速入门使用 Docker 在 Standalone 模式下启动集群并收发消息 导读 本文面向希望在本地快速体验 Apache P消息队列后端流处理Apache Pulsar Docker 快速入门在容器中运行 standalone 模式并完成消息收发Apache Pulsar Docker 快速入门在容器中运行 standalone 模式并完成消息收发 本篇技术指南以 Apache Pulsar 官方 D消息队列后端流处理Apache Pulsar Docker 单机部署实战在容器中运行 Standalone 集群并完成消息收发与统计查询Apache Pulsar Docker 单机部署实战在容器中运行 Standalone 集群并完成消息收发与统计查询 本文基于 Pulsar 官方文档《Se消息队列后端流处理上一篇SimpleScreenRecorderLinux上最简单易用的屏幕录制工具完整指南下一篇5分钟快速上手RPCS3模拟器完全配置指南创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考