消息队列后端流处理【免费下载链接】pulsarApache Pulsar - distributed pub-sub messaging system项目地址https://gitcode.com/gh_mirrors/pulsar28/pulsar点击查看免费下载导读本文基于 Apache Pulsar 2.2.1 版本文档完整剖析 Pulsar 客户端与 Broker 之间通信所依赖的自定义二进制协议。协议以 Protocol Buffersprotobuf承载命令、以 4 字节长度前缀完成帧定界并支持基于令牌permit的流控、批量消息、校验和与服务发现。读完本文你将掌握 Pulsar 协议帧的字节级布局、消息元数据字段语义、生产者/消费者全交互流程以及二进制协议下的 Topic 查找机制并能在仓库源码pulsar-common/src/main/proto/PulsarApi.proto、Commands.java中定位对应实现加以印证。协议总体设计为什么 Pulsar 需要一套自定义二进制协议Pulsar 在生产者/消费者与 Broker 之间使用一套自定义二进制协议进行通信。该协议设计目标是在支持确认acknowledgement与流控flow control等必需特性的同时获得最大的传输效率与实现效率。相比文本协议二进制 protobuf 的组合在序列化开销、带宽占用与解析性能上都有显著优势。协议中客户端与 Broker 之间交换的是命令Command。每条命令都是一个二进制 protobuf 消息其格式定义在 pulsar-common/src/main/proto/PulsarApi.proto 文件中文末Protobuf 接口一节会进一步说明。所有命令统一封装在一个BaseCommandprotobuf 消息中。BaseCommand内含一个Type枚举枚举值对应全部可能的子命令BaseCommand中每个子命令都以 optional 字段形式挂载且一条BaseCommand只能携带一个子命令。连接共享Connection sharing不同生产者、不同消费者的命令可以在同一条 TCP 连接上自由交错发送不受任何限制。这使得一个客户端只需维护少量连接即可驱动大量 Topic 的生产与消费。BaseCommand 与 Type 枚举在 PulsarApi.proto 中BaseCommand通过required Type type 1;声明当前帧携带哪种命令随后将对应子命令作为 optional 字段挂载例如optional CommandSend send 6;、optional CommandAck ack 10;。Type枚举覆盖了连接CONNECT/CONNECTED、订阅SUBSCRIBE、生产PRODUCER/SEND/SEND_RECEIPT、消费MESSAGE/ACK/FLOW、服务发现LOOKUP/PARTITIONED_METADATA、保活PING/PONG以及事务NEW_TXN等 50 号段等数十种命令类型。从源码结构看BaseCommand的单子命令约束保证了帧语义的确定性让接收方可以无歧义地按type分发到对应处理逻辑。帧格式Framing由于 protobuf 本身不提供消息帧定界Pulsar 协议中所有消息都在开头追加一个 4 字节字段用于指明该帧的大小frame size。单个帧允许的最大尺寸为5 MB——这一限制在源码中有直接对应pulsar-common/src/main/java/org/apache/pulsar/common/protocol/Commands.java第 115 行定义了public static final int DEFAULT_MAX_MESSAGE_SIZE 5 * 1024 * 1024;。协议支持两类命令简单命令Simple commands不携带消息负载payload。负载命令Payload commands携带负载用于消息发布或投递。负载命令中protobuf 命令数据之后紧跟着 protobuf 序列化的消息元数据再之后是负载负载以原始raw二进制格式在 protobuf 之外传递。所有大小字段均以4 字节无符号大端big endian整数传递。消息负载之所以以原始格式而非 protobuf 格式传递是为了效率考虑——避免对用户数据做无意义的二次序列化。简单命令布局简单无负载命令的基本结构如下组件说明大小字节totalSize帧的大小统计其之后所有内容的字节数4commandSizeprotobuf 序列化后命令的大小4message以原始二进制格式序列化的 protobuf 消息可变负载命令布局负载命令的基本结构如下组件说明大小字节totalSize帧的大小统计其之后所有内容的字节数4commandSizeprotobuf 序列化后命令的大小4message以原始二进制格式序列化的 protobuf 消息可变magicNumber2 字节标识当前格式的字节数组0x0e012checksum对其后所有内容的 CRC32-C 校验和4metadataSize消息元数据的大小4metadata以二进制 protobuf 消息存储的消息元数据可变payload帧中剩余的所有字节均视为负载可为任意字节序列可变源码印证Commands.java第 124 行定义了public static final short magicCrc32c 0x0e01;与文档描述的 magic number 完全一致第 1481 行附近以注释形式给出了负载命令的序列化布局[TOTAL_SIZE] [CMD_SIZE][CMD] [MAGIC_NUMBER][CHECKSUM] [METADATA_SIZE][METADATA] [PAYLOAD]。校验和计算通过com.scurrilous.circe.checksum.Crc32cIntChecksumcomputeChecksum/resumeChecksum实现并且ChecksumType枚举Crc32c/None允许在不需要校验时跳过 magic checksum 两段对应旧版本协议的行为详见文末协议版本演进。消息元数据Message Metadata消息元数据以序列化 protobuf 消息的形式与应用负载payload一并存储。元数据由生产者创建并在投递时原样传给消费者。字段说明producer_name发布该消息的生产者名称sequence_id消息的序列号由生产者分配publish_time发布时间戳Unix 时间即自 1970 年 1 月 1 日 UTC 以来的毫秒数properties一组键值对使用KeyValue消息。这些是应用自定义的键与值Pulsar 不赋予其特殊含义replicated_from(可选)表明消息已被复制并指明消息最初发布所在的集群名称partition_key(可选)在分区 Topic 上发布时若存在 key则使用该 key 的哈希决定选择哪个分区compression(可选)表明负载已压缩以及使用的压缩库uncompressed_size(可选)若使用了压缩生产者必须用原始负载大小填充该字段num_messages_in_batch(可选)若该消息实际上是批量消息则必须设为批量中消息的数量源码扩充在 PulsarApi.proto 的MessageMetadata定义中可以看到这些字段的完整 protobuf 声明例如required string producer_name 1;、required uint64 sequence_id 2;、required uint64 publish_time 3;。此外compression字段对应CompressionType枚举proto 定义枚举值含义NONE未压缩LZ4LZ4 压缩ZLIBZLib 压缩ZSTDZStandard 压缩SNAPPYSnappy 压缩在 2.2.1 之后的版本中MessageMetadata还扩展了event_time应用事件时间戳、encryption_keys/encryption_algo/encryption_param端到端加密相关、ordering_keyKey_Shared 模式下覆盖消息排序键、deliver_at_time延迟投递、marker_type内部 marker 消息等字段可结合仓库中的 proto 定义继续查阅。批量消息Batch Messages使用批量消息时负载中包含一组条目entries每个条目拥有各自的元数据由SingleMessageMetadata对象定义。单个批次的负载格式如下字段说明metadataSizeN单条消息元数据序列化为 protobuf 后的大小metadataN单条消息元数据payloadN应用传入的消息负载其中每条元数据字段如下字段说明properties应用定义的属性partition key(可选)用于指示哈希到特定分区的 keypayload_size批量中单条消息的负载大小当启用压缩时整个批次会被一次性整体压缩而非逐条压缩从而获得更高的压缩比。源码印证SingleMessageMetadata的 protobuf 定义位于 PulsarApi.proto其中required int32 payload_size 3;与文档表格一一对应。在BaseCommand.Type中SEND命令的num_messages字段CommandSend 定义即用于一次发送整批消息默认值为 1。交互流程Interactions连接建立Connection establishment在与 Broker 建立 TCP 连接后通常端口为6650由客户端负责发起会话。收到 Broker 的Connected响应后客户端即可认为连接可用。反之若 Broker 无法通过客户端认证则会回复Error命令并关闭 TCP 连接。客户端发起连接时发送的CommandConnect示例message CommandConnect { client_version : Pulsar-Client-Java-v1.15.2, auth_method_name : my-authentication-plugin, auth_data : my-auth-data, protocol_version : 6 }字段说明client_version→ 基于字符串的标识符格式不受强制auth_method_name→(可选)启用认证时使用的认证插件名称auth_data→(可选)插件特定的认证数据protocol_version→ 客户端支持的协议版本。Broker 不会发送更新协议版本中引入的命令且 Broker 可能强制一个最低版本Broker 回复的CommandConnected示例message CommandConnected { server_version : Pulsar-Broker-v1.15.2, protocol_version : 6 }字段说明server_version→ Broker 版本的字符串标识protocol_version→ Broker 支持的协议版本。客户端不得尝试发送更新协议版本中引入的命令源码扩充这两个命令的 protobuf 定义在 PulsarApi.proto 中。CommandConnect在后续版本中还加入了proxy_to_broker_url通过 Pulsar Proxy 中转、original_principal/original_auth_data代理场景下携带原始身份信息以及FeatureFlags如supports_auth_refresh等字段CommandConnected则增加了max_message_size用于通知客户端 Broker 允许的最大消息大小与上文 5 MB 默认值相关。保活机制Keep Alive为识别客户端与 Broker 之间长时间的网络分区或对端机器崩溃但未中断 TCP 连接的情况如断电、内核 panic、硬重启等协议引入了探测远端存活状态的机制。客户端与 Broker周期性发送Ping命令若在超时时间内未收到Pong响应则关闭 socketBroker 默认超时为60 秒。一个合格的 Pulsar 客户端实现不要求主动发送Ping探测但收到 Broker 的Ping后必须及时回复Pong以免远端强制关闭 TCP 连接。源码印证Broker 侧的超时配置来自keepAliveIntervalSeconds——BrokerService.java 中通过pulsar.getConfiguration().getKeepAliveIntervalSeconds()读取该配置并在 ServerCnx.java 中以该秒数周期调度 Ping 探测任务。CommandPing/CommandPong在 PulsarApi.proto 中定义为空消息仅凭BaseCommand.type即可区分。生产者Producer为发送消息客户端需要先创建一个生产者。创建生产者时Broker 首先验证该客户端是否有权在该 Topic 上发布。客户端收到生产者创建成功的确认后即可引用此前协商好的生产者 idproducer_id向 Broker 发布消息。CommandProducer创建生产者message CommandProducer { topic : persistent://my-property/my-cluster/my-namespace/my-topic, producer_id : 1, request_id : 1 }参数说明topic→ 要在其上创建生产者的完整 Topic 名称producer_id→ 客户端生成的生产者标识需在同一条连接内唯一request_id→ 该请求的标识用于将响应与原始请求匹配需在同一条连接内唯一producer_name→(可选)若指定了生产者名称则使用该名称否则由 Broker 生成唯一名称。生成的名称保证全局唯一。实现上建议生产者初次创建时让 Broker 生成新名称重连重建生产者时复用该名称Broker 会以ProducerSuccess或Error命令回复。CommandProducerSuccess创建成功message CommandProducerSuccess { request_id : 1, producer_name : generated-unique-producer-name }参数说明request_id→ 原始CreateProducer请求的 idproducer_name→ 生成的全局唯一生产者名称或客户端指定的名称若有源码扩充CommandProducerSuccess 后续版本还包含last_sequence_id配合去重功能返回上一次会话最后存储的序列号、schema_version、topic_epoch用于排他生产者接管后的连接篱笆与producer_ready字段CommandProducerproto 定义也扩展了producer_access_modeShared / Exclusive / WaitForExclusive、txn_enabled事务生产者等能力。CommandSend发送消息Send命令用于在已存在的生产者上下文中发布新消息。该命令使用携带命令与消息负载的帧传输完整格式见上文负载命令布局。message CommandSend { producer_id : 1, sequence_id : 0, num_messages : 1 }参数说明producer_id→ 已存在生产者的 idsequence_id→ 每条消息关联一个序列号预期用从 0 开始的计数器实现。确认消息有效发布的SendReceipt会以该序列号引用对应消息num_messages→(可选)一次发布一批消息时使用CommandSendReceipt发送回执消息按配置的副本数持久化完成后Broker 会向生产者发送确认回执。message CommandSendReceipt { producer_id : 1, sequence_id : 0, message_id : { ledgerId : 123, entryId : 456 } }参数说明producer_id→ 发起发送请求的生产者 idsequence_id→ 已发布消息的序列号message_id→ 系统为已发布消息分配的消息 id在单个集群内唯一。消息 id 由两个 long 组成ledgerId与entryId反映该唯一 id 是在追加到 BookKeeper ledger 时分配的CommandCloseProducer关闭生产者注意此命令可由生产者或 Broker任一方发送。收到CloseProducer命令后Broker 将停止接收该生产者的更多消息等待所有待处理消息持久化完成后再向客户端回复Success。Broker 可在优雅故障转移时向客户端发送CloseProducer例如 Broker 重启或负载均衡器卸载 Topic 以将其转移到其他 Broker 时。客户端收到CloseProducer后预期重新走一次服务发现查找lookup然后重建生产者。此过程不影响 TCP 连接本身。消费者Consumer消费者用于挂载到一个订阅subscription上并从中消费消息。每次重连后客户端都需要重新订阅 Topic若订阅尚不存在则会新建一个。流控Flow control消费者就绪后客户端需要授权Broker 推送消息这是通过Flow命令完成的。Flow命令授予 Broker 额外的**许可permits**来向消费者发送消息。典型的消费者实现会用队列累积这些消息等待应用准备就绪后再消费。当应用从队列中出队了一半消息后消费者便向 Broker 发送许可数量等于队列中消息数的一半以请求更多消息。例如队列大小为 1000消费者消费了队列中的 500 条消息则消费者向 Broker 发送许可请求 500 条消息。CommandSubscribe订阅message CommandSubscribe { topic : persistent://my-property/my-cluster/my-namespace/my-topic, subscription : my-subscription-name, subType : Exclusive, consumer_id : 1, request_id : 1 }参数说明topic→ 要在其上创建消费者的完整 Topic 名称subscription→ 订阅名称subType→ 订阅类型Exclusive、Shared、Failover、Key_Sharedconsumer_id→ 客户端生成的消费者标识需在同一条连接内唯一request_id→ 请求标识用于匹配响应与原始请求需在同一条连接内唯一consumer_name→(可选)客户端可指定消费者名称用于在统计信息中跟踪特定消费者在Failover订阅类型下该名称用于决定哪个消费者被选为master接收消息者消费者按名称排序第一个被选为 master源码扩充CommandSubscribe 中SubType枚举值与文档一致Exclusive0、Shared1、Failover2、Key_Shared3。后续版本还加入了durable订阅是否由持久化游标支撑、start_message_id/start_message_rollback_duration_sec指定起始消费位置、initialPositionLatest / Earliest、read_compacted读取 compacted Topic、replicate_subscription_state跨集群复制订阅状态以及keySharedMeta等字段。CommandFlow授予许可message CommandFlow { consumer_id : 1, messagePermits : 1000 }参数说明consumer_id→ 已建立消费者的 idmessagePermits→ 授予 Broker 用于推送更多消息的额外许可数量源码印证CommandFlow 的注释明确写着Max number of messages to prefetch, in addition of any number previously specified即许可具有累加语义与文档描述的增量授予一致。CommandMessage投递消息Message命令由 Broker 用于在给定许可额度内将消息推送给已存在的消费者。该命令同样使用携带消息负载的帧传输完整格式见负载命令布局。message CommandMessage { consumer_id : 1, message_id : { ledgerId : 123, entryId : 456 } }源码扩充CommandMessage 还包含redelivery_count重投递次数配合负向确认/ack 超时机制与consumer_epoch字段。CommandAck确认Ack用于向 Broker 表明某条消息已被应用成功处理、可以丢弃。此外Broker 也会基于已确认消息维护消费者的消费位置cursor。message CommandAck { consumer_id : 1, ack_type : Individual, message_id : { ledgerId : 123, entryId : 456 } }参数说明consumer_id→ 已建立消费者的 idack_type→ 确认类型Individual单条确认或Cumulative累积确认message_id→ 要确认的消息 idvalidation_error→(可选)表明消费者已丢弃消息原因为UncompressedSizeCorruption、DecompressionError、ChecksumMismatch、BatchDeSerializeError源码扩充CommandAck 在Individual类型下支持传递消息 id 列表repeated MessageIdData message_id 3;ValidationError枚举后续还增加了DecryptionError并扩展了事务相关字段txnid_least_bits/txnid_most_bits与request_id用于 AckResponse 应答。CommandCloseConsumer关闭消费者注意此命令可由生产者或 Broker任一方发送。行为与CloseProducer相同。CommandRedeliverUnacknowledgedMessages重投递未确认消息消费者可以请求 Broker 重投递已推送给该消费者但尚未确认的部分或全部待处理消息。protobuf 对象接受消费者希望重投递的消息 id 列表。若列表为空Broker 将重投递所有待处理消息。重投递时消息可发送给同一个消费者在共享Shared订阅场景下也可分摊到所有可用消费者。源码印证CommandRedeliverUnacknowledgedMessages 定义repeated MessageIdData message_ids 2;与空列表即全部重投递的语义对应。CommandReachedEndOfTopic到达 Topic 末尾当 Topic 已被终止terminated且订阅上的所有消息均已确认时Broker 会向特定消费者发送该命令。客户端应使用此命令通知应用该消费者不会再收到更多消息。CommandConsumerStats消费者统计该命令由客户端发送用于从 Broker 获取订阅Subscriber级与消费者Consumer级统计信息。参数说明request_id→ 请求 id用于关联请求与响应consumer_id→ 已建立消费者的 idCommandConsumerStatsResponse统计响应这是 Broker 对客户端ConsumerStats请求的响应包含请求中consumer_id的订阅级与消费者级统计信息。若设置了error_code或error_message字段则表示请求失败。源码扩充CommandConsumerStatsResponse 携带的统计项包括msgRateOut投递速率msg/s、msgThroughputOut吞吐bytes/s、msgRateRedeliver重投递速率、availablePermits可用许可数、unackedMessages未确认消息数、blockedConsumerOnUnackedMsgs是否因未确认消息超阈值被阻塞、msgBacklog订阅积压量、messageAckRate确认速率等可用于诊断消费链路健康度。CommandUnsubscribe取消订阅该命令由客户端发送将consumer_id从关联 Topic 上取消订阅。参数说明request_id→ 请求 idconsumer_id→ 需取消订阅的已建立消费者的 id服务发现Service discoveryTopic 查找Topic lookup客户端每次创建或重连生产者/消费者时都需要执行 Topic 查找lookup以发现当前由哪个 Broker提供该 Topic 的服务。查找可通过 REST 调用完成详见管理 API 中的 Topic 查找文档2.2.1 版本中对应admin-api-persistent-topics.md的Lookup of topic小节示例为pulsar-admin topics lookup。自 Pulsar 1.16 起也可以在二进制协议内执行查找。为便于说明假设服务发现组件运行在pulsar://broker.example.com:6650各 Broker 分别运行在pulsar://broker-1.example.com:6650、pulsar://broker-2.example.com:6650等地址。客户端可使用到发现服务主机的连接发起LookupTopic命令。响应可以是应连接的 Broker 主机名或需要重试查找的 Broker 主机名。LookupTopic命令必须用于已完成Connect/Connected初始握手的连接上。message CommandLookupTopic { topic : persistent://my-property/my-cluster/my-namespace/my-topic, request_id : 1, authoritative : false }字段说明topic→ 要查找的 Topic 名称request_id→ 请求 id会随其响应一起返回authoritative→ 首次查找请求应使用false。跟随重定向redirect响应时客户端应传递响应中携带的相同值LookupTopicResponse查找响应查找成功的响应示例message CommandLookupTopicResponse { request_id : 1, response : Connect, brokerServiceUrl : pulsar://broker-1.example.com:6650, brokerServiceUrlTls : pulsarssl://broker-1.example.com:6651, authoritative : true }带重定向的查找响应示例message CommandLookupTopicResponse { request_id : 1, response : Redirect, brokerServiceUrl : pulsar://broker-2.example.com:6650, brokerServiceUrlTls : pulsarssl://broker-2.example.com:6651, authoritative : true }在第二种情况下需要向broker-2.example.com重新发起LookupTopic请求由该 Broker 给出查找的最终确定答案。源码印证CommandLookupTopicResponse 中LookupType枚举为Redirect0、Connect1、Failed2响应同时携带明文brokerServiceUrl与 TLSbrokerServiceUrlTls两种服务地址。CommandLookupTopicproto 定义后续版本还增加了advertised_listener_name字段用于多监听器multi-listener场景下指定要连接的 listener 名称。分区 Topic 发现Partitioned topics discovery分区 Topic 元数据发现用于确定某个 Topic 是否为分区 Topic以及配置了多少个分区。若 Topic 标记为已分区客户端预期为每个分区创建一个生产者或消费者并使用partition-X后缀如my-topic-partition-0、my-topic-partition-1。该信息只需在首次创建生产者或消费者时获取重连后无需再次获取。分区 Topic 元数据发现的工作方式与 Topic 查找非常相似客户端向服务发现地址发送请求响应中包含实际元数据。CommandPartitionedTopicMetadata查询分区元数据message CommandPartitionedTopicMetadata { topic : persistent://my-property/my-cluster/my-namespace/my-topic, request_id : 1 }字段说明topic→ 要检查分区元数据的 Topicrequest_id→ 请求 id会随其响应一起返回CommandPartitionedTopicMetadataResponse分区元数据响应带元数据的响应示例message CommandPartitionedTopicMetadataResponse { request_id : 1, response : Success, partitions : 32 }源码印证CommandPartitionedTopicMetadataResponse 中LookupType枚举为Success0/Failed1错误场景下通过errorServerError枚举与message字段携带失败原因。协议版本演进ProtocolVersion连接握手中的protocol_version字段即取自ProtocolVersion枚举PulsarApi.proto每新增协议特性即递增一个版本号。结合上文各命令的出现时间可以清晰看到协议能力的演进脉络版本新增能力v0初始版本v1应用层保活keep-alivev2RedeliverUnacknowledgedMessages命令v3LZ4 与 ZLib 压缩v4批量消息支持v5不断开连接即可断开客户端v6元数据 负载的校验和计算v7LookupTopic二进制查找v8ConsumerStats消费者统计v9Topic 末尾通知ReachedEndOfTopicv10代理到 Brokerproxyv11修复 C 消费者对校验和字段的处理v12获取 Topic 最后一条消息 id、ActiveConsumerChange、GetTopicsOfNamespacev13Schema 注册JSON 使用 Avro schema 格式v14AuthChallenge/AuthResponse双向认证Key_Shared 订阅v15GetOrCreateSchema相关命令v16Broker entry metadata 支持v17Ack receipt 支持v18客户端支持 broker entry metadatav19事务协调器TC客户端连接相关命令这也解释了上文负载命令中 magic number 与 checksum 的存在它们在 v6 引入而低于 v11 的旧版 C 客户端存在校验和处理问题——协议版本协商正是为了让新旧客户端能够安全共存。Protobuf 接口Pulsar 的全部 protobuf 定义都集中在仓库的 pulsar-common/src/main/proto/PulsarApi.proto 文件中。该文件声明了syntax proto2包名为pulsar.protoJava 生成包为org.apache.pulsar.common.api.proto并采用LITE_RUNTIME优化以降低运行时开销。其中值得重点阅读的定义包括BaseCommand及其Type枚举第 896 行起全部命令的分发入口MessageMetadata/SingleMessageMetadata第 104 行起单条与批量消息的元数据结构CompressionType、ServerError、ProtocolVersion、KeySharedMode第 90-330 行压缩算法、服务端错误码、协议版本与 Key_Shared 模式各CommandXxx消息连接、生产、消费、服务发现、事务等全部交互命令协议在服务端的序列化/反序列化入口位于 pulsar-common/src/main/java/org/apache/pulsar/common/protocol/Commands.java如magicCrc32c、serializeCommandSendWithSize、ChecksumType等常量与方法客户端侧各语言实现Java、C、Go、Python均以本 proto 文件为蓝本生成对应语言的编解码代码。若要自行实现或深度调试一个 Pulsar 客户端/代理这份 proto 文件加上本文的帧结构说明就是最权威的协议参考。赞分享消息队列后端流处理【免费下载链接】pulsarApache Pulsar - distributed pub-sub messaging system项目地址https://gitcode.com/gh_mirrors/pulsar28/pulsar点击查看免费下载相关推荐Apache Pulsar 二进制协议Binary Protocol深度解析从帧格式到命令交互全指南Apache Pulsar 二进制协议Binary Protocol深度解析从帧格式到命令交互全指南 output_article Apache Pul消息队列后端流处理Apache Pulsar 二进制协议规范深度解析帧格式、命令交互与服务发现Apache Pulsar 二进制协议规范深度解析帧格式、命令交互与服务发现 Pulsar 的生产者/消费者与 Broker 之间通过一套自定义的二进制协议进消息队列后端流处理Apache Pulsar 二进制协议Binary Protocol深度解析命令帧格式、消息元数据与交互流程Apache Pulsar 二进制协议Binary Protocol深度解析命令帧格式、消息元数据与交互流程 本文以 Apache Pulsar 2.1.消息队列后端流处理上一篇Flet v0.24.0 版本发布详解新控件、破坏性变更与迁移指南下一篇NPU与CPU双平台部署nest_base_jx.goog_in1k环境配置与性能对比实测创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 
长河君 · 2026/9/25 5:28:27 · 阅读 1.2k Apache Pulsar 二进制协议（Binary Protocol）深度解析：帧结构、命令交互与服务发现全指南
消息队列后端流处理【免费下载链接】pulsarApache Pulsar - distributed pub-sub messaging system项目地址https://gitcode.com/gh_mirrors/pulsar28/pulsar点击查看免费下载导读本文基于 Apache Pulsar 2.2.1 版本文档完整剖析 Pulsar 客户端与 Broker 之间通信所依赖的自定义二进制协议。协议以 Protocol Buffersprotobuf承载命令、以 4 字节长度前缀完成帧定界并支持基于令牌permit的流控、批量消息、校验和与服务发现。读完本文你将掌握 Pulsar 协议帧的字节级布局、消息元数据字段语义、生产者/消费者全交互流程以及二进制协议下的 Topic 查找机制并能在仓库源码pulsar-common/src/main/proto/PulsarApi.proto、Commands.java中定位对应实现加以印证。协议总体设计为什么 Pulsar 需要一套自定义二进制协议Pulsar 在生产者/消费者与 Broker 之间使用一套自定义二进制协议进行通信。该协议设计目标是在支持确认acknowledgement与流控flow control等必需特性的同时获得最大的传输效率与实现效率。相比文本协议二进制 protobuf 的组合在序列化开销、带宽占用与解析性能上都有显著优势。协议中客户端与 Broker 之间交换的是命令Command。每条命令都是一个二进制 protobuf 消息其格式定义在 pulsar-common/src/main/proto/PulsarApi.proto 文件中文末Protobuf 接口一节会进一步说明。所有命令统一封装在一个BaseCommandprotobuf 消息中。BaseCommand内含一个Type枚举枚举值对应全部可能的子命令BaseCommand中每个子命令都以 optional 字段形式挂载且一条BaseCommand只能携带一个子命令。连接共享Connection sharing不同生产者、不同消费者的命令可以在同一条 TCP 连接上自由交错发送不受任何限制。这使得一个客户端只需维护少量连接即可驱动大量 Topic 的生产与消费。BaseCommand 与 Type 枚举在 PulsarApi.proto 中BaseCommand通过required Type type 1;声明当前帧携带哪种命令随后将对应子命令作为 optional 字段挂载例如optional CommandSend send 6;、optional CommandAck ack 10;。Type枚举覆盖了连接CONNECT/CONNECTED、订阅SUBSCRIBE、生产PRODUCER/SEND/SEND_RECEIPT、消费MESSAGE/ACK/FLOW、服务发现LOOKUP/PARTITIONED_METADATA、保活PING/PONG以及事务NEW_TXN等 50 号段等数十种命令类型。从源码结构看BaseCommand的单子命令约束保证了帧语义的确定性让接收方可以无歧义地按type分发到对应处理逻辑。帧格式Framing由于 protobuf 本身不提供消息帧定界Pulsar 协议中所有消息都在开头追加一个 4 字节字段用于指明该帧的大小frame size。单个帧允许的最大尺寸为5 MB——这一限制在源码中有直接对应pulsar-common/src/main/java/org/apache/pulsar/common/protocol/Commands.java第 115 行定义了public static final int DEFAULT_MAX_MESSAGE_SIZE 5 * 1024 * 1024;。协议支持两类命令简单命令Simple commands不携带消息负载payload。负载命令Payload commands携带负载用于消息发布或投递。负载命令中protobuf 命令数据之后紧跟着 protobuf 序列化的消息元数据再之后是负载负载以原始raw二进制格式在 protobuf 之外传递。所有大小字段均以4 字节无符号大端big endian整数传递。消息负载之所以以原始格式而非 protobuf 格式传递是为了效率考虑——避免对用户数据做无意义的二次序列化。简单命令布局简单无负载命令的基本结构如下组件说明大小字节totalSize帧的大小统计其之后所有内容的字节数4commandSizeprotobuf 序列化后命令的大小4message以原始二进制格式序列化的 protobuf 消息可变负载命令布局负载命令的基本结构如下组件说明大小字节totalSize帧的大小统计其之后所有内容的字节数4commandSizeprotobuf 序列化后命令的大小4message以原始二进制格式序列化的 protobuf 消息可变magicNumber2 字节标识当前格式的字节数组0x0e012checksum对其后所有内容的 CRC32-C 校验和4metadataSize消息元数据的大小4metadata以二进制 protobuf 消息存储的消息元数据可变payload帧中剩余的所有字节均视为负载可为任意字节序列可变源码印证Commands.java第 124 行定义了public static final short magicCrc32c 0x0e01;与文档描述的 magic number 完全一致第 1481 行附近以注释形式给出了负载命令的序列化布局[TOTAL_SIZE] [CMD_SIZE][CMD] [MAGIC_NUMBER][CHECKSUM] [METADATA_SIZE][METADATA] [PAYLOAD]。校验和计算通过com.scurrilous.circe.checksum.Crc32cIntChecksumcomputeChecksum/resumeChecksum实现并且ChecksumType枚举Crc32c/None允许在不需要校验时跳过 magic checksum 两段对应旧版本协议的行为详见文末协议版本演进。消息元数据Message Metadata消息元数据以序列化 protobuf 消息的形式与应用负载payload一并存储。元数据由生产者创建并在投递时原样传给消费者。字段说明producer_name发布该消息的生产者名称sequence_id消息的序列号由生产者分配publish_time发布时间戳Unix 时间即自 1970 年 1 月 1 日 UTC 以来的毫秒数properties一组键值对使用KeyValue消息。这些是应用自定义的键与值Pulsar 不赋予其特殊含义replicated_from(可选)表明消息已被复制并指明消息最初发布所在的集群名称partition_key(可选)在分区 Topic 上发布时若存在 key则使用该 key 的哈希决定选择哪个分区compression(可选)表明负载已压缩以及使用的压缩库uncompressed_size(可选)若使用了压缩生产者必须用原始负载大小填充该字段num_messages_in_batch(可选)若该消息实际上是批量消息则必须设为批量中消息的数量源码扩充在 PulsarApi.proto 的MessageMetadata定义中可以看到这些字段的完整 protobuf 声明例如required string producer_name 1;、required uint64 sequence_id 2;、required uint64 publish_time 3;。此外compression字段对应CompressionType枚举proto 定义枚举值含义NONE未压缩LZ4LZ4 压缩ZLIBZLib 压缩ZSTDZStandard 压缩SNAPPYSnappy 压缩在 2.2.1 之后的版本中MessageMetadata还扩展了event_time应用事件时间戳、encryption_keys/encryption_algo/encryption_param端到端加密相关、ordering_keyKey_Shared 模式下覆盖消息排序键、deliver_at_time延迟投递、marker_type内部 marker 消息等字段可结合仓库中的 proto 定义继续查阅。批量消息Batch Messages使用批量消息时负载中包含一组条目entries每个条目拥有各自的元数据由SingleMessageMetadata对象定义。单个批次的负载格式如下字段说明metadataSizeN单条消息元数据序列化为 protobuf 后的大小metadataN单条消息元数据payloadN应用传入的消息负载其中每条元数据字段如下字段说明properties应用定义的属性partition key(可选)用于指示哈希到特定分区的 keypayload_size批量中单条消息的负载大小当启用压缩时整个批次会被一次性整体压缩而非逐条压缩从而获得更高的压缩比。源码印证SingleMessageMetadata的 protobuf 定义位于 PulsarApi.proto其中required int32 payload_size 3;与文档表格一一对应。在BaseCommand.Type中SEND命令的num_messages字段CommandSend 定义即用于一次发送整批消息默认值为 1。交互流程Interactions连接建立Connection establishment在与 Broker 建立 TCP 连接后通常端口为6650由客户端负责发起会话。收到 Broker 的Connected响应后客户端即可认为连接可用。反之若 Broker 无法通过客户端认证则会回复Error命令并关闭 TCP 连接。客户端发起连接时发送的CommandConnect示例message CommandConnect { client_version : Pulsar-Client-Java-v1.15.2, auth_method_name : my-authentication-plugin, auth_data : my-auth-data, protocol_version : 6 }字段说明client_version→ 基于字符串的标识符格式不受强制auth_method_name→(可选)启用认证时使用的认证插件名称auth_data→(可选)插件特定的认证数据protocol_version→ 客户端支持的协议版本。Broker 不会发送更新协议版本中引入的命令且 Broker 可能强制一个最低版本Broker 回复的CommandConnected示例message CommandConnected { server_version : Pulsar-Broker-v1.15.2, protocol_version : 6 }字段说明server_version→ Broker 版本的字符串标识protocol_version→ Broker 支持的协议版本。客户端不得尝试发送更新协议版本中引入的命令源码扩充这两个命令的 protobuf 定义在 PulsarApi.proto 中。CommandConnect在后续版本中还加入了proxy_to_broker_url通过 Pulsar Proxy 中转、original_principal/original_auth_data代理场景下携带原始身份信息以及FeatureFlags如supports_auth_refresh等字段CommandConnected则增加了max_message_size用于通知客户端 Broker 允许的最大消息大小与上文 5 MB 默认值相关。保活机制Keep Alive为识别客户端与 Broker 之间长时间的网络分区或对端机器崩溃但未中断 TCP 连接的情况如断电、内核 panic、硬重启等协议引入了探测远端存活状态的机制。客户端与 Broker周期性发送Ping命令若在超时时间内未收到Pong响应则关闭 socketBroker 默认超时为60 秒。一个合格的 Pulsar 客户端实现不要求主动发送Ping探测但收到 Broker 的Ping后必须及时回复Pong以免远端强制关闭 TCP 连接。源码印证Broker 侧的超时配置来自keepAliveIntervalSeconds——BrokerService.java 中通过pulsar.getConfiguration().getKeepAliveIntervalSeconds()读取该配置并在 ServerCnx.java 中以该秒数周期调度 Ping 探测任务。CommandPing/CommandPong在 PulsarApi.proto 中定义为空消息仅凭BaseCommand.type即可区分。生产者Producer为发送消息客户端需要先创建一个生产者。创建生产者时Broker 首先验证该客户端是否有权在该 Topic 上发布。客户端收到生产者创建成功的确认后即可引用此前协商好的生产者 idproducer_id向 Broker 发布消息。CommandProducer创建生产者message CommandProducer { topic : persistent://my-property/my-cluster/my-namespace/my-topic, producer_id : 1, request_id : 1 }参数说明topic→ 要在其上创建生产者的完整 Topic 名称producer_id→ 客户端生成的生产者标识需在同一条连接内唯一request_id→ 该请求的标识用于将响应与原始请求匹配需在同一条连接内唯一producer_name→(可选)若指定了生产者名称则使用该名称否则由 Broker 生成唯一名称。生成的名称保证全局唯一。实现上建议生产者初次创建时让 Broker 生成新名称重连重建生产者时复用该名称Broker 会以ProducerSuccess或Error命令回复。CommandProducerSuccess创建成功message CommandProducerSuccess { request_id : 1, producer_name : generated-unique-producer-name }参数说明request_id→ 原始CreateProducer请求的 idproducer_name→ 生成的全局唯一生产者名称或客户端指定的名称若有源码扩充CommandProducerSuccess 后续版本还包含last_sequence_id配合去重功能返回上一次会话最后存储的序列号、schema_version、topic_epoch用于排他生产者接管后的连接篱笆与producer_ready字段CommandProducerproto 定义也扩展了producer_access_modeShared / Exclusive / WaitForExclusive、txn_enabled事务生产者等能力。CommandSend发送消息Send命令用于在已存在的生产者上下文中发布新消息。该命令使用携带命令与消息负载的帧传输完整格式见上文负载命令布局。message CommandSend { producer_id : 1, sequence_id : 0, num_messages : 1 }参数说明producer_id→ 已存在生产者的 idsequence_id→ 每条消息关联一个序列号预期用从 0 开始的计数器实现。确认消息有效发布的SendReceipt会以该序列号引用对应消息num_messages→(可选)一次发布一批消息时使用CommandSendReceipt发送回执消息按配置的副本数持久化完成后Broker 会向生产者发送确认回执。message CommandSendReceipt { producer_id : 1, sequence_id : 0, message_id : { ledgerId : 123, entryId : 456 } }参数说明producer_id→ 发起发送请求的生产者 idsequence_id→ 已发布消息的序列号message_id→ 系统为已发布消息分配的消息 id在单个集群内唯一。消息 id 由两个 long 组成ledgerId与entryId反映该唯一 id 是在追加到 BookKeeper ledger 时分配的CommandCloseProducer关闭生产者注意此命令可由生产者或 Broker任一方发送。收到CloseProducer命令后Broker 将停止接收该生产者的更多消息等待所有待处理消息持久化完成后再向客户端回复Success。Broker 可在优雅故障转移时向客户端发送CloseProducer例如 Broker 重启或负载均衡器卸载 Topic 以将其转移到其他 Broker 时。客户端收到CloseProducer后预期重新走一次服务发现查找lookup然后重建生产者。此过程不影响 TCP 连接本身。消费者Consumer消费者用于挂载到一个订阅subscription上并从中消费消息。每次重连后客户端都需要重新订阅 Topic若订阅尚不存在则会新建一个。流控Flow control消费者就绪后客户端需要授权Broker 推送消息这是通过Flow命令完成的。Flow命令授予 Broker 额外的**许可permits**来向消费者发送消息。典型的消费者实现会用队列累积这些消息等待应用准备就绪后再消费。当应用从队列中出队了一半消息后消费者便向 Broker 发送许可数量等于队列中消息数的一半以请求更多消息。例如队列大小为 1000消费者消费了队列中的 500 条消息则消费者向 Broker 发送许可请求 500 条消息。CommandSubscribe订阅message CommandSubscribe { topic : persistent://my-property/my-cluster/my-namespace/my-topic, subscription : my-subscription-name, subType : Exclusive, consumer_id : 1, request_id : 1 }参数说明topic→ 要在其上创建消费者的完整 Topic 名称subscription→ 订阅名称subType→ 订阅类型Exclusive、Shared、Failover、Key_Sharedconsumer_id→ 客户端生成的消费者标识需在同一条连接内唯一request_id→ 请求标识用于匹配响应与原始请求需在同一条连接内唯一consumer_name→(可选)客户端可指定消费者名称用于在统计信息中跟踪特定消费者在Failover订阅类型下该名称用于决定哪个消费者被选为master接收消息者消费者按名称排序第一个被选为 master源码扩充CommandSubscribe 中SubType枚举值与文档一致Exclusive0、Shared1、Failover2、Key_Shared3。后续版本还加入了durable订阅是否由持久化游标支撑、start_message_id/start_message_rollback_duration_sec指定起始消费位置、initialPositionLatest / Earliest、read_compacted读取 compacted Topic、replicate_subscription_state跨集群复制订阅状态以及keySharedMeta等字段。CommandFlow授予许可message CommandFlow { consumer_id : 1, messagePermits : 1000 }参数说明consumer_id→ 已建立消费者的 idmessagePermits→ 授予 Broker 用于推送更多消息的额外许可数量源码印证CommandFlow 的注释明确写着Max number of messages to prefetch, in addition of any number previously specified即许可具有累加语义与文档描述的增量授予一致。CommandMessage投递消息Message命令由 Broker 用于在给定许可额度内将消息推送给已存在的消费者。该命令同样使用携带消息负载的帧传输完整格式见负载命令布局。message CommandMessage { consumer_id : 1, message_id : { ledgerId : 123, entryId : 456 } }源码扩充CommandMessage 还包含redelivery_count重投递次数配合负向确认/ack 超时机制与consumer_epoch字段。CommandAck确认Ack用于向 Broker 表明某条消息已被应用成功处理、可以丢弃。此外Broker 也会基于已确认消息维护消费者的消费位置cursor。message CommandAck { consumer_id : 1, ack_type : Individual, message_id : { ledgerId : 123, entryId : 456 } }参数说明consumer_id→ 已建立消费者的 idack_type→ 确认类型Individual单条确认或Cumulative累积确认message_id→ 要确认的消息 idvalidation_error→(可选)表明消费者已丢弃消息原因为UncompressedSizeCorruption、DecompressionError、ChecksumMismatch、BatchDeSerializeError源码扩充CommandAck 在Individual类型下支持传递消息 id 列表repeated MessageIdData message_id 3;ValidationError枚举后续还增加了DecryptionError并扩展了事务相关字段txnid_least_bits/txnid_most_bits与request_id用于 AckResponse 应答。CommandCloseConsumer关闭消费者注意此命令可由生产者或 Broker任一方发送。行为与CloseProducer相同。CommandRedeliverUnacknowledgedMessages重投递未确认消息消费者可以请求 Broker 重投递已推送给该消费者但尚未确认的部分或全部待处理消息。protobuf 对象接受消费者希望重投递的消息 id 列表。若列表为空Broker 将重投递所有待处理消息。重投递时消息可发送给同一个消费者在共享Shared订阅场景下也可分摊到所有可用消费者。源码印证CommandRedeliverUnacknowledgedMessages 定义repeated MessageIdData message_ids 2;与空列表即全部重投递的语义对应。CommandReachedEndOfTopic到达 Topic 末尾当 Topic 已被终止terminated且订阅上的所有消息均已确认时Broker 会向特定消费者发送该命令。客户端应使用此命令通知应用该消费者不会再收到更多消息。CommandConsumerStats消费者统计该命令由客户端发送用于从 Broker 获取订阅Subscriber级与消费者Consumer级统计信息。参数说明request_id→ 请求 id用于关联请求与响应consumer_id→ 已建立消费者的 idCommandConsumerStatsResponse统计响应这是 Broker 对客户端ConsumerStats请求的响应包含请求中consumer_id的订阅级与消费者级统计信息。若设置了error_code或error_message字段则表示请求失败。源码扩充CommandConsumerStatsResponse 携带的统计项包括msgRateOut投递速率msg/s、msgThroughputOut吞吐bytes/s、msgRateRedeliver重投递速率、availablePermits可用许可数、unackedMessages未确认消息数、blockedConsumerOnUnackedMsgs是否因未确认消息超阈值被阻塞、msgBacklog订阅积压量、messageAckRate确认速率等可用于诊断消费链路健康度。CommandUnsubscribe取消订阅该命令由客户端发送将consumer_id从关联 Topic 上取消订阅。参数说明request_id→ 请求 idconsumer_id→ 需取消订阅的已建立消费者的 id服务发现Service discoveryTopic 查找Topic lookup客户端每次创建或重连生产者/消费者时都需要执行 Topic 查找lookup以发现当前由哪个 Broker提供该 Topic 的服务。查找可通过 REST 调用完成详见管理 API 中的 Topic 查找文档2.2.1 版本中对应admin-api-persistent-topics.md的Lookup of topic小节示例为pulsar-admin topics lookup。自 Pulsar 1.16 起也可以在二进制协议内执行查找。为便于说明假设服务发现组件运行在pulsar://broker.example.com:6650各 Broker 分别运行在pulsar://broker-1.example.com:6650、pulsar://broker-2.example.com:6650等地址。客户端可使用到发现服务主机的连接发起LookupTopic命令。响应可以是应连接的 Broker 主机名或需要重试查找的 Broker 主机名。LookupTopic命令必须用于已完成Connect/Connected初始握手的连接上。message CommandLookupTopic { topic : persistent://my-property/my-cluster/my-namespace/my-topic, request_id : 1, authoritative : false }字段说明topic→ 要查找的 Topic 名称request_id→ 请求 id会随其响应一起返回authoritative→ 首次查找请求应使用false。跟随重定向redirect响应时客户端应传递响应中携带的相同值LookupTopicResponse查找响应查找成功的响应示例message CommandLookupTopicResponse { request_id : 1, response : Connect, brokerServiceUrl : pulsar://broker-1.example.com:6650, brokerServiceUrlTls : pulsarssl://broker-1.example.com:6651, authoritative : true }带重定向的查找响应示例message CommandLookupTopicResponse { request_id : 1, response : Redirect, brokerServiceUrl : pulsar://broker-2.example.com:6650, brokerServiceUrlTls : pulsarssl://broker-2.example.com:6651, authoritative : true }在第二种情况下需要向broker-2.example.com重新发起LookupTopic请求由该 Broker 给出查找的最终确定答案。源码印证CommandLookupTopicResponse 中LookupType枚举为Redirect0、Connect1、Failed2响应同时携带明文brokerServiceUrl与 TLSbrokerServiceUrlTls两种服务地址。CommandLookupTopicproto 定义后续版本还增加了advertised_listener_name字段用于多监听器multi-listener场景下指定要连接的 listener 名称。分区 Topic 发现Partitioned topics discovery分区 Topic 元数据发现用于确定某个 Topic 是否为分区 Topic以及配置了多少个分区。若 Topic 标记为已分区客户端预期为每个分区创建一个生产者或消费者并使用partition-X后缀如my-topic-partition-0、my-topic-partition-1。该信息只需在首次创建生产者或消费者时获取重连后无需再次获取。分区 Topic 元数据发现的工作方式与 Topic 查找非常相似客户端向服务发现地址发送请求响应中包含实际元数据。CommandPartitionedTopicMetadata查询分区元数据message CommandPartitionedTopicMetadata { topic : persistent://my-property/my-cluster/my-namespace/my-topic, request_id : 1 }字段说明topic→ 要检查分区元数据的 Topicrequest_id→ 请求 id会随其响应一起返回CommandPartitionedTopicMetadataResponse分区元数据响应带元数据的响应示例message CommandPartitionedTopicMetadataResponse { request_id : 1, response : Success, partitions : 32 }源码印证CommandPartitionedTopicMetadataResponse 中LookupType枚举为Success0/Failed1错误场景下通过errorServerError枚举与message字段携带失败原因。协议版本演进ProtocolVersion连接握手中的protocol_version字段即取自ProtocolVersion枚举PulsarApi.proto每新增协议特性即递增一个版本号。结合上文各命令的出现时间可以清晰看到协议能力的演进脉络版本新增能力v0初始版本v1应用层保活keep-alivev2RedeliverUnacknowledgedMessages命令v3LZ4 与 ZLib 压缩v4批量消息支持v5不断开连接即可断开客户端v6元数据 负载的校验和计算v7LookupTopic二进制查找v8ConsumerStats消费者统计v9Topic 末尾通知ReachedEndOfTopicv10代理到 Brokerproxyv11修复 C 消费者对校验和字段的处理v12获取 Topic 最后一条消息 id、ActiveConsumerChange、GetTopicsOfNamespacev13Schema 注册JSON 使用 Avro schema 格式v14AuthChallenge/AuthResponse双向认证Key_Shared 订阅v15GetOrCreateSchema相关命令v16Broker entry metadata 支持v17Ack receipt 支持v18客户端支持 broker entry metadatav19事务协调器TC客户端连接相关命令这也解释了上文负载命令中 magic number 与 checksum 的存在它们在 v6 引入而低于 v11 的旧版 C 客户端存在校验和处理问题——协议版本协商正是为了让新旧客户端能够安全共存。Protobuf 接口Pulsar 的全部 protobuf 定义都集中在仓库的 pulsar-common/src/main/proto/PulsarApi.proto 文件中。该文件声明了syntax proto2包名为pulsar.protoJava 生成包为org.apache.pulsar.common.api.proto并采用LITE_RUNTIME优化以降低运行时开销。其中值得重点阅读的定义包括BaseCommand及其Type枚举第 896 行起全部命令的分发入口MessageMetadata/SingleMessageMetadata第 104 行起单条与批量消息的元数据结构CompressionType、ServerError、ProtocolVersion、KeySharedMode第 90-330 行压缩算法、服务端错误码、协议版本与 Key_Shared 模式各CommandXxx消息连接、生产、消费、服务发现、事务等全部交互命令协议在服务端的序列化/反序列化入口位于 pulsar-common/src/main/java/org/apache/pulsar/common/protocol/Commands.java如magicCrc32c、serializeCommandSendWithSize、ChecksumType等常量与方法客户端侧各语言实现Java、C、Go、Python均以本 proto 文件为蓝本生成对应语言的编解码代码。若要自行实现或深度调试一个 Pulsar 客户端/代理这份 proto 文件加上本文的帧结构说明就是最权威的协议参考。赞分享消息队列后端流处理【免费下载链接】pulsarApache Pulsar - distributed pub-sub messaging system项目地址https://gitcode.com/gh_mirrors/pulsar28/pulsar点击查看免费下载相关推荐Apache Pulsar 二进制协议Binary Protocol深度解析从帧格式到命令交互全指南Apache Pulsar 二进制协议Binary Protocol深度解析从帧格式到命令交互全指南 output_article Apache Pul消息队列后端流处理Apache Pulsar 二进制协议规范深度解析帧格式、命令交互与服务发现Apache Pulsar 二进制协议规范深度解析帧格式、命令交互与服务发现 Pulsar 的生产者/消费者与 Broker 之间通过一套自定义的二进制协议进消息队列后端流处理Apache Pulsar 二进制协议Binary Protocol深度解析命令帧格式、消息元数据与交互流程Apache Pulsar 二进制协议Binary Protocol深度解析命令帧格式、消息元数据与交互流程 本文以 Apache Pulsar 2.1.消息队列后端流处理上一篇Flet v0.24.0 版本发布详解新控件、破坏性变更与迁移指南下一篇NPU与CPU双平台部署nest_base_jx.goog_in1k环境配置与性能对比实测创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考