WebSocket协议与HTTP升级握手原理
WebSocket协议在单个TCP连接上提供全双工通信，通过HTTP Upgrade机制建立连接。客户端发送带有Upgrade: websocket头的HTTP请求，服务端返回101 Switching Protocols响应完成握手。与HTTP轮询和SSE相比，WebSocket的优势在于双向通信和低延迟，适合实时聊天、协同编辑、实时通知和股票行情推送等场景。Go语言标准库没有内置WebSocket支持，社区广泛使用gorilla/websocket库。
# 安装gorilla/websocket
go get github.com/gorilla/websocketWebSocket服务端基础配置
WebSocket服务端需要实现HTTP升级握手和消息读写循环。upgrader配置控制连接参数：
package main
import (
"log"
"net/http"
"time"
"github.com/gorilla/websocket"
)
var upgrader = websocket.Upgrader{
ReadBufferSize: 4096,
WriteBufferSize: 4096,
// 允许跨域，生产环境应限制来源
CheckOrigin: func(r *http.Request) bool {
origin := r.Header.Get("Origin")
allowed := []string{"https://app.example.com", "https://admin.example.com"}
for _, o := range allowed {
if origin == o {
return true
}
}
return false
},
HandshakeTimeout: 10 * time.Second,
}
func handleWebSocket(w http.ResponseWriter, r *http.Request) {
// 升级HTTP连接为WebSocket
conn, err := upgrader.Upgrade(w, r, nil)
if err != nil {
log.Printf("升级失败: %v", err)
return
}
defer conn.Close()
// 设置读写超时
conn.SetReadLimit(65536) // 最大消息64KB
for {
messageType, message, err := conn.ReadMessage()
if err != nil {
if websocket.IsUnexpectedCloseError(err,
websocket.CloseGoingAway, websocket.CloseNormalClosure) {
log.Printf("连接异常关闭: %v", err)
}
break
}
log.Printf("收到消息: type=%d, %s", messageType, message)
// 回显消息
if err := conn.WriteMessage(messageType, message); err != nil {
log.Printf("写入失败: %v", err)
break
}
}
}
func main() {
http.HandleFunc("/ws", handleWebSocket)
log.Println("WebSocket服务启动于 :8080")
log.Fatal(http.ListenAndServe(":8080", nil))
}Hub模式连接池管理与广播机制
单连接模式无法满足多用户实时通信需求。Hub模式维护所有活跃连接，负责消息广播和连接生命周期管理。每个连接运行独立的读写Goroutine，Hub通过channel与各连接通信。
package main
import (
"encoding/json"
"log"
"net/http"
"sync"
"time"
"github.com/gorilla/websocket"
)
// Client 表示一个WebSocket连接
type Client struct {
Hub *Hub
Conn *websocket.Conn
Send chan []byte
UserID string
RoomID string
}
// Hub 管理所有连接和广播
type Hub struct {
Clients map[*Client]bool
Register chan *Client
Unregister chan *Client
Broadcast chan []byte
Rooms map[string]map[*Client]bool
mu sync.RWMutex
}
func NewHub() *Hub {
return &Hub{
Clients: make(map[*Client]bool),
Register: make(chan *Client),
Unregister: make(chan *Client),
Broadcast: make(chan []byte, 256),
Rooms: make(map[string]map[*Client]bool),
}
}
func (h *Hub) Run() {
ticker := time.NewTicker(30 * time.Second)
defer ticker.Stop()
for {
select {
case client := <-h.Register:
h.mu.Lock()
h.Clients[client] = true
if client.RoomID != "" {
if h.Rooms[client.RoomID] == nil {
h.Rooms[client.RoomID] = make(map[*Client]bool)
}
h.Rooms[client.RoomID][client] = true
}
h.mu.Unlock()
log.Printf("客户端连接: userID=%s, roomID=%s, 在线=%d",
client.UserID, client.RoomID, len(h.Clients))
case client := <-h.Unregister:
h.mu.Lock()
if _, ok := h.Clients[client]; ok {
delete(h.Clients, client)
if client.RoomID != "" {
delete(h.Rooms[client.RoomID], client)
if len(h.Rooms[client.RoomID]) == 0 {
delete(h.Rooms, client.RoomID)
}
}
close(client.Send)
}
h.mu.Unlock()
log.Printf("客户端断开: userID=%s, 在线=%d",
client.UserID, len(h.Clients))
case message := <-h.Broadcast:
h.mu.RLock()
for client := range h.Clients {
select {
case client.Send <- message:
default:
// 发送缓冲区满，关闭连接
close(client.Send)
delete(h.Clients, client)
}
}
h.mu.RUnlock()
}
}
}
// 广播到指定房间
func (h *Hub) BroadcastToRoom(roomID string, message []byte) {
h.mu.RLock()
defer h.mu.RUnlock()
if clients, ok := h.Rooms[roomID]; ok {
for client := range clients {
select {
case client.Send <- message:
default:
close(client.Send)
delete(clients, client)
}
}
}
}心跳检测与超时断开机制
WebSocket连接可能因网络中断或客户端异常而变成僵尸连接。心跳检测通过定期发送Ping帧检测连接存活状态，超时未收到Pong则主动关闭连接。
// 读Goroutine：处理客户端消息和Pong
func (c *Client) ReadPump() {
defer func() {
c.Hub.Unregister <- c
c.Conn.Close()
}()
// 设置Pong处理和读超时
c.Conn.SetReadDeadline(time.Now().Add(60 * time.Second))
c.Conn.SetPongHandler(func(string) error {
c.Conn.SetReadDeadline(time.Now().Add(60 * time.Second))
return nil
})
for {
_, message, err := c.Conn.ReadMessage()
if err != nil {
break
}
// 解析消息
var msg struct {
Type string `json:"type"`
Content string `json:"content"`
RoomID string `json:"room_id"`
}
if err := json.Unmarshal(message, &msg); err != nil {
continue
}
switch msg.Type {
case "join":
c.RoomID = msg.RoomID
c.Hub.Register <- c
case "message":
broadcastMsg, _ := json.Marshal(map[string]string{
"type": "message",
"user_id": c.UserID,
"content": msg.Content,
"timestamp": time.Now().Format(time.RFC3339),
})
c.Hub.BroadcastToRoom(msg.RoomID, broadcastMsg)
}
}
}
// 写Goroutine：发送消息和心跳Ping
func (c *Client) WritePump() {
ticker := time.NewTicker(30 * time.Second) // 心跳间隔
defer func() {
ticker.Stop()
c.Conn.Close()
}()
for {
select {
case message, ok := <-c.Send:
c.Conn.SetWriteDeadline(time.Now().Add(10 * time.Second))
if !ok {
// Hub关闭了Send通道
c.Conn.WriteMessage(websocket.CloseMessage, []byte{})
return
}
w, err := c.Conn.NextWriter(websocket.TextMessage)
if err != nil {
return
}
w.Write(message)
w.Close()
case <-ticker.C:
// 发送Ping帧
c.Conn.SetWriteDeadline(time.Now().Add(10 * time.Second))
if err := c.Conn.WriteMessage(websocket.PingMessage, nil); err != nil {
return
}
}
}
}心跳间隔30秒，读超时60秒。客户端必须在60秒内回复Pong，否则服务端主动断开连接。WritePump中的default case处理Send缓冲区满的情况，避免Goroutine阻塞。
Nginx反向代理WebSocket配置
生产环境中WebSocket服务通常部署在Nginx之后，需要配置Upgrade和Connection头转发：
upstream websocket_backend {
server 127.0.0.1:8080;
# 多实例负载均衡
server 127.0.0.1:8081;
server 127.0.0.1:8082;
ip_hash; # 确保同一客户端连接到同一后端
keepalive 64;
}
server {
listen 443 ssl http2;
server_name ws.example.com;
ssl_certificate /etc/nginx/ssl/cert.pem;
ssl_certificate_key /etc/nginx/ssl/key.pem;
location /ws {
proxy_pass http://websocket_backend;
proxy_http_version 1.1;
# WebSocket升级头
proxy_set_header Upgrade $http_upgrade;
proxy_set_header Connection "upgrade";
# 透传客户端信息
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
proxy_set_header X-Forwarded-Proto $scheme;
# 超时配置：长连接需要较长超时
proxy_read_timeout 300s;
proxy_send_timeout 300s;
proxy_connect_timeout 10s;
# 缓冲关闭：WebSocket需要实时透传
proxy_buffering off;
}
}proxy_read_timeout设置较长（300秒）避免Nginx在心跳间隔内主动断开连接。proxy_buffering off关闭响应缓冲，确保消息即时到达客户端。ip_hash保证同一客户端的连接始终路由到同一后端实例，避免连接在不同实例间漂移导致状态丢失。多实例部署时需要在应用层实现连接状态共享或使用Redis Pub/Sub跨实例广播消息。
原创文章，作者：小编，如若转载，请注明出处：https://www.yunthe.com/go-yu-yan-websocket-zhang-lian-jie-guan-li-yu-shi-shi-tong/
