登录
首页 >  Golang >  Go问答

Pulsar 生产者使用“SendAsync”方法时出现意外消息发送收据的警告

来源:stackoverflow

时间:2024-03-08 18:09:13 280浏览 收藏

最近发现不少小伙伴都对Golang很感兴趣,所以今天继续给大家介绍Golang相关的知识,本文《Pulsar 生产者使用“SendAsync”方法时出现意外消息发送收据的警告》主要内容涉及到等等知识点,希望能帮到你!当然如果阅读本文时存在不同想法,可以在评论中表达,但是请勿使用过激的措辞~

问题内容

我有一个 pulsar 主题,我通过多个生产者向该主题发布消息。这些生产者只是 goroutine,我在生产者上使用 sendasync 方法来发送消息,但我不断收到 got 意外的 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)
        }
    })
}

解决方案


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

以上就是本文的全部内容了,是否有顺利帮助你解决问题?若是能给你带来学习上的帮助,请大家多多支持golang学习网!更多关于Golang的相关知识,也可关注golang学习网公众号。

声明:本文转载于:stackoverflow 如有侵犯,请联系study_golang@163.com删除
相关阅读
更多>
最新阅读
更多>
课程推荐
更多>