Apache Pulsar Go Client出现意外创建Producer/Consumer的问题

核心问题定位 #

你遇到的「创建Consumer时意外生成第三个Producer」，大概率是代码里Producer实例没复用+Consumer逻辑里误触发了Producer创建导致的。而消息计数异常，要么是重复创建的Producer多发送了消息，要么是消费端的订阅位点、消息确认逻辑出了问题。

具体排查步骤 #

• 检查Producer的创建逻辑
  • 看看/produce接口是不是每次调用都新建一个Producer，没复用全局实例——两次调用就会生成2个，再加上Consumer代码里误创建的1个，正好是3个。
  • 仔细扒Consumer的初始化代码，有没有不小心抄了Producer的创建代码，或者调用了会创建Producer的工具函数。
• 验证Pulsar客户端配置
  • 检查Go客户端的ProducerOptions，有没有配置导致自动创建新实例的参数，比如错误开启了每次请求新建连接的选项。
  • 确认Consumer的订阅名是不是固定的，如果每次创建Consumer都用随机订阅名，会导致每个订阅都有自己的消费位点，统计出来的数量肯定乱。
• 消息计数异常的具体排查
  • 用pulsar-admin topics stats <你的topic名>查看：topic的总消息数、每个订阅的已消费/未消费数量，先搞清楚是生产多了还是消费漏了/重复了。
  • 检查/produce接口是不是每次调用确实只发1条消息，有没有重复发送的逻辑；/consume接口消费完消息后有没有调用msg.Ack()，没确认的消息Pulsar会重复投递，计数自然不对。

修复建议 #

1. 全局复用Producer实例
   • 把Producer做成全局单例，应用启动时初始化一次，不要每次调用/produce都新建。示例代码：

     var globalProducer pulsar.Producer
     func init() {
         client, err := pulsar.NewClient(pulsar.ClientOptions{URL: "pulsar://你的pulsar服务地址:6650"})
         if err != nil {
             // 处理错误
         }
         globalProducer, err = client.CreateProducer(pulsar.ProducerOptions{Topic: "你的topic名"})
         if err != nil {
             // 处理错误
         }
     }
     

2. 清理Consumer代码
   • 逐行检查Consumer的创建逻辑，删掉所有不该出现的Producer创建代码，确保Consumer只负责初始化自己的实例。
3. 固定订阅名+正确确认消息
   • Consumer的SubscriptionName必须固定，不能每次创建都变，不然每个订阅独立统计，数据肯定乱。
   • 消费消息后一定要调用msg.Ack()，告诉Pulsar这条消息已经处理完了，避免重复投递导致计数异常。
4. 加日志追踪实例创建
   • 在Producer和Consumer的创建代码里加明确的日志，比如log.Printf("创建Producer实例，ID: %v", producer)、log.Printf("创建Consumer实例，订阅名: %s", subName)，这样就能清楚看到什么时候创建了哪些实例，快速定位误创建的来源。

内容的提问来源于stack exchange，提问作者Frank Wu