使用自定义处理程序 nats golang 保留订阅方法- 技术经验 -卓越飞翔博客

使用自定义处理程序 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/发布-订阅/执行

问题内容

解决方法

相关推荐