登录
首页 >  Golang >  Go问答

使用 jscontext.newConsumer() 在 Golang 中创建 NATS 推送消费者

来源:stackoverflow

时间:2024-03-26 15:18:37 332浏览 收藏

在 Golang 中,使用 JetStream 库时,可以通过 `jscontext.AddConsumer()` 方法创建推送消费者。与创建拉取消费者不同,推送消费者需要指定一个传递主题 (`deliverysubject`),它可以与消费者名称相同。文档对该主题的解释不清,导致许多开发者感到困惑。本文将提供一个完整的工作示例,展示如何创建推送消费者并从指定主题中订阅事件。

问题内容

在创建js上下文AddConsumer(topic.ID(), &nats.ConsumerConfig{})创建pull消费者时,有没有办法使用addconsumer创建push消费者


正确答案


文档关于这部分的内容非常糟糕,并且错误描述性不强。我花了几个小时才弄清楚。这是我发现的...

要使用 addconsumer 创建推送使用者,您必须提供 deliverysubject 参数,它可以是任何内容,并且为了简单起见,它可以与使用者名称相同。这部分的文档非常糟糕,没有解释它应该如何使用。

以下是一个完整的工作示例,它创建一个消费者并从该消费者订阅某个主题。

nc, err := nats.Connect("nats://localhost:4222")
if err != nil {
    log.Fatalf("could not connect to nats: %s", err)
}

js, err := nc.JetStream()
if err != nil {
    log.Fatal("Could not get jetstream:", err)
}

streamName := "TEST_STREAM"
_, err = js.AddStream(&nats.StreamConfig{
    Name:      streamName,
    Subjects:  []string{"test.>"},
    Retention: nats.InterestPolicy,
    Replicas:  1,
})

if err != nil {
    fmt.Printf("error creating stream %q: %s\n", streamName, err)
    return
}

groupName := "group-1"
consumerName := fmt.Sprintf("%s-consumer", groupName)
_, err = js.AddConsumer(
    streamName,
    &nats.ConsumerConfig{
        Name:           consumerName,
        Durable:        consumerName,
        DeliverSubject: consumerName,
        DeliverGroup:   groupName,
    },
)

if err != nil {
    fmt.Printf("error creating consumer %q: %s\n", consumerName, err)
    return
}

subject := "test.resource.created"
_, err = js.QueueSubscribe(
    subject,
    groupName,
    func(msg *nats.Msg) { 
        fmt.Printf("Jetstream - received %q: %q\n", msg.Subject, msg.Data) 
    },
    nats.Bind(streamName, consumerName),
)

if err != nil {
    fmt.Printf("error subscribing to %q: %s\n", subject, err)
}
提供

传递组,以便多个应用程序实例可以订阅同一个消费者,并且事件只会传递给其中一个(随机)。也就是说,如果您想水平缩放应用程序。

如果您想限制消费者接收哪些事件。然后您应该向消费者配置提供 subjectfilter 。每个消费者将保留其过滤器匹配的所有事件,或者如果未提供过滤器,则保留流中的所有事件。

创建使用者后,您可以使用任何 jetstream 订阅方法(js.subscribesyncjs.queuesubscribesyncjs.subscribejs.queuesubscribe,...)来订阅它,但您必须提供 nats 选项将订阅绑定到消费者,这是使用 nats.bind(streamname, consumername) 完成的

订阅时可以将主题名称保留为空字符串,以订阅来自消费者的所有事件。

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

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