Go WebSocket即時聊天實戰：從零構建可擴展訊息伺服器的5個關鍵架構

技术架构

Go WebSocket即時聊天：為什麼2026年你還在用輪詢

還在用HTTP輪詢實現即時訊息？每秒發100次請求，伺服器扛不住，訊息延遲3-5秒？WebSocket是全雙工即時通訊的標準協定，Go語言的goroutine模型天然適合高併發WebSocket服務。2026年，gorilla/websocket和nhooyr.io/websocket兩大庫已經成熟，配合Redis Pub/Sub和Kafka，可以輕鬆構建支援10萬+線上連線的即時聊天系統。

本文將從5個關鍵架構出發，帶你完成連線管理→房間模型→訊息廣播→持久化儲存→水平擴展的全鏈路實戰。

核心概念

概念 說明
WebSocket 全雙工持久連線協定，RFC 6455
Hub 連線管理中心，維護所有客戶端
Room 聊天房間，訊息廣播的隔離單元
Client WebSocket客戶端連線封裝
Pub/Sub 發布訂閱模式，用於跨程序訊息分發
Backpressure 背壓控制，防止慢客戶端拖垮服務
Heartbeat 心跳偵測，及時發現斷開連線
Message Queue 訊息佇列，保證訊息可靠投遞

問題分析：WebSocket聊天的5大挑戰

  1. 連線管理：10萬+連線的記憶體和goroutine開銷
  2. 訊息廣播：一條訊息需要推送給房間內所有線上使用者
  3. 慢客戶端：某個客戶端消費慢導致服務端記憶體溢位
  4. 水平擴展：單機無法承載，需要多實例協同
  5. 訊息持久化：離線使用者上線後需要取得歷史訊息

分步實操：5個WebSocket聊天核心架構

架構1：Hub-Client連線管理模型

// internal/chat/client.go
package chat

import (
	"encoding/json"
	"log"
	"time"

	"github.com/gorilla/websocket"
)

const (
	writeWait      = 10 * time.Second
	pongWait       = 60 * time.Second
	pingPeriod     = (pongWait * 9) / 10
	maxMessageSize = 512
)

type Client struct {
	hub  *Hub
	conn *websocket.Conn
	send chan []byte
	userID   string
	username string
	rooms    map[string]*Room
}

func NewClient(hub *Hub, conn *websocket.Conn, userID, username string) *Client {
	return &Client{
		hub:      hub,
		conn:     conn,
		send:     make(chan []byte, 256),
		userID:   userID,
		username: username,
		rooms:    make(map[string]*Room),
	}
}

func (c *Client) ReadPump() {
	defer func() {
		c.hub.unregister <- c
		c.conn.Close()
	}()

	c.conn.SetReadLimit(maxMessageSize)
	c.conn.SetReadDeadline(time.Now().Add(pongWait))
	c.conn.SetPongHandler(func(string) error {
		c.conn.SetReadDeadline(time.Now().Add(pongWait))
		return nil
	})

	for {
		_, message, err := c.conn.ReadMessage()
		if err != nil {
			if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway, websocket.CloseAbnormalClosure) {
				log.Printf("read error: %v", err)
			}
			break
		}

		var msg Message
		if err := json.Unmarshal(message, &msg); err != nil {
			log.Printf("invalid message: %v", err)
			continue
		}

		msg.Sender = c.userID
		msg.Timestamp = time.Now().UnixMilli()

		c.hub.HandleMessage(c, &msg)
	}
}

func (c *Client) WritePump() {
	ticker := time.NewTicker(pingPeriod)
	defer func() {
		ticker.Stop()
		c.conn.Close()
	}()

	for {
		select {
		case message, ok := <-c.send:
			c.conn.SetWriteDeadline(time.Now().Add(writeWait))
			if !ok {
				c.conn.WriteMessage(websocket.CloseMessage, []byte{})
				return
			}

			w, err := c.conn.NextWriter(websocket.TextMessage)
			if err != nil {
				return
			}
			w.Write(message)

			n := len(c.send)
			for i := 0; i < n; i++ {
				w.Write([]byte{'\n'})
				w.Write(<-c.send)
			}

			if err := w.Close(); err != nil {
				return
			}

		case <-ticker.C:
			c.conn.SetWriteDeadline(time.Now().Add(writeWait))
			if err := c.conn.WriteMessage(websocket.PingMessage, nil); err != nil {
				return
			}
		}
	}
}

