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

Question

我有一个 Pulsar 主题，我通过多个生产者向该主题发布消息。这些生产者只是 goroutines，我正在使用SendAsync生产者上的方法发送消息，但我不断收到Got unexpected send receipt for message警告。我查看了客户端的库代码（Pulsar 的 golang 客户端），发现只有当生产者未在此 map 中注册为侦听器时，才会显示此警告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)        }    })}

Answer 1

生产者未在map[uint64]ConnectionListener. 要么，生产者从未创建，要么生产者关闭。就我而言，我能够发布消息，这意味着确实创建了生产者，但由于我使用SendAsync方法发送消息，因此它不会阻止接收确认。而且，我曾经defer在生成所有消息后关闭生产者（我确实使用 await group来确保在关闭生产者之前发送所有消息）但生产者在收到确认之前关闭，因此收到警告。

Answer 2

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