Golang中多个WebSocket客户端冲突问题探讨

根据你提供的代码和描述，问题主要出现在WebSocket连接的并发管理和错误处理上。以下是具体问题和解决方案：

问题分析

1. 连接共享冲突：多个goroutine使用相同的WebSocket连接，导致读取冲突
2. 错误处理不当：1006错误是WebSocket的异常关闭，需要正确处理重连
3. 上下文管理问题：reading变量的并发访问存在竞态条件

解决方案

1. 每个goroutine独立连接

type WebSocketClient struct {
    symbol     string
    conn       *websocket.Conn
    ctx        context.Context
    cancel     context.CancelFunc
    reconnect  chan struct{}
    mu         sync.RWMutex
}

func NewWebSocketClient(symbol string) *WebSocketClient {
    ctx, cancel := context.WithCancel(context.Background())
    return &WebSocketClient{
        symbol:    symbol,
        ctx:       ctx,
        cancel:    cancel,
        reconnect: make(chan struct{}, 1),
    }
}

func (c *WebSocketClient) Start() {
    go c.run()
}

func (c *WebSocketClient) run() {
    defer c.cleanup()
    
    for {
        select {
        case <-c.ctx.Done():
            return
        default:
            if err := c.connectAndRead(); err != nil {
                log.Printf("%s connection error: %v", c.symbol, err)
                c.scheduleReconnect()
            }
        }
    }
}

func (c *WebSocketClient) connectAndRead() error {
    conn, err := c.dialWithRetry()
    if err != nil {
        return err
    }
    
    c.mu.Lock()
    c.conn = conn
    c.mu.Unlock()
    
    defer func() {
        c.mu.Lock()
        if c.conn != nil {
            c.conn.Close()
            c.conn = nil
        }
        c.mu.Unlock()
    }()
    
    return c.readMessages()
}

2. 改进的连接和重试机制

func (c *WebSocketClient) dialWithRetry() (*websocket.Conn, error) {
    var conn *websocket.Conn
    var err error
    
    for i := 0; i < 10; i++ {
        select {
        case <-c.ctx.Done():
            return nil, c.ctx.Err()
        default:
        }
        
        dialer := websocket.Dialer{
            HandshakeTimeout: 10 * time.Second,
            ReadBufferSize:   1024,
            WriteBufferSize:  1024,
        }
        
        conn, _, err = dialer.Dial("wss://ws.finnhub.io?token=xxxxxxxxxxxxxxxxx", nil)
        if err == nil {
            if err := c.subscribe(conn); err != nil {
                conn.Close()
                continue
            }
            return conn, nil
        }
        
        if i < 9 {
            backoff := time.Duration(math.Pow(2, float64(i))) * time.Second
            if backoff > 30*time.Second {
                backoff = 30 * time.Second
            }
            time.Sleep(backoff)
        }
    }
    
    return nil, fmt.Errorf("failed to connect after retries: %v", err)
}

func (c *WebSocketClient) subscribe(conn *websocket.Conn) error {
    msg, _ := json.Marshal(map[string]interface{}{
        "type":   "subscribe",
        "symbol": c.symbol,
    })
    
    conn.SetWriteDeadline(time.Now().Add(10 * time.Second))
    if err := conn.WriteMessage(websocket.TextMessage, msg); err != nil {
        return fmt.Errorf("subscribe failed: %v", err)
    }
    
    conn.SetReadDeadline(time.Time{})
    return nil
}

3. 改进的消息读取处理

func (c *WebSocketClient) readMessages() error {
    c.mu.RLock()
    conn := c.conn
    c.mu.RUnlock()
    
    if conn == nil {
        return errors.New("no connection")
    }
    
    conn.SetReadDeadline(time.Time{})
    conn.SetPongHandler(func(string) error {
        conn.SetReadDeadline(time.Now().Add(60 * time.Second))
        return nil
    })
    
    ticker := time.NewTicker(30 * time.Second)
    defer ticker.Stop()
    
    go func() {
        for range ticker.C {
            c.mu.RLock()
            if c.conn != nil {
                c.conn.WriteControl(websocket.PingMessage, []byte{}, time.Now().Add(5*time.Second))
            }
            c.mu.RUnlock()
        }
    }()
    
    for {
        select {
        case <-c.ctx.Done():
            return c.ctx.Err()
        default:
        }
        
        var msg map[string]interface{}
        if err := conn.ReadJSON(&msg); err != nil {
            if websocket.IsUnexpectedCloseError(err, 
                websocket.CloseGoingAway,
                websocket.CloseAbnormalClosure,
                websocket.CloseNormalClosure) {
                return fmt.Errorf("websocket closed: %v", err)
            }
            return err
        }
        
        if err := c.handleMessage(msg); err != nil {
            log.Printf("%s message handling error: %v", c.symbol, err)
        }
    }
}

func (c *WebSocketClient) handleMessage(msg map[string]interface{}) error {
    msgType, ok := msg["type"].(string)
    if !ok {
        return errors.New("invalid message format")
    }
    
    switch msgType {
    case "trade":
        data, ok := msg["data"].([]interface{})
        if !ok {
            return errors.New("invalid trade data")
        }
        
        for _, item := range data {
            if trade, ok := item.(map[string]interface{}); ok {
                log.Printf("%s trade: %v", c.symbol, trade)
            }
        }
    case "ping":
        c.mu.RLock()
        if c.conn != nil {
            c.conn.WriteJSON(map[string]string{"type": "pong"})
        }
        c.mu.RUnlock()
    }
    
    return nil
}

4. 使用示例

func main() {
    clients := []*WebSocketClient{
        NewWebSocketClient("AAPL"),
        NewWebSocketClient("TXN"),
        NewWebSocketClient("MSFT"),
    }
    
    for _, client := range clients {
        client.Start()
    }
    
    // 等待信号
    ch := make(chan os.Signal, 1)
    signal.Notify(ch, syscall.SIGINT, syscall.SIGTERM)
    <-ch
    
    for _, client := range clients {
        client.Stop()
    }
}

关键改进点

1. 连接隔离：每个股票代码使用独立的WebSocket连接和goroutine
2. 优雅重连：指数退避重试机制，避免频繁重连
3. 心跳保持：实现Ping/Pong机制保持连接活跃
4. 并发安全：使用互斥锁保护共享状态
5. 资源清理：确保连接正确关闭，避免资源泄漏

这种设计可以避免多个客户端之间的冲突，同时提供更稳定的连接管理。1006错误通常表示连接异常断开，通过上述重连机制可以自动恢复。