如何加速Golang单服务器单客户端的WebSocket传输

开发者阿泽

2026-6-20

单服务器单客户端高速传输优化需求

我用Golang搭建单服务器单客户端连接，目标是实现最快端到端传输速率，需满足60帧/秒全高清视频传输（压缩后约30Mbps）。当前测试：以太网环境速率约11Mbps，WiFi环境约6Mbps。我给服务端和客户端都做了多通道发送方案，服务端多通道让WiFi速率翻倍，但客户端多通道没效果，这符合客户端代码逻辑。下面是基于WebSocket实现的服务端和客户端代码，希望优化到最快传输速率，也求推荐Golang生态里更高效的数据传输技术或库。

服务端代码（main.go）

package main

import (
    "fmt"
    "log"
    "net/http"
    "sync"
    "time"

    "github.com/gorilla/websocket"
)

var upgrader = websocket.Upgrader{
    CheckOrigin: func(r *http.Request) bool {
        return true
    },
}

type Channel struct {
    name      string
    buffer    chan []byte
    bytesSent int
    mu        sync.Mutex
}

type Server struct {
    clients   map[*websocket.Conn]bool
    broadcast chan []byte
    channels  map[string]*Channel
    mu        sync.Mutex
}

func newServer() *Server {
    return &Server{
        clients:   make(map[*websocket.Conn]bool),
        broadcast: make(chan []byte),
        channels:  make(map[string]*Channel),
    }
}

func (s *Server) handleConnections(w http.ResponseWriter, r *http.Request) {
    ws, err := upgrader.Upgrade(w, r, nil)
    if err != nil {
        log.Fatalf("Failed to upgrade to WebSocket: %v", err)
    }
    defer ws.Close()

    s.mu.Lock()
    s.clients[ws] = true
    s.mu.Unlock()

    for {
        _, msg, err := ws.ReadMessage()
        if err != nil {
            log.Printf("Error reading message: %v", err)
            delete(s.clients, ws)
            break
        }
        s.broadcast <- msg
    }
}

func (s *Server) handleMessages() {
    for {
        msg := <-s.broadcast
        for client := range s.clients {
            err := client.WriteMessage(websocket.BinaryMessage, msg)
            if err != nil {
                log.Printf("Error writing message: %v", err)
                client.Close()
                delete(s.clients, client)
            }
        }
    }
}

func (s *Server) createChannel(name string) {
    s.mu.Lock()
    defer s.mu.Unlock()
    if _, exists := s.channels[name]; !exists {
        s.channels[name] = &Channel{name: name, buffer: make(chan []byte, 100)}
    }
}

func (s *Server) startChannel(name string) {
    s.mu.Lock()
    channel, exists := s.channels[name]
    s.mu.Unlock()
    if !exists {
        log.Printf("Channel %s does not exist", name)
        return
    }

    go func() {
        for msg := range channel.buffer {
            s.broadcast <- append([]byte(name+": "), msg...)
            channel.mu.Lock()
            channel.bytesSent += len(msg)
            channel.mu.Unlock()
        }
    }()
}

func (s *Server) sendToChannel(name string, msg []byte) {
    s.mu.Lock()
    channel, exists := s.channels[name]
    s.mu.Unlock()
    if !exists {
        log.Printf("Channel %s does not exist", name)
        return
    }
    channel.buffer <- msg
}

func main() {
    server := newServer()

    http.HandleFunc("/ws", server.handleConnections)

    go server.handleMessages()

    numChannels := 10 // Number of channels
    for i := 1; i <= numChannels; i++ {
        channelName := fmt.Sprintf("channel_%d", i)
        server.createChannel(channelName)
        server.startChannel(channelName)
    }

    // Queue for distributing data
    dataQueue := make(chan []byte, 1000)

    // Generate dummy data
    go func() {
        dummyData := make([]byte, 512*1024)             // 0.5 MB of dummy data
        ticker := time.NewTicker(time.Millisecond * 17) // 60 times per second
        for {
            select {
            case <-ticker.C:
                dataQueue <- dummyData
            }
        }
    }()

    // Distribute data among channels
    go func() {
        channelId := 1
        for data := range dataQueue {
            channelName := fmt.Sprintf("channel_%d", channelId)
            server.sendToChannel(channelName, data)
            channelId++
            if channelId > numChannels {
                channelId = 1
            }
        }
    }()

    // Log the total sent data every second
    go func() {
        ticker := time.NewTicker(1 * time.Second)
        defer ticker.Stop()
        for range ticker.C {
            totalSent := 0
            for _, channel := range server.channels {
                channel.mu.Lock()
                totalSent += channel.bytesSent
                channel.bytesSent = 0 // reset for the next second
                channel.mu.Unlock()
            }
            log.Printf("Total Sent: %.2f MB/s", float64(totalSent)/1024/1024)
        }
    }()

    log.Println("Server started on :8080")
    err := http.ListenAndServe(":8080", nil)
    if err != nil {
        log.Fatalf("Server failed to start: %v", err)
    }
}

