使用自定义处理程序 nats golang 保留订阅方法-Golang-PHP中文网

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

WBOY

发布： 2024-02-09 08:30:33

转载

1134人浏览过

使用自定义处理程序 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()

}

登录后复制

测试用例：

立即学习“go语言免费学习笔记（深入）”；

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 保留订阅方法，但我认为这不是正确的解决方案。