消息队列后端流处理【免费下载链接】pulsarApache Pulsar - distributed pub-sub messaging system项目地址https://gitcode.com/gh_mirrors/pulsar28/pulsar点击查看免费下载导读本文聚焦 Apache Pulsar 官方 C 客户端库libpulsar完整覆盖在 Linux 与 macOS 上通过预编译包或源码构建的方式安装客户端、理解pulsar://与pulsarssl://协议 URL 的连接规则以及基于Client、Consumer、Producer三大核心类编写消息生产、消费与 TLS 认证程序的实战方法。文中所有命令、配置与 API 说明均以当前仓库 pulsar-client-cpp 目录下的源码、打包脚本与头文件为准读者可据此在真实环境中直接落地使用。支持的平台Pulsar C 客户端已在macOS与Linux两大平台上完成测试验证。客户端库本体位于仓库的 pulsar-client-cpp 目录公共头文件集中在 pulsar-client-cpp/include/pulsar既提供 C API也通过 pulsar-client-cpp/include/pulsar/c 提供 C API可被 C/C 程序直接引用。Linux 安装自 Pulsar2.1.0版本起官方随发布版本提供预编译的 RPM 与 Debian 包用户可以直接下载安装无需自行编译。RPM 包安装RPM 发布包含三个包包内容pulsar-client动态库libpulsar.sopulsar-client-devel静态库libpulsar.a以及 C 与 C 头文件pulsar-client-debuginfolibpulsar.so的调试符号下载对应架构的 RPM 包后在包所在目录执行$ rpm -ivh apache-pulsar-client*.rpmDEB 包安装Debian/Ubuntu 系列发行版提供两个包包内容pulsar-client动态库libpulsar.sopulsar-client-dev静态库libpulsar.a以及 C 与 C 头文件下载 DEB 包后安装$ dpkg -i apache-pulsar-client*.deb原版文档中的apt-install命令并非标准工具实际安装 DEB 包请使用dpkg -i如需自动解析依赖可先apt-get update后再配合apt-get install -f修复依赖关系。包内容与目录约定以 RPM 为例仓库中的打包规格文件 pulsar-client-cpp/pkg/rpm/SPECS/pulsar-client.spec 明确了安装路径与文件清单动态库安装为/usr/lib/libpulsar.so.version并创建符号链接/usr/lib/libpulsar.so同时提供不依赖 OpenSSL 的libpulsarnossl.so变体开发包安装/usr/lib/libpulsar.a、/usr/lib/libpulsarwithdeps.a将第三方依赖一并静态链入的版本以及/usr/include/pulsar头文件目录文档与 LICENSE 文件安装至/usr/share/doc/下。也就是说安装 devel/dev 包后即可在代码中直接#include pulsar/Client.h并链接-lpulsar。从源码构建 RPM / Debian 包如果你希望基于最新 master 分支自行产出安装包仓库提供了现成的 Docker 构建脚本所有命令均在 Pulsar 仓库根目录执行。先构建 Java 模块C 库的打包过程会引用版本信息与部分生成资源因此需要先构建 Java 模块mvn install -DskipTests构建 RPMpulsar-client-cpp/pkg/rpm/docker-build-rpm.sh该脚本会拉取apachepulsar/pulsar-build:centos-7构建镜像将仓库根目录挂载进容器后执行容器内的构建详见 docker-build-rpm.sh。构建产物 RPM 文件位于pulsar-client-cpp/pkg/rpm/RPMS/x86_64/包内产物与前面表格一致pulsar-client共享库、pulsar-client-devel静态库与头文件、pulsar-client-debuginfo调试符号。构建 Debian 包pulsar-client-cpp/pkg/deb/docker-build-deb.sh对应地该脚本使用apachepulsar/pulsar-build:debian-9镜像见 docker-build-deb.shDebian 包产出目录为pulsar-client-cpp/pkg/deb/BUILD/DEB/构建细节静态链接与多目标产物从 RPM 的 spec 文件pulsar-client.spec可以看到实际构建命令cmake . -DBUILD_TESTSOFF -DLINK_STATICON -DBUILD_PYTHON_WRAPPEROFF make pulsarShared pulsarSharedNossl pulsarStatic pulsarStaticWithDeps -j 3这段构建逻辑说明了几点关键事实LINK_STATICON使能静态链接依赖因此libpulsar.a/libpulsarwithdeps.a内已包含 Boost、Protobuf、cURL 等依赖的代码用户链接时无需再单独指定同时产出pulsarShared带 SSL 的动态库、pulsarSharedNossl不带 SSL 的动态库、pulsarStatic静态库与pulsarStaticWithDeps含依赖的静态库四种目标打包时排除测试BUILD_TESTSOFF与 Python wrapperBUILD_PYTHON_WRAPPEROFF。如果不想走 Docker也可以直接在本地运行 build-rpm.sh / build-deb.sh但需要自行保证本机具备完整构建依赖。macOS 安装在 macOS 上Pulsar 发布版本通过 Homebrew 提供。可以直接安装brew install libpulsar安装完成后库文件libpulsar.dylib、libpulsar.a与include/pulsar头文件会一并就位。仓库中保留了对应的 Homebrew 配方 homebrew/libpulsar.rb从中可以看出它依赖cmake、openssl、boost、jsoncpp、protobuf2.6等构建组件并通过cmake . -DBUILD_TESTSOFF -DLINK_STATICON与make pulsarShared pulsarStatic完成编译安装与 Linux 上的构建目标一致。Connection URLs连接地址规则使用客户端连接 Pulsar 时必须指定一个 Pulsar 协议 URLPulsar 协议 URL 与具体集群绑定使用pulsarURI 协议方案默认端口为6650。本地连接的示例pulsar://localhost:6650生产集群的典型地址pulsar://pulsar.us-west.example.com:6650启用 TLS 加密后协议方案变为pulsarssl端口相应改为6651pulsarssl://pulsar.us-west.example.com:6651从源码结构看客户端通过Client构造函数的serviceUrl参数解析上述地址例如 Client.h 中定义的两个构造重载默认配置与自定义配置都接收该 URL 字符串底层由ClientImpl负责与 broker 建立连接。Consumer订阅并消费消息下面的完整示例创建一个连接本地 broker 的客户端订阅my-topic并以订阅名my-subscribtion-name持续接收、打印并确认消息#include pulsar/Client.h #include iostream using namespace pulsar; int main() { Client client(pulsar://localhost:6650); Consumer consumer; Result result client.subscribe(my-topic, my-subscribtion-name, consumer); if (result ! ResultOk) { LOG_ERROR(Failed to subscribe: result); return -1; } Message msg; while (true) { consumer.receive(msg); LOG_INFO(Received: msg with payload msg.getDataAsString() ); consumer.acknowledge(msg); } client.close(); }要点client.subscribe(topic, subscriptionName, consumer)同步订阅成功后consumer即被填充为可用实例订阅失败时返回Result错误码可用result ! ResultOk判断consumer.receive(msg)阻塞式接收消息msg.getDataAsString()取出 payload 文本每条消息处理完成后务必调用consumer.acknowledge(msg)进行确认broker 据此推进消费位点、避免消息重复投递除了单 topic 订阅Client.h 还提供了多 topic 订阅subscribe(const std::vectorstd::string, ...)与正则订阅subscribeWithRegex以及对应的subscribeAsync异步变体适合需要一次性消费多个 topic 的场景。Producer向 topic 发布消息生产者示例创建一个指向my-topic的生产者循环发送 10 条消息#include pulsar/Client.h #include iostream using namespace pulsar; int main() { Client client(pulsar://localhost:6650); Producer producer; Result result client.createProducer(my-topic, producer); if (result ! ResultOk) { LOG_ERROR(Error creating producer: result); return -1; } // Publish 10 messages to the topic for (int i 0; i 10; i) { Message msg MessageBuilder().setContent(my-message).build(); Result res producer.send(msg); LOG_INFO(Message sent: res); } client.close(); }要点MessageBuilder().setContent(...).build()是构造消息的标准方式也可通过setProperties、setPartitionKey等进一步定制对应头文件 MessageBuilder.hproducer.send(msg)为同步发送返回Result表示本次发送成功与否需要更高吞吐时Client提供createProducerAsync异步创建生产者的回调接口见 Client.h生产端还有sendAsync可避免逐个等待 broker 确认client.close()会等待所有挂起的写请求持久化完成后再释放资源若需立即释放可使用shutdown()。AuthenticationTLS 双向认证当 broker 开启 TLS 与双向认证mTLS时通过ClientConfiguration配置证书与认证插件#include pulsar/Client.h using namespace pulsar; int main() { ClientConfiguration config ClientConfiguration(); config.setUseTls(true); config.setTlsTrustCertsFilePath(/path/to/cacert.pem); config.setTlsAllowInsecureConnection(false); config.setAuth(pulsar::AuthTls::create( /path/to/client-cert.pem, /path/to/client-key.pem)); Client client(pulsarssl://my-broker.com:6651, config); }配置项说明setUseTls(true)启用 TLS 加密此时连接地址必须使用pulsarssl://方案setTlsTrustCertsFilePath指定用于校验 broker 证书的 CA 证书cacert.pem路径setTlsAllowInsecureConnection(false)禁止接受 broker 端未受信任的证书setAuth(pulsar::AuthTls::create(clientCert, clientKey))使用 TLS 客户端证书/私钥完成客户端身份认证这是 Pulsar TLS 双向认证的标准配置方式相关接口定义见 ClientConfiguration.h同头文件还提供了setValidateHostName(bool)用于开启基于 RFC 2818 的主机名CN/SAN校验。客户端核心 API 与配置深入Client 对象Client.h 是 C 客户端的门面核心能力包括能力方法创建生产者createProducer/createProducerAsync订阅消费subscribe/subscribeAsync单 topic、多 topic、正则创建 ReadercreateReader/createReaderAsync按消息 ID 定位支持MessageId::earliest、MessageId::latest或指定位置分区发现getPartitionsForTopic/getPartitionsForTopicAsync自 2.3.0 起返回 topic 的全部分区列表关闭释放close/closeAsync有序关闭并等待写请求持久化、shutdown立即释放状态查询getNumberOfProducers/getNumberOfConsumers此外Reader 提供不依赖订阅的低层读取能力适合需要手动定位消息位置的场景如重放指定消息 ID 之后的数据但只能作用于非分区 topic。ClientConfiguration 常用参数结合 ClientConfiguration.h客户端级配置的核心参数如下配置项默认值说明setOperationTimeoutSeconds30 秒订阅、创建 producer、关闭、取消订阅等客户端操作的超时setIOThreads1客户端使用的 IO 线程数setMessageListenerThreads1消息 listener 投递线程数多线程时不同 listener 分派到不同线程但单个 listener 始终固定同一线程setConcurrentLookupRequest50000每条 broker 连接上允许的并发 lookup 请求数避免 broker 过载在单客户端需要创建/订阅数千 topic 时再调高setConnectionTimeout10000 ms建立 broker 连接的等待超时超时后放弃该次连接尝试setStatsIntervalInSeconds600 秒统计信息打印与重置间隔设为 0 表示关闭统计setPartititionsUpdateInterval60 秒分区 topic 元数据分区数的刷新间隔分区扩容后客户端自动为新增分区创建 producer/consumersetMemoryLimit0不限制客户端实例允许分配的内存上限字节用于控制内存占用setListenerName-指定 broker 返回的advertisedListener对应的 listener 名称小结Apache Pulsar C 客户端提供了一条从安装到上线的完整路径Linux 上可直接使用 2.1.0 起发布的 RPM/DEB 预编译包rpm -ivh/dpkg -i也可借助仓库自带的 Docker 脚本在 pulsar-client-cpp/pkg/rpm 与 pulsar-client-cpp/pkg/deb 产出包含静态链接依赖的安装包macOS 用户通过 Homebrew 的libpulsar公式一键安装。连接层牢记pulsar://6650与pulsarssl://6651两种 URL 方案业务侧则围绕Client、Producer、Consumer三大对象编写同步或异步的生产消费逻辑并通过ClientConfiguration精确控制超时、线程数、内存上限与 TLS 认证参数。如需在项目里进一步研究源码细节可重点阅读 Client.h、ClientConfiguration.h 与客户端测试目录 pulsar-client-cpp/tests 中的端到端用例。赞分享消息队列后端流处理【免费下载链接】pulsarApache Pulsar - distributed pub-sub messaging system项目地址https://gitcode.com/gh_mirrors/pulsar28/pulsar点击查看免费下载相关推荐Apache Pulsar C 客户端开发指南构建安装、生产者与消费者实战Apache Pulsar C 客户端开发指南构建安装、生产者与消费者实战 本指南基于当前仓库的官方文档 client libraries cpp.md消息队列后端流处理Apache Pulsar C 客户端完全指南安装、构建、连接与收发消息实战Apache Pulsar C 客户端完全指南安装、构建、连接与收发消息实战 本指南系统讲解 Apache Pulsar C 客户端 libpuls消息队列后端流处理Apache Pulsar C 客户端完整指南安装、构建与消息收发实战Apache Pulsar C 客户端完整指南安装、构建与消息收发实战 Apache Pulsar 官方提供基于 C 编写的客户端库 libpuls消息队列后端流处理上一篇React Select Search 项目教程下一篇Grit内部机制揭秘混合执行模式与纯Ruby实现创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考