Apache Pulsar Go 客户端完整实战指南：生产者、消费者与读者开发详解

• 消息队列
• 后端
• 流处理

Apache Pulsar 官方为 Go（Golang）开发者提供了与 Java、C++ 客户端能力对齐的 Go 客户端库，用于创建消息生产者（Producer）、消费者（Consumer）与读者（Reader）。本指南以 Pulsar 2.3.1 版本文档为骨架，结合本仓库中 C++ 客户端源码（Go 客户端基于 C++ 客户端库封装），完整覆盖安装方式、连接 URL 格式、Client/Producer/Consumer/Reader 的全部配置参数、消息构造与 TLS 加密认证配置，并补充实现级原理说明，帮助你从零开始构建可运行的 Pulsar Go 应用。

说明：本仓库当前包含的是 Pulsar 2.3.1 时代 Go 客户端（基于 C++ 客户端库的 CGo 封装）的官方文档（见 client-libraries-go.md）。在后续版本中，Apache Pulsar 推出了独立的纯 Go 客户端（github.com/apache/pulsar-client-go）并逐步弃用 CGo 封装，本指南严格以 2.3.1 版本文档内容为准。

安装与依赖

前置要求

Pulsar Go 客户端库基于 C++ 客户端库实现，因此在安装 Go 包之前，需要先按照 C++ 客户端的安装指引安装二进制库。安装方式有三种（详见 C++ 客户端文档）：

• RPM 包：适用于基于 RPM 的 Linux 发行版；
• Deb 包：适用于 Debian/Ubuntu 系发行版；
• Homebrew 包（macOS）：brew install libpulsar一类的方式安装。

兼容性警告：Go 客户端的版本号必须与Pulsar C++ 客户端库的版本号严格一致。C++ 客户端库正是本仓库中 pulsar-client-cpp 目录对应的模块。

安装 Go 包

安装pulsarGo 库可以使用go get：

$ go get -u github.com/apache/pulsar/pulsar-client-go/pulsar

注意：go get不支持拉取指定 tag，它总是拉取 master 分支版本的 Go 客户端，因此你需要一个与 master 匹配的 C++ 客户端库。

也可以使用 dep 进行依赖管理（其中@v@pulsar:version@是版本占位符，实际使用时应替换为具体版本号）：

$ dep ensure -add github.com/apache/pulsar/pulsar-client-go/pulsar@v@pulsar:version@

安装完成后即可在项目中导入：

import "github.com/apache/pulsar/pulsar-client-go/pulsar"

连接 URL

使用任何 Pulsar 客户端库连接集群，都需要指定一个 Pulsar 二进制协议 URL。Pulsar 协议 URL 归属于特定集群，使用pulsarscheme，默认端口为6650。本机示例：

pulsar://localhost:6650

生产环境集群的 URL 形如：

pulsar://pulsar.us-west.example.com:6650

如果使用 TLS 认证，URL 使用pulsar+sslscheme 并切换到 TLS 默认端口6651：

pulsar+ssl://pulsar.us-west.example.com:6651

创建客户端（Client）

要与 Pulsar 交互，第一步是创建一个Client对象。使用NewClient函数并传入一个ClientOptions对象即可：

import ( "log" "runtime" "github.com/apache/pulsar/pulsar-client-go/pulsar" ) func main() { client, err := pulsar.NewClient(pulsar.ClientOptions{ URL: "pulsar://localhost:6650", OperationTimeoutSeconds: 5, MessageListenerThreads: runtime.NumCPU(), }) if err != nil { log.Fatalf("Could not instantiate Pulsar client: %v", err) } }

Client 配置参数

实现级补充：IOThreads、OperationTimeoutSeconds、ConcurrentLookupRequests、StatsIntervalInSeconds等参数在 C++ 客户端中都能找到对应实现。例如 ClientConfiguration.h 中setIOThreads、setOperationTimeoutSeconds、setConcurrentLookupRequest、setStatsIntervalInSeconds的 javadoc 明确说明默认值分别为 1、30 秒、50000（Go 侧文档取 5000 为更低的安全上限）和 600 秒；其中setConcurrentLookupRequest明确指出该值"仅在需要基于单个客户端生产/订阅数千个 topic 时才应调高"。TLSAllowInsecureConnection与TLSTrustCertsFilePath分别对应 C++ 侧setTlsAllowInsecureConnection与setTlsTrustCertsFilePath，后者描述为"设置受信任 TLS 证书文件的路径"。

生产者（Producers）

生产者负责向 Pulsar topic 发布消息。使用ProducerOptions对象配置 Go 生产者：

producer, err := client.CreateProducer(pulsar.ProducerOptions{ Topic: "my-topic", }) if err != nil { log.Fatalf("Could not instantiate Pulsar producer: %v", err) } defer producer.Close() msg := pulsar.ProducerMessage{ Payload: []byte("Hello, Pulsar"), } if err := producer.Send(msg); err != nil { log.Fatalf("Producer could not send message: %v", err) }