// internal/chat/hub.go
package chat

import (
	"log"
	"sync"
)

type Hub struct {
	clients    map[*Client]bool
	rooms      map[string]*Room
	register   chan *Client
	unregister chan *Client
	mu         sync.RWMutex
}

func NewHub() *Hub {
	return &Hub{
		clients:    make(map[*Client]bool),
		rooms:      make(map[string]*Room),
		register:   make(chan *Client),
		unregister: make(chan *Client),
	}
}

func (h *Hub) Run() {
	for {
		select {
		case client := <-h.register:
			h.mu.Lock()
			h.clients[client] = true
			h.mu.Unlock()
			log.Printf("client connected: %s (total: %d)", client.userID, len(h.clients))

		case client := <-h.unregister:
			h.mu.Lock()
			if _, ok := h.clients[client]; ok {
				delete(h.clients, client)
				close(client.send)
				for _, room := range client.rooms {
					room.Leave(client)
				}
			}
			h.mu.Unlock()
			log.Printf("client disconnected: %s (total: %d)", client.userID, len(h.clients))
		}
	}
}

func (h *Hub) GetOrCreateRoom(roomID string) *Room {
	h.mu.Lock()
	defer h.mu.Unlock()

	if room, ok := h.rooms[roomID]; ok {
		return room
	}

	room := NewRoom(roomID)
	h.rooms[roomID] = room
	go room.Run()
	log.Printf("room created: %s", roomID)
	return room
}

func (h *Hub) HandleMessage(client *Client, msg *Message) {
	switch msg.Type {
	case "join":
		room := h.GetOrCreateRoom(msg.RoomID)
		room.Join(client)
	case "leave":
		if room, ok := h.rooms[msg.RoomID]; ok {
			room.Leave(client)
		}
	case "chat":
		if room, ok := h.rooms[msg.RoomID]; ok {
			room.Broadcast(msg)
		}
	}
}

架構2：Room聊天房間與訊息廣播

// internal/chat/room.go
package chat

import (
	"encoding/json"
	"log"
	"sync"
)

type Room struct {
	ID      string
	clients map[*Client]bool
	forward chan *Message
	join    chan *Client
	leave   chan *Client
	mu      sync.RWMutex
}

type Message struct {
	Type      string `json:"type"`
	RoomID    string `json:"roomId"`
	Sender    string `json:"sender"`
	Content   string `json:"content"`
	Timestamp int64  `json:"timestamp"`
}

func NewRoom(id string) *Room {
	return &Room{
		ID:      id,
		clients: make(map[*Client]bool),
		forward: make(chan *Message, 100),
		join:    make(chan *Client),
		leave:   make(chan *Client),
	}
}

func (r *Room) Run() {
	for {
		select {
		case client := <-r.join:
			r.mu.Lock()
			r.clients[client] = true
			r.mu.Unlock()
			r.broadcastSystemMsg(client.username + " joined the room")

		case client := <-r.leave:
			r.mu.Lock()
			if _, ok := r.clients[client]; ok {
				delete(r.clients, client)
			}
			r.mu.Unlock()
			r.broadcastSystemMsg(client.username + " left the room")

		case msg := <-r.forward:
			r.mu.RLock()
			data, _ := json.Marshal(msg)
			for client := range r.clients {
				select {
				case client.send <- data:
				default:
					r.mu.RUnlock()
					r.mu.Lock()
					delete(r.clients, client)
					close(client.send)
					r.mu.Unlock()
					r.mu.RLock()
				}
			}
			r.mu.RUnlock()
		}
	}
}

func (r *Room) Join(client *Client) {
	r.join <- client
	client.rooms[r.ID] = r
}

func (r *Room) Leave(client *Client) {
	r.leave <- client
	delete(client.rooms, r.ID)
}

func (r *Room) Broadcast(msg *Message) {
	r.forward <- msg
}

func (r *Room) broadcastSystemMsg(content string) {
	msg := &Message{
		Type:      "system",
		RoomID:    r.ID,
		Content:   content,
		Timestamp: time.Now().UnixMilli(),
	}
	r.forward <- msg
}

