首页 >后端开发 >Golang >使用自定义处理程序 nats golang 保留订阅方法

使用自定义处理程序 nats golang 保留订阅方法

WBOY
WBOY转载
2024-02-09 08:30:331132浏览

使用自定义处理程序 nats golang 保留订阅方法

php小编香蕉在本文中将为你介绍如何使用自定义处理程序 nats golang 来保留订阅方法。在开发过程中,我们经常需要使用消息队列来处理异步任务,而 nats golang 是一个轻量级的消息队列系统,具有高性能和可扩展性。通过自定义处理程序,我们可以保留订阅方法,实现更灵活的消息处理和流程控制。下面我们将详细介绍如何在 nats golang 中实现这一功能。

问题内容

我正在 golang 的 nats 客户端之上编写包装器,我想获取处理函数,一旦我从 nats 服务器收到消息,就可以从消费者调用该函数。 我想保留自定义订阅方法,直到它收到来自 nats 的消息。

发布:

func (busconfig busconfig) publish(service string, data []byte) error {
    puberr := conn.publish(service, data)
    if puberr != nil {
        return puberr
    }
    return nil
}

订阅:

func (busconfig busconfig) subscribe(subject string, handler func(msg []byte)) {
    fmt.println("subscrbing on : ", subject)

    //wg := sync.waitgroup{}
    //wg.add(1)
    subscription, err := conn.subscribe(subject, func(msg *nats.msg) {
        go func() {
            handler(msg.data)
        }()
        //wg.done()
    })
    if err != nil {
        fmt.println("subscriber error : ", err)
    }
    //wg.wait()
    defer subscription.unsubscribe()

}

测试用例:

func TestLifeCycleEvent(t *testing.T) {
    busClient := GetBusClient()
    busClient.Subscribe(SUBJECT, func(input []byte) {
        fmt.Println("Life cycle event received :", string(input))
    })

    busClient.Publish(SUBJECT, []byte("complete notification"))
}

我看到消息已发布但未订阅,我尝试使用 waitgroup 保留订阅方法,但我认为这不是正确的解决方案。

解决方法

您看不到正在传递的消息,因为 Subscribe 是一个异步方法,它会生成一个 goroutine 来处理传入消息并调用回调。

调用 busClient.Publish() 之后,您的应用程序立即退出。它不会等待 Subscribe() 内部发生任何事情。

当您使用 nats.Subscribe() 时,您通常会有一个长时间运行的应用程序,该应用程序会在特定条件下退出(例如收到关闭信号)。 WaitGroup 可以在这里工作,但可能不适用于实际应用程序,仅用于测试。

您还应该在 NATS 连接上调用 Flush() 方法,以确保在退出程序之前已发送所有缓冲的消息。

如果想要同步方法,可以使用nats.SubscribeSync()

查看示例:https://natsbyexample.com/examples/messaging/发布-订阅/执行

以上是使用自定义处理程序 nats golang 保留订阅方法的详细内容。更多信息请关注PHP中文网其他相关文章!

声明:
本文转载于:stackoverflow.com。如有侵权,请联系admin@php.cn删除