客户端代码（main.go）

package main

import (
    "log"
    "net/url"
    "sync"
    "time"

    "github.com/gorilla/websocket"
)

type ChannelStats struct {
    sync.Mutex
    bytesReceived int
}

func readFromWebSocket(u url.URL, messageChan chan []byte, done chan bool) {
    u.Path = "/ws"
    c, _, err := websocket.DefaultDialer.Dial(u.String(), nil)
    if err != nil {
        log.Fatalf("Dial error: %v", err)
    }
    defer c.Close()

    for {
        _, message, err := c.ReadMessage()
        if err != nil {
            log.Printf("Read error: %v", err)
            done <- true
            return
        }
        messageChan <- message
    }
}

func processMessages(stats *ChannelStats, messageChan chan []byte, wg *sync.WaitGroup, done chan bool) {
    defer wg.Done()

    for {
        select {
        case message := <-messageChan:
            stats.Lock()
            stats.bytesReceived += len(message)
            stats.Unlock()
        case <-done:
            return
        }
    }
}

func main() {
    u := url.URL{Scheme: "ws", Host: "192.168.0.86:8080"}
    var wg sync.WaitGroup
    done := make(chan bool)
    messageChan := make(chan []byte, 100) // Buffered channel to hold messages
    stats := &ChannelStats{}
    numWorkers := 10 // Number of worker goroutines

    // Start a single reader goroutine
    go readFromWebSocket(u, messageChan, done)

    // Start multiple worker goroutines
    for i := 0; i < numWorkers; i++ {
        wg.Add(1)
        go processMessages(stats, messageChan, &wg, done)
    }

    // Log stats every second
    go func() {
        for {
            time.Sleep(1 * time.Second)
            stats.Lock()
            totalReceived := stats.bytesReceived
            stats.bytesReceived = 0 // reset for the next second
            stats.Unlock()
            log.Printf("Total Received: %.2f MB/s", float64(totalReceived)/1024/1024)
        }
    }()

    // Wait for all workers to finish
    wg.Wait()
    close(done)
}

优化方案与技术推荐

一、WebSocket现有代码优化

1. 服务端优化

重构多通道逻辑：当前服务端的Channel只是逻辑拆分，本质还是单WebSocket连接发送数据。要真正利用多通道，应为每个通道创建独立的WebSocket连接，让客户端同时建立多个连接并行传输。

增大缓冲区：调整WebSocket的读写缓冲区大小，减少系统调用和数据拷贝开销：

upgrader = websocket.Upgrader{
    CheckOrigin:       func(r *http.Request) bool { return true },
    ReadBufferSize:    1024 * 1024,
    WriteBufferSize:   1024 * 1024,
    EnableCompression: false, // 已压缩的视频数据无需再压缩
}

减少锁开销：用原子操作替代Channel中统计bytesSent的互斥锁，降低并发竞争：

import "sync/atomic"
// 修改Channel结构
type Channel struct {
    name      string
    buffer    chan []byte
    bytesSent int64
}
// 统计时用原子操作
atomic.AddInt64(&channel.bytesSent, int64(len(msg)))

2. 客户端优化

多连接并行传输：把当前单WebSocket连接改成多个独立连接，每个连接对应服务端的一个通道，这样才能真正提升传输速率，匹配服务端的多通道设计。

调整客户端缓冲区：Dial时设置更大的读写缓冲区并关闭压缩：

dialer := websocket.Dialer{
    ReadBufferSize:    1024 * 1024,
    WriteBufferSize:   1024 * 1024,
    EnableCompression: false,
}

减少数据拷贝：直接在读取goroutine内处理数据（比如更新统计），避免通过messageChan二次拷贝数据。

二、更高效的传输技术推荐

1. 裸TCP连接

跳过WebSocket的HTTP握手和帧头开销，直接用Golang的net.TCPConn实现自定义二进制传输协议。可以设计简单的帧格式（比如前4字节表示数据长度），大幅降低协议开销，提升传输效率。

2. QUIC协议

基于UDP的可靠传输协议，相比TCP有更低的延迟、更好的多路复用能力，适合WiFi等不稳定网络。Golang的quic-go库可快速实现QUIC传输，能有效提升高带宽场景下的传输速率。

3. 专用流媒体库

RTP协议实现：针对视频传输场景，使用RTP协议做流媒体传输，Golang有成熟的RTP库支持丢包重传、时序控制等特性，适配实时视频需求。
mediasoup-go：基于WebRTC的流媒体服务器库，自带低延迟音视频传输优化，支持编解码和传输调度，适合专业实时视频场景。

内容的提问来源于stack exchange，提问作者ZZ0R0

如何加速Golang单服务器单客户端的WebSocket传输

服务端代码（main.go） #

客户端代码（main.go） #

一、WebSocket现有代码优化 #

1. 服务端优化 #

2. 客户端优化 #

二、更高效的传输技术推荐 #

1. 裸TCP连接 #

2. QUIC协议 #

3. 专用流媒体库 #