在 Pulsar 生产者上使用“SendAsync”方法时收到“收到消息的意外发送回执”警告答案

【问题标题】：Receiving `Got unexpected send receipt for message` warning when using `SendAsync` method on Pulsar producer在 Pulsar 生产者上使用“SendAsync”方法时收到“收到消息的意外发送回执”警告
【发布时间】：2020-08-26 00:10:26
【问题描述】：

我有一个 Pulsar 主题，我通过多个生产者向该主题发布消息。这些生产者只是 goroutine，我在生产者上使用 SendAsync 方法发送消息，但我不断收到 Got unexpected send receipt for message 警告。我查看了客户端的库代码（Pulsar 的 golang 客户端），发现只有当生产者未在此映射 map[uint64]ConnectionListener 中注册为侦听器时，才会显示此警告。但是，如果生产者没有注册为监听器，我如何发布呢？

func benchmarkConcurrentProducers(b *testing.B, pool []pulsar.Producer, concur int, payload []byte) {
    wg := &sync.WaitGroup{}
    wg.Add(concur)

    for i := 0; i < concur; i++ {
        go Produce(pool[i], payload, wg)
    }
    wg.Wait()
}

func BenchmarkProducer(b *testing.B) {
    type config struct {
        Pulsarcfg struct {
            Url string `required: "true"`
        }
    }
    cfg := &config{}
    err := configor.Load(cfg, "../config.yml")
    require.Nil(b, err)

    client, err := pulsar.NewClient(pulsar.ClientOptions{
        URL:               cfg.Pulsarcfg.Url,
        OperationTimeout:  30 * time.Second,
        ConnectionTimeout: 30 * time.Second,
    })
    require.Nil(b, err)
    defer client.Close()

    concur := 4

    producerPool := make([]pulsar.Producer, concur)

    for i := 0; i < concur; i++ {
        p, err := New(client, string(i))
        require.Nil(b, err)
        defer p.Close()

        producerPool[i] = p
    }

    path, err := filepath.Abs("../events.json")
    if err != nil {
        log.Fatalf("invalid payload file path, %v", err)
    }
    f, err := ioutil.ReadFile(path)
    if err != nil {
        log.Fatalf("bad payload: %v", err)
    }

    b.Run("benchmark producer", func(b *testing.B) {
        // run the producer b.N times
        for n := 0; n < b.N; n++ {
            benchmarkConcurrentProducers(b, producerPool, concur, f)
        }
    })
}

【问题讨论】：

标签： go publish-subscribe apache-pulsar

【解决方案1】：

生产者未在map[uint64]ConnectionListener 中注册为侦听器可能有两个原因。要么，生产者从未创建，要么生产者关闭。就我而言，我能够发布消息，这意味着确实创建了生产者，但由于我使用SendAsync 方法发送消息，因此它没有阻止接收确认。而且，我在生成所有消息后使用defer 关闭生产者（我确实使用wait group 来确保在关闭生产者之前发送所有消息）但是生产者在收到确认之前就关闭了，因此，警告。

【讨论】：

你能发布代码示例吗？正如您之前所说，连接侦听器映射由生产者 ID 跟踪。你能在日志中追踪生产者ID吗？
@MingL。我已经添加了代码，是的，我能够在日志中看到生产者 ID。

【解决方案2】：

我收到了完全相同的错误消息。在我的例子中，我使用同一个客户端创建了一个阅读器和一个生产者。虽然在我生成新消息时阅读器已经关闭，但我一直收到这个警告，更糟糕的是，它会挂在对producer.Send() 的调用上。我通过使用单独的客户端进行阅读和制作解决了这个问题。

【讨论】：