使用 gorun 结束 （币安） Web 服务流使用 go 例程

Question

我正在将币安API集成到现有系统中，虽然大多数部分都是直截了当的，但数据流API击中了我对go-routines的有限理解。我不认为币安的golang SDK中有什么特别之处，但本质上我只需要两个函数，一个函数启动数据流并使用作为参数给出的事件处理程序处理事件，另一个函数结束数据流而不会实际关闭客户端，因为它将关闭所有其他连接。在之前的一个项目中，有两种消息类型，但币安 SDK 使用一个实现，该实现返回两个 go 通道，一个用于错误，另一个用于停止数据 stram，我猜从名称中可以。我为启动数据流编写的代码如下所示：func startDataStream(symbol, interval string, wsKlineHandler futures.WsKlineHandler, errHandler futures.ErrHandler) (err error){    doneC, stopC, err := futures.WsKlineServe(symbol, interval, wsKlineHandler, errHandler)    if err != nil {        fmt.Println(err)        return err    }    return nil}这按预期工作并流式传输数据。一个简单的测试来验证它：func runWSDataTest() {    symbol := "BTCUSDT"    interval := "15m"    errHandler := func(err error) {fmt.Println(err)}    wsKlineHandler := func(event *futures.WsKlineEvent) {fmt.Println(event)}    _ = startDataStream(symbol, interval, wsKlineHandler, errHandler)}对我来说不太清楚的事情，主要是由于不完全理解，真的是我如何停止流。我认为返回的 stopC 通道可用于以某种方式发出类似于系统级 sigterm 的结束单句，然后流应该结束。比如说，我有一个停止数据流函数，它采用一个符号作为参数func stopDataStream(symbol){}

Answer 1

假设我为五个交易品种启动 5 个数据流，现在我只想停止其中一个流。这就引出了一个问题：

1. 如何跟踪所有这些止损通道？

2. 我是否可以使用带有符号键控的集合，拉动 stopC 通道，然后发出信号来结束该数据流？

3. 如何从停止函数实际写入 stopC 通道？

同样，我不认为这特别困难，只是我无法从文档中弄清楚，所以任何帮助将不胜感激。

谢谢

Answer 2

事实证明，只需保存&关闭频道即可解决所有问题。我真的很惊讶这是多么容易，但这是更新函数的代码：

func startDataStream(symbol, interval string, wsKlineHandler futures.WsKlineHandler, errHandler futures.ErrHandler) (err error) {

    _, stopC, err := futures.WsKlineServe(symbol, interval, wsKlineHandler, errHandler)

    if err != nil {

        fmt.Println(err)

        return err

    }

    // just save the stop channel 

    chanMap[symbol] = stopC

    return nil

}

然后，停止函数真的变得令人尴尬：

func stopDataStream(symbol string) {

    stopC := chanMap[symbol]  // load the stop channel for the symbol

    close(stopC) // just close it. 

}

最后，测试一下：

var (

    chanMap map[string]chan struct{}

)

func runWSDataTest() {

    chanMap = make(map[string]chan struct{})

    symbol := "BTCUSDT"

    interval := "15m"

    errHandler := func(err error) { fmt.Println(err) }

    wsKlineHandler := getKLineHandler()

    println("Start stream")

    _ = startDataStream(symbol, interval, wsKlineHandler, errHandler)

    time.Sleep(3 * time.Second)

    println("Stop stream")

    stopDataStream(symbol)

    time.Sleep(1 * time.Second)

}

就是这样。