关闭Goroutine安全通道无法终止多实例Binance WebSocket数据流问题

• 并发操作无锁导致竞态：你使用的嵌套哈希表clientSymChanMap没有加互斥锁保护，Go语言中普通map不支持并发读写，多协程同时启动/停止数据流时会出现数据覆盖、读取到错误值的问题，导致你存储的stopC通道和实际启动的WebSocket连接不匹配，自然无法正确关闭对应数据流。
• 通道关闭判断存在竞态漏洞：你实现的isClosed函数的判断逻辑和后续关闭操作不是原子操作，高并发场景下两个停止请求可能同时通过isClosed的未关闭判断，后续两次关闭同一个通道就会触发panic。同时如果在isClosed判断执行后、关闭执行前，通道被其他协程关闭，也会触发panic。
• Map键设计不合理：你当前存储stopC的键只有clientType和symbol，没有包含K线周期interval参数，如果同一个交易对同时启动多个不同周期的数据流，后面启动的数据流的stopC会直接覆盖前面的，导致之前的通道丢失，永远无法主动关闭。

• 第一步：优化Map结构与并发控制
  给嵌套Map增加读写锁，同时将interval纳入键的组成部分，避免同交易对不同周期的数据流通道被覆盖，参考定义：

import "sync"

type StreamState struct {
    mu sync.RWMutex
    // 第一层key是clientType，第二层key是 symbol:interval 拼接字符串
    clientSymChanMap map[bn.ClientType]map[string]*StreamEntry
}

type StreamEntry struct {
    stopC chan struct{}
    closeOnce sync.Once
}

• 第二步：移除不靠谱的isClosed判断，用sync.Once保证通道仅关闭一次
  修改后的StartDataStream逻辑：

func (c BinanceClient) StartDataStream(clientType bn.ClientType, symbol, interval string) error {
    switch clientType {
    case bn.SPOT_LIVE:
        wsKlineHandler := c.handlers.klineHandler.SpotKlineHandler
        wsErrHandler := c.handlers.klineHandler.ErrHandler

        doneC, stopC, err := binance.WsKlineServe(symbol, interval, wsKlineHandler, wsErrHandler)
        if err != nil {
            fmt.Println(err)
            return err
        }
        // 加写锁操作map
        c.state.mu.Lock()
        defer c.state.mu.Unlock()
        mapKey := symbol + ":" + interval
        if _, ok := c.state.clientSymChanMap[clientType]; !ok {
            c.state.clientSymChanMap[clientType] = make(map[string]*StreamEntry)
        }
        c.state.clientSymChanMap[clientType][mapKey] = &StreamEntry{
            stopC: stopC,
        }
        // 额外开协程监听doneC，连接断开后自动清理map
        go func() {
            <-doneC
            c.state.mu.Lock()
            defer c.state.mu.Unlock()
            delete(c.state.clientSymChanMap[clientType], mapKey)
        }()
        return nil
  ... 
}

修改后的StopDataStream逻辑：

func (c BinanceClient) StopDataStream(clientType bn.ClientType, symbol, interval string) {
    c.state.mu.Lock()
    defer c.state.mu.Unlock()
    mapKey := symbol + ":" + interval
    entry, ok := c.state.clientSymChanMap[clientType][mapKey]
    if !ok {
        DbgPrint("Stream not exist for: " + mapKey)
        return
    }
    // 用Once保证仅关闭一次，不会触发panic
    entry.closeOnce.Do(func() {
        close(entry.stopC)
    })
    delete(c.state.clientSymChanMap[clientType], mapKey)
    return
}

• 第三步：调整对外接口的入参，停止数据流时需要传入和启动时一致的interval参数，确保能匹配到对应的数据流通道。

内容的提问来源于stack exchange，提问作者Marvin.Hansen