架構3：HTTP升級與WebSocket服務端

// internal/chat/server.go
package chat

import (
	"log"
	"net/http"

	"github.com/gorilla/websocket"
)

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

type Server struct {
	hub *Hub
}

func NewServer(hub *Hub) *Server {
	return &Server{hub: hub}
}

func (s *Server) ServeWS(w http.ResponseWriter, r *http.Request) {
	userID := r.URL.Query().Get("userId")
	username := r.URL.Query().Get("username")
	if userID == "" || username == "" {
		http.Error(w, "userId and username required", http.StatusBadRequest)
		return
	}

	conn, err := upgrader.Upgrade(w, r, nil)
	if err != nil {
		log.Printf("upgrade error: %v", err)
		return
	}

	client := NewClient(s.hub, conn, userID, username)
	s.hub.register <- client

	go client.WritePump()
	go client.ReadPump()
}

架構4：Redis Pub/Sub跨實例訊息分發

// internal/chat/redis_pubsub.go
package chat

import (
	"context"
	"encoding/json"
	"log"

	"github.com/redis/go-redis/v9"
)

type RedisBroker struct {
	client  *redis.Client
	hub     *Hub
	channel string
}

func NewRedisBroker(addr, channel string, hub *Hub) *RedisBroker {
	rdb := redis.NewClient(&redis.Options{
		Addr:     addr,
		PoolSize: 10,
	})

	return &RedisBroker{
		client:  rdb,
		hub:     hub,
		channel: channel,
	}
}

func (rb *RedisBroker) Start(ctx context.Context) error {
	sub := rb.client.Subscribe(ctx, rb.channel)
	_, err := sub.Receive(ctx)
	if err != nil {
		return err
	}

	ch := sub.Channel()
	go func() {
		for msg := range ch {
			var m Message
			if err := json.Unmarshal([]byte(msg.Payload), &m); err != nil {
				log.Printf("redis message parse error: %v", err)
				continue
			}

			if room, ok := rb.hub.rooms[m.RoomID]; ok {
				room.forward <- &m
			}
		}
	}()

	return nil
}

func (rb *RedisBroker) Publish(ctx context.Context, msg *Message) error {
	data, err := json.Marshal(msg)
	if err != nil {
		return err
	}
	return rb.client.Publish(ctx, rb.channel, data).Err()
}

架構5：訊息持久化與歷史查詢

// internal/chat/repository.go
package chat

import (
	"context"
	"database/sql"
	"fmt"

	_ "github.com/lib/pq"
)

type MessageRepository struct {
	db *sql.DB
}

func NewMessageRepository(databaseURL string) (*MessageRepository, error) {
	db, err := sql.Open("postgres", databaseURL)
	if err != nil {
		return nil, fmt.Errorf("failed to open database: %w", err)
	}
	return &MessageRepository{db: db}, nil
}

func (r *MessageRepository) Save(ctx context.Context, msg *Message) error {
	_, err := r.db.ExecContext(ctx,
		`INSERT INTO chat_messages (room_id, sender, content, type, timestamp)
		 VALUES ($1, $2, $3, $4, $5)`,
		msg.RoomID, msg.Sender, msg.Content, msg.Type, msg.Timestamp,
	)
	return err
}

func (r *MessageRepository) GetHistory(ctx context.Context, roomID string, limit, offset int) ([]*Message, error) {
	rows, err := r.db.QueryContext(ctx,
		`SELECT room_id, sender, content, type, timestamp
		 FROM chat_messages
		 WHERE room_id = $1
		 ORDER BY timestamp DESC
		 LIMIT $2 OFFSET $3`,
		roomID, limit, offset,
	)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	var messages []*Message
	for rows.Next() {
		msg := &Message{}
		if err := rows.Scan(&msg.RoomID, &msg.Sender, &msg.Content, &msg.Type, &msg.Timestamp); err != nil {
			return nil, err
		}
		messages = append(messages, msg)
	}
	return messages, nil
}

避坑指南

坑1：send channel緩衝區溢位

// ❌ 錯誤：無緩衝channel，慢客戶端阻塞廣播
send chan []byte

// ✅ 正確：帶緩衝channel + 溢位踢出
send chan []byte  // buffered channel with 256 capacity
select {
case client.send <- data:
default:
    delete(r.clients, client)
    close(client.send)
}

