如何复用Redis Subscriber以减少Redis集群连接数？Golang场景

Redis本身不支持内置的订阅者实例复用机制，也没有提供所谓的"Subscriber ID"来标识并复用已有的订阅连接。这是因为Redis的订阅连接有一个核心特性：一旦进入订阅状态，该连接会被阻塞，只能处理SUBSCRIBE/UNSUBSCRIBE/PSUBSCRIBE这类订阅相关命令，无法执行其他读写操作，且每个订阅连接是独立的会话，无法被多个业务逻辑（比如你的WebSocket连接）直接共享。

但针对你用Golang实现WebSocket连接订阅Redis频道、减少集群连接数的需求，可以通过应用层转发机制来实现订阅连接的复用，具体方案如下：

实现思路 #

1. 全局复用Redis订阅连接：维护少量（甚至单个）Redis订阅客户端，负责监听所有需要的频道，而不是为每个WebSocket连接创建新的订阅实例。
2. 应用层维护频道-连接映射：用线程安全的数据结构（比如Go的sync.Map）存储「Redis频道」到「对应WebSocket连接列表」的映射。
3. 消息转发：当Redis订阅客户端收到频道消息时，遍历该频道对应的WebSocket连接列表，将消息推送给每个连接。
4. 连接生命周期管理：WebSocket连接关闭时，从对应频道的连接列表中移除；若频道的连接列表为空，可选择取消该频道的订阅以节省资源。

Golang示例代码 #

package main

import (
	"context"
	"log"
	"net/http"
	"sync"

	"github.com/go-redis/redis/v8"
	"github.com/gorilla/websocket"
)

var (
	upgrader = websocket.Upgrader{
		CheckOrigin: func(r *http.Request) bool {
			return true // 生产环境需根据实际情况配置跨域规则
		},
	}

	// 频道到WebSocket连接列表的映射，线程安全
	channelConnMap = sync.Map{}

	// Redis订阅客户端
	redisSubClient *redis.Client
)

func init() {
	// 初始化Redis订阅客户端
	redisSubClient = redis.NewClient(&redis.Options{
		Addr: "your-redis-cluster-addr",
		// 其他集群配置，比如Password、DB等
	})

	// 启动订阅消息处理goroutine
	go handleRedisMessages()
}

func handleRedisMessages() {
	ctx := context.Background()
	// 这里可以先订阅基础频道，或者后续动态添加订阅
	pubsub := redisSubClient.Subscribe(ctx, "initial-channel")
	defer pubsub.Close()

	for {
		msg, err := pubsub.ReceiveMessage(ctx)
		if err != nil {
			log.Printf("Redis订阅消息接收失败: %v", err)
			// 处理重连逻辑，比如重新创建订阅客户端
			pubsub = redisSubClient.Subscribe(ctx, getAllChannels()...)
			continue
		}

		// 获取该频道对应的WebSocket连接列表
		if conns, ok := channelConnMap.Load(msg.Channel); ok {
			for _, conn := range conns.([]*websocket.Conn) {
				err := conn.WriteMessage(websocket.TextMessage, []byte(msg.Payload))
				if err != nil {
					log.Printf("推送消息到WebSocket失败: %v", err)
					// 移除失效的WebSocket连接
					removeConnFromChannel(msg.Channel, conn)
				}
			}
		}
	}
}

// 动态订阅新频道（当有WebSocket连接订阅未被监听的频道时调用）
func subscribeChannel(ctx context.Context, channel string) error {
	// 先检查是否已经订阅该频道，避免重复订阅
	// 可以通过维护一个已订阅频道的集合来实现
	// 此处省略检查逻辑，直接订阅
	_, err := redisSubClient.Subscribe(ctx, channel).Receive(ctx)
	return err
}

// 添加WebSocket连接到指定频道的列表
func addConnToChannel(channel string, conn *websocket.Conn) {
	conns, _ := channelConnMap.LoadOrStore(channel, []*websocket.Conn{})
	newConns := append(conns.([]*websocket.Conn), conn)
	channelConnMap.Store(channel, newConns)

	// 若该频道未被订阅，则动态订阅
	// 此处需配合已订阅频道集合检查，省略实现
	// subscribeChannel(context.Background(), channel)
}

// 从指定频道的列表中移除WebSocket连接
func removeConnFromChannel(channel string, conn *websocket.Conn) {
	if conns, ok := channelConnMap.Load(channel); ok {
		connList := conns.([]*websocket.Conn)
		newList := make([]*websocket.Conn, 0, len(connList))
		for _, c := range connList {
			if c != conn {
				newList = append(newList, c)
			}
		}
		if len(newList) == 0 {
			channelConnMap.Delete(channel)
			// 取消该频道的订阅，省略实现
			// redisSubClient.Unsubscribe(context.Background(), channel)
		} else {
			channelConnMap.Store(channel, newList)
		}
	}
}

// WebSocket处理函数
func wsHandler(w http.ResponseWriter, r *http.Request) {
	conn, err := upgrader.Upgrade(w, r, nil)
	if err != nil {
		log.Printf("WebSocket升级失败: %v", err)
		return
	}
	defer conn.Close()

	// 获取客户端请求的频道（比如从URL参数或请求体中获取）
	channel := r.URL.Query().Get("channel")
	if channel == "" {
		conn.WriteMessage(websocket.TextMessage, []byte("缺少频道参数"))
		return
	}

	// 将连接添加到频道列表
	addConnToChannel(channel, conn)
	defer removeConnFromChannel(channel, conn)

	// 处理WebSocket客户端的消息（此处示例忽略客户端消息，只处理推送）
	for {
		_, _, err := conn.ReadMessage()
		if err != nil {
			break
		}
	}
}

func getAllChannels() []string {
	// 获取所有已订阅的频道，省略实现
	return []string{}
}

func main() {
	http.HandleFunc("/ws", wsHandler)
	log.Fatal(http.ListenAndServe(":8080", nil))
}

注意事项 #

• Redis连接重连：要实现订阅客户端的自动重连逻辑，避免因Redis集群故障导致消息推送中断。
• 内存泄漏防范：确保WebSocket连接关闭时，及时从频道连接列表中移除，避免无效连接占用内存。
• 频道分组策略：如果需要监听的频道数量极大，单个Redis订阅连接可能成为性能瓶颈，此时可以按频道前缀或其他规则分组，用多个订阅客户端分担负载。
• 线程安全：由于多个goroutine会同时操作channelConnMap，必须使用线程安全的数据结构（如sync.Map）或加锁保护。

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