阻塞操作：创建新的 Pulsar 生产者时，操作会阻塞（等待一个 go channel），直到生产者创建成功或抛出错误。

生产者操作（Producer operations）

下面是一个更完整的生产者示例，同时演示同步与异步发送：

import ( "context" "fmt" "log" "github.com/apache/pulsar/pulsar-client-go/pulsar" ) func main() { // 实例化 Pulsar 客户端 client, err := pulsar.NewClient(pulsar.ClientOptions{ URL: "pulsar://localhost:6650", }) if err != nil { log.Fatal(err) } // 使用客户端实例化生产者 producer, err := client.CreateProducer(pulsar.ProducerOptions{ Topic: "my-topic", }) if err != nil { log.Fatal(err) } ctx := context.Background() // 同步发送 10 条消息、异步发送 10 条消息 for i := 0; i < 10; i++ { // 创建一条消息 msg := pulsar.ProducerMessage{ Payload: []byte(fmt.Sprintf("message-%d", i)), } // 尝试同步发送消息 if err := producer.Send(ctx, msg); err != nil { log.Fatal(err) } // 创建另一条用于异步发送的消息 asyncMsg := pulsar.ProducerMessage{ Payload: []byte(fmt.Sprintf("async-message-%d", i)), } // 尝试异步发送消息并处理回调 producer.SendAsync(ctx, asyncMsg, func(msg pulsar.ProducerMessage, err error) { if err != nil { log.Fatal(err) } fmt.Printf("Message %s successfully published", msg.ID()) }) } }

生产者配置（Producer configuration）

实现级补充：

• 哈希方案：三种HashingScheme在 C++ 客户端中分别有独立实现。JavaStringHash.cc 用hash = 31 * hash + val[i]的经典 JavaString.hashCode()算法计算并对int32取正；Murmur3_32Hash与BoostHash分别位于 Murmur3_32Hash.cc 与 BoostHash.cc，三个类都实现了同一个makeHash接口。
• 路由实现：默认的 round-robin 路由在 RoundRobinMessageRouter.cc 中实现：当消息带 key 时直接对 key 哈希取模选分区；无 key 且未开启批量时按消息粒度轮询；开启批量时则在同一分区累积消息，直到消息数、批量字节数或最大批量延迟任一阈值被触发才切换分区。这解释了MessageRoutingMode与HashingScheme协作时的真实行为。
• 队列控制：MaxPendingMessages与BlockIfQueueFull对应 C++ 侧setMaxPendingMessages（默认 1000）与setBlockIfQueueFull；C++ javadoc 明确写道"当队列满时，默认情况下所有Producer::send与Producer::sendAsync调用都会失败，除非blockIfQueueFull为 true"，与 Go 文档描述一致（见 ProducerConfiguration.h）。
• 压缩类型：C++ 侧setCompressionType的 javadoc 补充了重要限制：ZSTD 自 Pulsar 2.3 起支持，但要求消费端应用版本也必须 >= 2.3（见 ProducerConfiguration.h）。

消费者（Consumers）

消费者订阅一个或多个 Pulsar topic 并监听其上产生的消息。使用ConsumerOptions对象配置 Go 消费者。下面是一个使用 channel 的基础示例：

msgChannel := make(chan pulsar.ConsumerMessage) consumerOpts := pulsar.ConsumerOptions{ Topic: "my-topic", SubscriptionName: "my-subscription-1", Type: pulsar.Exclusive, MessageChannel: msgChannel, } consumer, err := client.Subscribe(consumerOpts) if err != nil { log.Fatalf("Could not establish subscription: %v", err) } defer consumer.Close() for cm := range msgChannel { msg := cm.Message fmt.Printf("Message ID: %s", msg.ID()) fmt.Printf("Message value: %s", string(msg.Payload())) consumer.Ack(msg) }

阻塞操作：创建新的 Pulsar 消费者时，操作会阻塞（在一个 go channel 上），直到订阅成功建立或抛出错误。

消费者操作（Consumer operations）

Receive 示例

使用Receive()方法处理消息的消费者示例：

import ( "context" "log" "github.com/apache/pulsar/pulsar-client-go/pulsar" ) func main() { // 实例化 Pulsar 客户端 client, err := pulsar.NewClient(pulsar.ClientOptions{ URL: "pulsar://localhost:6650", }) if err != nil { log.Fatal(err) } // 使用客户端对象实例化消费者 consumer, err := client.Subscribe(pulsar.ConsumerOptions{ Topic: "my-golang-topic", SubscriptionName: "sub-1", SubscriptionType: pulsar.Exclusive, }) if err != nil { log.Fatal(err) } defer consumer.Close() ctx := context.Background() // 在 topic 上无限监听 for { msg, err := consumer.Receive(ctx) if err != nil { log.Fatal(err) } // 对消息做处理 err = processMessage(msg) if err == nil { // 消息处理成功 consumer.Ack(msg) } else { // 消息处理失败 consumer.Nack(msg) } } }

消费者配置（Consumer configuration）

读者（Readers）

读者从 Pulsar topic 处理消息。与消费者的区别在于：使用读者时必须显式指定要从流中的哪条消息开始处理（消费者则自动从最近的未确认消息开始）。使用ReaderOptions对象配置 Go 读者：

reader, err := client.CreateReader(pulsar.ReaderOptions{ Topic: "my-golang-topic", StartMessageId: pulsar.LatestMessage, })

阻塞操作：创建新的 Pulsar 读者时，操作会阻塞（在一个 go channel 上），直到读者成功创建或抛出错误。

读者操作（Reader operations）

"Next" 示例

使用Next()方法处理消息的读者示例：

import ( "context" "log" "github.com/apache/pulsar/pulsar-client-go/pulsar" ) func main() { // 实例化 Pulsar 客户端 client, err := pulsar.NewClient(pulsar.ClientOptions{ URL: "pulsar://localhost:6650", }) if err != nil { log.Fatalf("Could not create client: %v", err) } // 使用客户端实例化读者 reader, err := client.CreateReader(pulsar.ReaderOptions{ Topic: "my-golang-topic", StartMessageID: pulsar.EarliestMessage, }) if err != nil { log.Fatalf("Could not create reader: %v", err) } defer reader.Close() ctx := context.Background() // 在 topic 上监听消息 for { msg, err := reader.Next(ctx) if err != nil { log.Fatalf("Error reading from topic: %v", err) } // 处理消息 } }

在上面的示例中，读者从最早可用消息（pulsar.EarliestMessage）开始读取。读者也可以从最新消息（pulsar.LatestMessage）开始，或通过DeserializeMessageID函数从字节数组反序列化得到某个指定的MessageID作为起点：

lastSavedId := // Read last saved message id from external store as byte[] reader, err := client.CreateReader(pulsar.ReaderOptions{ Topic: "my-golang-topic", StartMessageID: DeserializeMessageID(lastSavedId), })

这种从外部存储恢复MessageID的用法，使读者天然适合"断点续读"场景：例如将上次处理到的消息 ID 序列化后持久化，重启后精确地从断点继续消费。

读者配置（Reader configuration）

消息（Messages）

Pulsar Go 客户端提供了ProducerMessage接口，用于构造要发布到 Pulsar topic 的消息。示例：

msg := pulsar.ProducerMessage{ Payload: []byte("Here is some message data"), Key: "message-key", Properties: map[string]string{ "foo": "bar", }, EventTime: time.Now(), ReplicationClusters: []string{"cluster1", "cluster3"}, } if err := producer.send(msg); err != nil { log.Fatalf("Could not publish message due to: %v", err) }

ProducerMessage对象可用参数如下：

TLS 加密与认证

要启用 TLS 加密传输，需要按以下三步配置客户端：

1. 使用pulsar+sslURL 类型；
2. 将TLSTrustCertsFilePath设置为客户端与 Pulsar broker 使用的 TLS 证书路径；
3. 配置Authentication选项。

完整示例：

opts := pulsar.ClientOptions{ URL: "pulsar+ssl://my-cluster.com:6651", TLSTrustCertsFilePath: "/path/to/certs/my-cert.csr", Authentication: NewAuthenticationTLS("my-cert.pem", "my-key.pem"), }

实现级补充：TLS 相关配置项在 C++ 客户端中同样一一对应：ClientConfiguration.h 提供了setUseTls（默认 false）、setTlsTrustCertsFilePath、setTlsAllowInsecureConnection（默认 false）以及setValidateHostName（默认 false）等方法。其中setValidateHostName启用 TLS 主机名校验，会按 RFC 2818 的服务器身份主机名验证规则，用 x509 证书的 CN/SAN 与期望的 broker 主机名比对——生产环境建议在 TLS 之外同时开启主机名校验以防范中间人攻击。

延伸阅读

• 客户端库总览与功能矩阵：client-libraries.md
• C++ 客户端安装细节（RPM/Deb/Homebrew）：client-libraries-cpp.md
• 消息模型、订阅类型（Exclusive/Shared/Failover）与确认语义：concepts-messaging.md
• 分区 topic 与路由概念：concepts-architecture-overview.md
• TLS 传输加密与认证配置：security-tls-transport.md、security-tls-authentication.md
• 消息去重与发送超时设置建议：cookbooks-deduplication.md
• Go 客户端底层所依赖的 C++ 客户端源码：pulsar-client-cpp

• 消息队列
• 后端
• 流处理