坑2：缺少心跳偵測

// ❌ 錯誤：不設定ping/pong，無法偵測斷開連線
c.conn.SetReadDeadline(time.Time{})

// ✅ 正確：設定ping/pong心跳
c.conn.SetReadDeadline(time.Now().Add(pongWait))
c.conn.SetPongHandler(func(string) error {
    c.conn.SetReadDeadline(time.Now().Add(pongWait))
    return nil
})

坑3：併發寫入WebSocket

// ❌ 錯誤：多個goroutine同時寫入
go func() { c.conn.WriteMessage(1, data) }()
go func() { c.conn.WriteMessage(1, data2) }()

// ✅ 正確：透過channel串行化寫入
c.send <- data

坑4：未處理Close訊息

// ❌ 錯誤：忽略關閉幀
_, message, err := c.conn.ReadMessage()

// ✅ 正確：區分正常關閉和異常
if err != nil {
    if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway) {
        log.Printf("abnormal close: %v", err)
    }
    break
}

坑5：Redis Pub/Sub訊息遺失

// ❌ 錯誤：先廣播再持久化，Redis斷連時訊息遺失
room.Broadcast(msg)
redisBroker.Publish(ctx, msg)

// ✅ 正確：先持久化再廣播
msgRepo.Save(ctx, msg)
room.Broadcast(msg)
redisBroker.Publish(ctx, msg)

報錯排查

序號 報錯資訊 原因 解決方法
1 websocket: close sent 連線已關閉 檢查客戶端是否斷開
2 write: broken pipe 寫入已關閉連線 WritePump中處理close channel
3 read: connection reset 客戶端異常斷開 ReadPump中defer清理
4 upgrade: Origin check failed CORS檢查失敗 設定upgrader.CheckOrigin
5 concurrent write to websocket 併發寫入 使用channel串行化
6 channel full send緩衝區滿 增大緩衝或踢出慢客戶端
7 redis: connection refused Redis未啟動 檢查Redis連線設定
8 OOM killed 記憶體溢位 限制單機連線數和訊息緩衝
9 too many open files 檔案描述符耗盡 調大ulimit -n
10 ping timeout 心跳逾時 調整pongWait時間

進階最佳化

  1. 連線限流：使用令牌桶限制單IP連線速率，防止惡意連線攻擊
  2. 訊息壓縮：使用permessage-deflate擴充套件壓縮WebSocket幀，減少頻寬
  3. 分片廣播：大房間按連線分片廣播，減少單goroutine壓力
  4. 訊息ACK：實現訊息確認機制，保證訊息至少投遞一次
  5. 連線遷移：使用Redis記錄連線對映，支援服務端重啟後客戶端重連

對比分析

維度 gorilla/websocket nhooyr.io/websocket gobwas/ws
API易用性 ⭐⭐⭐⭐ ⭐⭐⭐⭐⭐ ⭐⭐⭐
效能 ⭐⭐⭐⭐ ⭐⭐⭐⭐ ⭐⭐⭐⭐⭐
記憶體佔用 ⭐⭐⭐ ⭐⭐⭐⭐ ⭐⭐⭐⭐⭐
Context支援 ⭐⭐ ⭐⭐⭐⭐⭐ ⭐⭐⭐
壓縮支援 ⭐⭐⭐⭐ ⭐⭐⭐⭐⭐ ⭐⭐⭐⭐
社群活躍度 ⭐⭐⭐⭐⭐ ⭐⭐⭐⭐ ⭐⭐⭐
文件品質 ⭐⭐⭐⭐ ⭐⭐⭐⭐⭐ ⭐⭐⭐

總結：Go的goroutine模型是構建高併發WebSocket服務的天然優勢。Hub-Client連線管理→Room訊息廣播→Redis Pub/Sub跨實例→訊息持久化→水平擴展五位一體，讓你從零構建支援10萬+線上的即時聊天系統。核心要點：串行化寫入、心跳偵測、背壓控制、先持久化再廣播。

線上工具推薦

本站提供瀏覽器本地工具，免註冊即可試用 →

#Go WebSocket#实时聊天#gorilla/websocket#Go实时通信#2026#技术架构