Golang中如何实现发布变更与订阅变更的设计模式

在Go中实现用户专属的发布-订阅模式，核心是建立用户ID与连接之间的映射关系。以下是具体实现思路和示例代码：

核心架构

package main

import (
    "log"
    "net/http"
    "sync"
    "github.com/gorilla/websocket"
)

// 用户连接管理器
type ClientManager struct {
    clients    map[string]*websocket.Conn // userID -> connection
    broadcast  chan UserMessage
    register   chan Client
    unregister chan string
    mu         sync.RWMutex
}

type Client struct {
    UserID string
    Conn   *websocket.Conn
}

type UserMessage struct {
    UserID  string
    Message []byte
}

连接注册与用户绑定

func (manager *ClientManager) Start() {
    for {
        select {
        case client := <-manager.register:
            manager.mu.Lock()
            manager.clients[client.UserID] = client.Conn
            manager.mu.Unlock()
            log.Printf("User %s registered", client.UserID)

        case userID := <-manager.unregister:
            manager.mu.Lock()
            if conn, ok := manager.clients[userID]; ok {
                conn.Close()
                delete(manager.clients, userID)
            }
            manager.mu.Unlock()

        case msg := <-manager.broadcast:
            manager.mu.RLock()
            if conn, ok := manager.clients[msg.UserID]; ok {
                if err := conn.WriteMessage(websocket.TextMessage, msg.Message); err != nil {
                    conn.Close()
                    delete(manager.clients, msg.UserID)
                }
            }
            manager.mu.RUnlock()
        }
    }
}

WebSocket处理器

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

func (manager *ClientManager) ServeWS(w http.ResponseWriter, r *http.Request) {
    conn, err := upgrader.Upgrade(w, r, nil)
    if err != nil {
        log.Println(err)
        return
    }

    // 从认证token或请求参数中获取用户ID
    userID := r.URL.Query().Get("user_id")
    if userID == "" {
        conn.Close()
        return
    }

    client := Client{
        UserID: userID,
        Conn:   conn,
    }

    manager.register <- client

    defer func() {
        manager.unregister <- userID
    }()

    // 保持连接活跃
    for {
        _, _, err := conn.ReadMessage()
        if err != nil {
            break
        }
    }
}

发布用户专属消息

func (manager *ClientManager) SendToUser(userID string, message []byte) {
    msg := UserMessage{
        UserID:  userID,
        Message: message,
    }
    manager.broadcast <- msg
}

// 使用示例
func main() {
    manager := &ClientManager{
        clients:    make(map[string]*websocket.Conn),
        broadcast:  make(chan UserMessage),
        register:   make(chan Client),
        unregister: make(chan string),
    }

    go manager.Start()

    http.HandleFunc("/ws", manager.ServeWS)
    
    // 模拟业务逻辑：向特定用户发送更新
    go func() {
        // 当Joe的数据变更时
        manager.SendToUser("joe123", []byte(`{"type":"order_update","data":{...}}`))
    }()

    log.Fatal(http.ListenAndServe(":8080", nil))
}

消息路由扩展

对于更复杂的路由需求，可以使用多级映射：

type SubscriptionManager struct {
    userSubscriptions map[string]map[string]bool // userID -> [channel]bool
    channelSubscribers map[string]map[string]bool // channel -> [userID]bool
    mu sync.RWMutex
}

func (sm *SubscriptionManager) Subscribe(userID, channel string) {
    sm.mu.Lock()
    defer sm.mu.Unlock()
    
    if sm.userSubscriptions[userID] == nil {
        sm.userSubscriptions[userID] = make(map[string]bool)
    }
    sm.userSubscriptions[userID][channel] = true
    
    if sm.channelSubscribers[channel] == nil {
        sm.channelSubscribers[channel] = make(map[string]bool)
    }
    sm.channelSubscribers[channel][userID] = true
}

func (sm *SubscriptionManager) Publish(channel string, message []byte) {
    sm.mu.RLock()
    defer sm.mu.RUnlock()
    
    if subscribers, ok := sm.channelSubscribers[channel]; ok {
        for userID := range subscribers {
            // 通过ClientManager发送给具体用户
            clientManager.SendToUser(userID, message)
        }
    }
}

连接心跳保持

func (c *Client) KeepAlive() {
    ticker := time.NewTicker(30 * time.Second)
    defer ticker.Stop()
    
    for {
        select {
        case <-ticker.C:
            if err := c.Conn.WriteControl(
                websocket.PingMessage,
                []byte{},
                time.Now().Add(10*time.Second),
            ); err != nil {
                return
            }
        }
    }
}

这种设计模式的关键点：

1. 每个WebSocket连接与用户ID强绑定
2. 通过用户ID到连接映射实现精准消息投递
3. 支持用户订阅多个频道，实现细粒度控制
4. 使用读写锁保证并发安全

实际部署时需要考虑连接重连、消息持久化、集群扩展等问题，但以上代码提供了用户专属发布-订阅的核心实现框架。