Apache Pulsar C 客户端实战指南三平台编译安装、生产者/消费者开发与 Schema 配置【免费下载链接】pulsarApache Pulsar - distributed pub-sub messaging system项目地址: https://gitcode.com/gh_mirrors/pulsar28/pulsarApache Pulsar 提供了原生 C 客户端库pulsar-client-cpp让 C 应用可以创建线程安全的 Pulsar 生产者Producer、消费者Consumer与 Reader。本文以仓库中的官方文档 site2/docs/client-libraries-cpp.md 为主线结合仓库内真实的头文件、CMake 构建脚本与示例源码完整覆盖 Linux / macOS / Windows 三大平台的编译与二进制安装方式、四种链接库形态libpulsar.so、libpulsarnossl.so、libpulsar.a、libpulsarwithdeps.a的选型与链接命令并手把手演示阻塞/异步生产者、分区主题上的懒加载生产者、消息分块chunking、阻塞式/监听式消费者、TLS 认证以及 Avro 与 ProtobufNative 两种 Schema 的接入。读完本文你可以独立完成 C 客户端环境的搭建并写出可直接编译运行的生产消费示例。概述与支持平台Pulsar C 客户端用于在 C 程序中创建 Pulsar 生产者与消费者。其官方 API 文档由 Doxygen 生成。从客户端 API 的头文件可以看到客户端入口类为pulsar::Client见 pulsar-client-cpp/include/pulsar/Client.h它封装了createProducer、subscribe、createReader、getPartitionsForTopic、close等核心方法pulsar::Producer、pulsar::Consumer、pulsar::Reader的所有方法均为线程安全因此可以在多线程场景中共享使用。当前仓库中的 C 客户端支持Linux、macOS和Windows64 位三大平台下文分别给出编译安装路径。需要特别说明以下安装步骤以当前仓库pulsar-client-cpp模块为准不同 Pulsar 版本间的依赖版本与安装命令可能略有差异。Linux 平台的安装方式在 Linux 上可以按需选择三种方式之一源码编译、安装 RPM或安装 Debian包。系统依赖编译 C 客户端前需要安装以下组件CMakeBoostProtocol Buffers 3ProtobufNative Schema 依赖libcurlGoogle Test仅测试需要源码编译步骤克隆 Pulsar 仓库$ git clone https://github.com/apache/pulsar安装所有必要依赖Debian/Ubuntu 系$ apt-get install cmake libssl-dev libcurl4-openssl-dev liblog4cxx-dev \ libprotobuf-dev protobuf-compiler libboost-all-dev google-mock libgtest-dev libjsoncpp-dev编译并安装 Google Test。不同libgtest-dev版本的源码目录不同请按版本选择# libgtest-dev 版本为 1.18.0 及以上 $ cd /usr/src/googletest $ sudo cmake . $ sudo make $ sudo cp ./googlemock/libgmock.a ./googlemock/gtest/libgtest.a /usr/lib/ # 版本低于 1.18.0 $ cd /usr/src/gtest $ sudo cmake . $ sudo make $ sudo cp libgtest.a /usr/lib $ cd /usr/src/gmock $ sudo cmake . $ sudo make $ sudo cp libgmock.a /usr/lib在 Pulsar 仓库内编译 C 客户端库$ cd pulsar-client-cpp $ cmake . $ make编译成功后libpulsar.so与libpulsar.a位于仓库的lib目录下性能测试工具perfProducer与perfConsumer位于perf目录中。CMake 脚本中定义了多个可开关的构建选项见 pulsar-client-cpp/CMakeLists.txtBUILD_TESTS默认 ON构建测试、BUILD_PYTHON_WRAPPER默认 ON构建 Python 包装、BUILD_PERF_TOOLS默认 OFF构建性能工具、LINK_STATIC默认 OFF是否静态链接依赖、USE_LOG4CXX默认 OFF等实际编译时可按需传参。安装预编译依赖自 2.1.0 起提供预构建包自 Pulsar 2.1.0 起官方随发行版提供预构建的 RPM 与 Debian 包可直接下载安装。安装完成后libpulsar.so、libpulsarnossl.so、libpulsar.a与libpulsarwithdeps.a会位于系统的/usr/lib目录。默认情况下这些库在代码路径${PULSAR_HOME}/pulsar-client-cpp下构建可用如下命令构建全部四种库形态cmake . -DBUILD_TESTSOFF -DLINK_STATICON make pulsarShared pulsarSharedNossl pulsarStatic pulsarStaticWithDeps -j 3这些库还依赖其他第三方库各依赖的精确版本可查看仓库内的 pulsar-client-cpp/pkg/rpm/Dockerfile 与 pulsar-client-cpp/pkg/deb/Dockerfile。四种库形态的差异与链接方式libpulsar.so共享库静态链接了boost与openssl动态链接其余必要依赖。编译示例g --stdc11 PulsarTest.cpp -o test /usr/lib/libpulsar.so -I/usr/local/ssl/includelibpulsarnossl.so共享库无内置 SSL与libpulsar.so类似区别在于openssl与crypto采用动态链接适合希望使用系统 OpenSSL 的场景。编译示例g --stdc11 PulsarTest.cpp -o test /usr/lib/libpulsarnossl.so -lssl -lcrypto -I/usr/local/ssl/include -L/usr/local/ssl/liblibpulsar.a静态库使用前需要显式加载所有依赖。编译示例g --stdc11 PulsarTest.cpp -o test /usr/lib/libpulsar.a -lssl -lcrypto -ldl -lpthread -I/usr/local/ssl/include -L/usr/local/ssl/lib -lboost_system -lboost_regex -lcurl -lprotobuf -lzstd -lzlibpulsarwithdeps.a静态库含依赖归档基于libpulsar.a并进一步把libboost_regex、libboost_system、libcurl、libprotobuf、libzstd与libz归档进来。编译示例g --stdc11 PulsarTest.cpp -o test /usr/lib/libpulsarwithdeps.a -lssl -lcrypto -ldl -lpthread -I/usr/local/ssl/include -L/usr/local/ssl/lib注意libpulsarwithdeps.a刻意不包含与安全相关的libssl与libcrypto以便让使用者采用本地系统提供的 OpenSSL 版本处理安全问题与升级。安装 RPM从发行版页面下载 RPM 包配套.asc与.sha512校验文件链接加密校验文件clientclient.asc、client.sha512client-debuginfoclient-debuginfo.asc、client-debuginfo.sha512client-develclient-devel.asc、client-devel.sha512安装$ rpm -ivh apache-pulsar-client*.rpm安装成功后 Pulsar 库位于/usr/lib目录例如lrwxrwxrwx 1 root root 18 Dec 30 22:21 libpulsar.so - libpulsar.so.2.9.1 lrwxrwxrwx 1 root root 23 Dec 30 22:21 libpulsarnossl.so - libpulsarnossl.so.2.9.1注意若启动 Pulsar 客户端时出现libpulsar.so: cannot open shared object file: No such file or directory错误通常需要先执行ldconfig刷新共享库缓存。另外若后续需要安装 Node.js 等依赖请先安装 GCC 与 g否则会报错$ sudo yum -y install gcc automake autoconf libtool make $ sudo yum -y install gcc-c安装 Debian下载 Debian 包配套.asc与.sha512校验文件链接加密校验文件clientclient.asc、client.sha512client-develclient-devel.asc、client-devel.sha512安装$ apt install ./apache-pulsar-client*.deb安装成功后Pulsar 库同样位于/usr/lib目录。从最新 master 构建 RPM / Debian 包以下命令需要在克隆的 Pulsar 仓库根目录下执行。打包脚本会生成静态链接了全部依赖的libpulsar.so/libpulsarnossl.so/libpulsar.a/libpulsarwithdeps.a。由于 C 库包构建依赖 Java 包产物需要先构建 Java 部分mvn install -DskipTests构建 RPM在 Docker 容器内构建pulsar-client-cpp/pkg/rpm/docker-build-rpm.shRPM 产物位于pulsar-client-cpp/pkg/rpm/RPMS/x86_64/路径。各包内容如下包名内容pulsar-client共享库libpulsar.so与libpulsarnossl.sopulsar-client-devel静态库libpulsar.a、libpulsarwithdeps.a以及 C / C 头文件pulsar-client-debuginfolibpulsar.so的调试符号构建 Debianpulsar-client-cpp/pkg/deb/docker-build-deb.shDebian 产物位于pulsar-client-cpp/pkg/deb/BUILD/DEB/路径。各包内容如下包名内容pulsar-client共享库libpulsar.so与libpulsarnossl.sopulsar-client-dev静态库libpulsar.a、libpulsarwithdeps.a以及 C / C 头文件macOS 平台的安装方式源码编译克隆 Pulsar 仓库$ git clone https://github.com/apache/pulsar使用 Homebrew 安装依赖# OpenSSL 安装 $ brew install openssl $ export OPENSSL_INCLUDE_DIR/usr/local/opt/openssl/include/ $ export OPENSSL_ROOT_DIR/usr/local/opt/openssl/ # Protocol Buffers、Boost 与 log4cxx 安装 $ brew install protobuf boost boost-python log4cxx # 如果使用 python3需要额外安装 boost-python3 # Google Test 安装 $ git clone https://github.com/google/googletest.git $ cd googletest $ cmake . $ make install在克隆的仓库内编译客户端库$ cd pulsar-client-cpp $ cmake . $ make通过 Homebrew 安装 libpulsarPulsar 发行版已进入 Homebrew core 仓库可直接安装包内自带库文件与头文件brew install libpulsarWindows64 位平台的安装方式源码编译克隆 Pulsar 仓库$ git clone https://github.com/apache/pulsar使用 vcpkg 安装依赖仓库已提供vcpkg.json清单见 pulsar-client-cpp/vcpkg.jsoncd ${PULSAR_HOME}/pulsar-client-cpp vcpkg install --feature-flagsmanifests --triplet x64-windows构建 C 库cmake -B ./build -A x64 -DBUILD_PYTHON_WRAPPEROFF -DBUILD_TESTSOFF -DVCPKG_TRIPLETx64-windows -DCMAKE_BUILD_TYPERelease -S . cmake --build ./build --config Release说明若为 Windows 32 位需要改用-A Win32与-DVCPKG_TRIPLETx86-windows。若使用 MSVC Debug 模式需要将CMAKE_BUILD_TYPE变量与--config选项中的Release都替换为Debug。构建成功后客户端库位于以下位置${PULSAR_HOME}/pulsar-client-cpp/build/lib/Release/pulsar.lib ${PULSAR_HOME}/pulsar-client-cpp/build/lib/Release/pulsar.dllConnection URLs连接地址的写法使用客户端库连接 Pulsar需要指定 Pulsar 协议 URL。Pulsar 协议 URL 隶属于具体集群采用pulsarURI scheme默认端口为 6650。本地示例pulsar://localhost:6650生产环境集群示例pulsar://pulsar.us-west.example.com:6650若启用 TLS 加密需要在 scheme 中加入ssl默认端口变为 6651pulsarssl://pulsar.us-west.example.com:6651创建 Producer阻塞与异步两种风格要在 C 客户端中作为生产者发布消息需要先创建 Producer。pulsar::Client::createProducer有两种重载默认配置与自定义ProducerConfiguration见 pulsar-client-cpp/include/pulsar/Client.h分别对应以下两种用法阻塞式示例见下文每次调用send都会等待 broker 返回 ack。非阻塞异步式示例见下文改用sendAsync并传入回调broker 返回 ack 时触发回调。仓库中的 pulsar-client-cpp/examples/SampleProducer.cc 给出了最小可运行示例可配合下文代码对照学习。简单阻塞式示例以下示例以阻塞方式发送 100 条消息。这种方式代码最简单但每次发送都要等待 ack 返回后才发下一条吞吐量不高#include pulsar/Client.h #include thread using namespace pulsar; int main() { Client client(pulsar://localhost:6650); Result result client.createProducer(persistent://public/default/my-topic, producer); if (result ! ResultOk) { std::cout Error creating producer: result std::endl; return -1; } // Send 100 messages synchronously int ctr 0; while (ctr 100) { std::string content msg std::to_string(ctr); Message msg MessageBuilder().setContent(content).setProperty(x, 1).build(); Result result producer.send(msg); if (result ! ResultOk) { std::cout The message content could not be sent, received code: result std::endl; } else { std::cout The message content sent successfully std::endl; } std::this_thread::sleep_for(std::chrono::milliseconds(100)); ctr; } std::cout Finished producing synchronously! std::endl; client.close(); return 0; }非阻塞异步式示例以下示例改用sendAsync发送 100 条消息允许多条消息同时在途inflight从而显著提升吞吐量#include pulsar/Client.h #include thread using namespace pulsar; std::atomicuint32_t acksReceived; void callback(Result code, const MessageId msgId, std::string msgContent) { // message processing logic here std::cout Received ack for msg: msgContent with code: code -- MsgID: msgId std::endl; acksReceived; } int main() { Client client(pulsar://localhost:6650); ProducerConfiguration producerConf; producerConf.setBlockIfQueueFull(true); Producer producer; Result result client.createProducer(persistent://public/default/my-topic, producerConf, producer); if (result ! ResultOk) { std::cout Error creating producer: result std::endl; return -1; } // Send 100 messages asynchronously int ctr 0; while (ctr 100) { std::string content msg std::to_string(ctr); Message msg MessageBuilder().setContent(content).setProperty(x, 1).build(); producer.sendAsync(msg, std::bind(callback, std::placeholders::_1, std::placeholders::_2, content)); std::this_thread::sleep_for(std::chrono::milliseconds(100)); ctr; } // wait for 100 messages to be acked while (acksReceived 100) { std::this_thread::sleep_for(std::chrono::milliseconds(100)); } std::cout Finished producing asynchronously! std::endl; client.close(); return 0; }这里用到的setBlockIfQueueFull(true)很关键当内部发送队列满时默认行为下send/sendAsync会失败并返回ResultProducerQueueIsFull错误码开启该选项后sendAsync在队列满时会转为阻塞等待代码更简单。若不开此选项则必须在回调中自行处理ResultProducerQueueIsFull重试或丢弃。从 pulsar-client-cpp/include/pulsar/ProducerConfiguration.h 可以看到与队列相关的还有setMaxPendingMessages默认 1000单分区在途消息上限与setMaxPendingMessagesAcrossPartitions默认 50000跨分区在途消息总上限。分区主题与懒加载生产者Lazy Producers当 Pulsar 主题被扩展为数百个分区、生产者数量也扩展到数百甚至上千时会给 broker 带来较大压力在分区主题上创建生产者时客户端内部会为每个分区各创建一个内部生产者每个内部生产者都要与 broker 通信完成定位lookup与连接握手。以「1000 个分区 × 1000 个生产者应用」为例内部生产者总数高达 1,000,000 个。可以通过以下三个手段显著降低压力使用UseSinglePartition分区路由模式保证所有消息只发往随机选定的单个分区使用无 key 消息带 key 的消息会按 key 的哈希路由到多个分区使用懒加载生产者setLazyStartPartitionedProducers(true)仅在消息需要路由到某分区时才按需创建该分区的内部生产者。按上文例子组合使用后内部生产者数量可从 1,000,000 降至 1000。源码注释进一步说明见 pulsar-client-cpp/include/pulsar/ProducerConfiguration.h路由策略选中的首个分区内部生产者总是立即启动其余分区按需懒启动。ProducerConfiguration producerConf; producerConf.setPartitionsRoutingMode(ProducerConfiguration::UseSinglePartition); producerConf.setLazyStartPartitionedProducers(true);注意懒加载模式下首条消息可能有额外延迟——若设置了很短的发送超时setSendTimeout默认 30000ms而初始连接握手较慢首条消息可能触发超时。启用消息分块Chunking消息分块Chunking让 Pulsar 能够处理超大负载消息生产端把大消息拆成多个 chunk 分别发布消费端再聚合还原。概念细节可参考 site2/docs/concepts-messaging.md 中关于 chunking 的章节。分块功能默认关闭。启用示例ProducerConfiguration conf; conf.setBatchingEnabled(false); conf.setChunkingEnabled(true); Producer producer; client.createProducer(my-topic, conf, producer);注意启用 chunking 必须同时关闭批处理setBatchingEnabled(false)否则创建生产者会失败。源码注释见 pulsar-client-cpp/include/pulsar/ProducerConfiguration.h还给出三条实践建议目前仅支持非共享订阅与持久化主题建议调小setMaxPendingMessages以避免缓冲消息占用过多内存建议在命名空间上设置消息 TTL以便在 broker 重启或发布中断导致整条大消息未发完时能清理未消费完成的 chunk 消息。创建 Consumer阻塞与监听两种风格创建消费者同样有两种主要方式阻塞式同步调用receive(msg)。非阻塞事件驱动式注册消息监听器MessageListener。阻塞式示例阻塞式写法最简单不断调用receive(msg)该方法会阻塞直到收到消息。以下示例从最早位点InitialPositionEarliest开始订阅并消费 100 条消息#include pulsar/Client.h using namespace pulsar; int main() { Client client(pulsar://localhost:6650); Consumer consumer; ConsumerConfiguration config; config.setSubscriptionInitialPosition(InitialPositionEarliest); Result result client.subscribe(persistent://public/default/my-topic, consumer-1, config, consumer); if (result ! ResultOk) { std::cout Failed to subscribe: result std::endl; return -1; } Message msg; int ctr 0; // consume 100 messages while (ctr 100) { consumer.receive(msg); std::cout Received: msg with payload msg.getDataAsString() std::endl; consumer.acknowledge(msg); ctr; } std::cout Finished consuming synchronously! std::endl; client.close(); return 0; }关于订阅初始位点setSubscriptionInitialPosition的默认值是InitialPositionLatest见 pulsar-client-cpp/include/pulsar/ConsumerConfiguration.h即新订阅默认只消费订阅之后到达的消息示例中显式设置为InitialPositionEarliest才会从头消费积压消息。使用消息监听器MessageListener事件驱动风格可以避免在应用侧写阻塞循环为每条收到的消息触发一次监听回调。以下示例从最早位点订阅并消费 100 条消息#include pulsar/Client.h #include atomic #include thread using namespace pulsar; std::atomicuint32_t messagesReceived; void handleAckComplete(Result res) { std::cout Ack res: res std::endl; } void listener(Consumer consumer, const Message msg) { std::cout Got message msg with content msg.getDataAsString() std::endl; messagesReceived; consumer.acknowledgeAsync(msg.getMessageId(), handleAckComplete); } int main() { Client client(pulsar://localhost:6650); Consumer consumer; ConsumerConfiguration config; config.setMessageListener(listener); config.setSubscriptionInitialPosition(InitialPositionEarliest); Result result client.subscribe(persistent://public/default/my-topic, consumer-1, config, consumer); if (result ! ResultOk) { std::cout Failed to subscribe: result std::endl; return -1; } // wait for 100 messages to be consumed while (messagesReceived 100) { std::this_thread::sleep_for(std::chrono::milliseconds(100)); } std::cout Finished consuming asynchronously! std::endl; client.close(); return 0; }监听回调中使用了acknowledgeAsync异步确认。从源码看MessageListener的类型是std::functionvoid(Consumer, const Message)见 pulsar-client-cpp/include/pulsar/ConsumerConfiguration.h监听器会按顺序被逐条调用。配置消费者侧 Chunking消费者需要把收到的多个 chunk 聚合还原为原始消息。由于不同生产者的 chunk 可能交错到达例如M1-C1, M2-C1, M1-C2, M2-C2消费端必须同时维护多个缓冲区这会带来内存压力。可以通过两个参数限制并发的分块消息数量setMaxPendingChunkedMessage未完成 chunk 消息的并发上限默认 10设为 0 表示不限制见 pulsar-client-cpp/include/pulsar/ConsumerConfiguration.hsetAutoAckOldestChunkedMessageOnQueueFull达到阈值后对最旧的分块消息是静默确认true还是标记未确认交由 broker 稍后重投false默认 false见 pulsar-client-cpp/include/pulsar/ConsumerConfiguration.h。配置示例ConsumerConfiguration conf; conf.setAutoAckOldestChunkedMessageOnQueueFull(true); conf.setMaxPendingChunkedMessage(100); Consumer consumer; client.subscribe(my-topic, my-sub, conf, consumer);在连接 URL 中启用 TLS 认证使用 TLS 认证连接 Pulsar 时连接 URL 需要加入ssl默认端口 6651并在ClientConfiguration中配置证书与认证插件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);与 TLS 相关的配置项均可从 pulsar-client-cpp/include/pulsar/ClientConfiguration.h 找到对应实现setUseTls默认 false、setTlsTrustCertsFilePath受信 CA 证书路径、setTlsAllowInsecureConnection是否接受 broker 的不可信证书默认 false、setValidateHostName是否按 RFC 2818 校验服务器主机名默认 false。生产环境建议将setTlsAllowInsecureConnection与setValidateHostName分别配置为false与true以强制证书与主机名校验。完整的可运行示例可参考仓库中的 pulsar-client-cpp/examples 目录包含SampleProducer.cc、SampleAsyncProducer.cc、SampleConsumer.cc、SampleConsumerListener.cc以及 C API 版本示例。使用 SchemaAvro 与 ProtobufNativeSchema 让生产与消费双方约定数据的结构化格式broker 侧会基于 topic 的 Schema 对消息做校验与演化管理。关于 Pulsar Schema 的整体概念可参考 site2/docs/schema-get-started.md。C 客户端的 Schema 类型枚举定义在 pulsar-client-cpp/include/pulsar/Schema.h包括STRING、JSON、PROTOBUF、AVRO、PROTOBUF_NATIVE、INT8~DOUBLE、KEY_VALUE、BYTES等并通过SchemaInfo(SchemaType, name, schema)构造后传给ProducerConfiguration::setSchema或ConsumerConfiguration::setSchema。Avro Schema使用 Avro Schema 创建生产者static const std::string exampleSchema {\type\:\record\,\name\:\Example\,\namespace\:\test\, \fields\:[{\name\:\a\,\type\:\int\},{\name\:\b\,\type\:\int\}]}; Producer producer; ProducerConfiguration producerConf; producerConf.setSchema(SchemaInfo(AVRO, Avro, exampleSchema)); client.createProducer(topic-avro, producerConf, producer);使用 Avro Schema 创建消费者static const std::string exampleSchema {\type\:\record\,\name\:\Example\,\namespace\:\test\, \fields\:[{\name\:\a\,\type\:\int\},{\name\:\b\,\type\:\int\}]}; ConsumerConfiguration consumerConf; Consumer consumer; consumerConf.setSchema(SchemaInfo(AVRO, Avro, exampleSchema)); client.subscribe(topic-avro, sub-2, consumerConf, consumer)ProtobufNative SchemaProtobufNative 基于 Protobuf 的Descriptor直接推断消息结构要求使用Protobuf3 或更高版本。接入步骤如下。用 Protobuf3 定义并生成User类syntax proto3; message User { string name 1; int32 age 2; }在源码中包含ProtobufNativeSchema.h头文件确保工程已加入 Protobuf 依赖#include pulsar/ProtobufNativeSchema.hcreateProtobufNativeSchema的声明位于 pulsar-client-cpp/include/pulsar/ProtobufNativeSchema.h接收google::protobuf::Descriptor*并返回SchemaInfo。创建生产者并发送一个User实例ProducerConfiguration producerConf; producerConf.setSchema(createProtobufNativeSchema(User::GetDescriptor())); Producer producer; client.createProducer(topic-protobuf, producerConf, producer); User user; user.set_name(my-name); user.set_age(10); std::string content; user.SerializeToString(content); producer.send(MessageBuilder().setContent(content).build());创建消费者接收并反序列化User实例ConsumerConfiguration consumerConf; consumerConf.setSchema(createProtobufNativeSchema(User::GetDescriptor())); consumerConf.setSubscriptionInitialPosition(InitialPositionEarliest); Consumer consumer; client.subscribe(topic-protobuf, my-sub, consumerConf, consumer); Message msg; consumer.receive(msg); User user2; user2.ParseFromArray(msg.getData(), msg.getLength());注意此处消费者显式设置了InitialPositionEarliest以从最早位点读取此前由生产者写入的消息。总结本文完整覆盖了 Apache Pulsar C 客户端从环境搭建到功能开发的闭环在 Linux / macOS / Windows 三大平台既可以从源码编译也可以直接安装 RPM / Debian / Homebrew 预构建包针对四种库形态libpulsar.so、libpulsarnossl.so、libpulsar.a、libpulsarwithdeps.a给出了精确的链接命令随后围绕生产者阻塞 / 异步 / 分区懒加载 / chunking与消费者阻塞 / 监听器 / chunking 限流给出了可直接编译运行的 C 示例并讲解了 TLS 认证配置与 Avro、ProtobufNative 两种 Schema 的接入方法。相关头文件Client.h、ProducerConfiguration.h、ConsumerConfiguration.h、ClientConfiguration.h、Schema.h、示例pulsar-client-cpp/examples与打包脚本pulsar-client-cpp/pkg/rpm、pulsar-client-cpp/pkg/deb均在当前仓库中可作为深入阅读与二次开发的起点。【免费下载链接】pulsarApache Pulsar - distributed pub-sub messaging system项目地址: https://gitcode.com/gh_mirrors/pulsar28/pulsar创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考