AWS ElastiCache Redis集群模式下Pub/Sub重复消息问题排查

问题背景 #

我在聊天应用中使用github.com/redis/go-redis/v9 v9.3.1实现Redis操作，通过Redis Pub/Sub完成用户输入状态提示和消息投递功能：每台服务器启动一个专属订阅协程监听指定频道，将频道消息推送给所有WebSocket连接的客户端。目前遇到的问题是终端用户/测试人员会收到大量重复消息。

排查后定位到诱因：Redis集群模式下，Pub/Sub消息会被复制到所有从节点，而当前客户端连接了集群中的所有节点，导致同一条消息被多次接收。

相关代码如下：

var redisClient = redis.NewClusterClient(&redis.ClusterOptions{
    Addrs: []string{"c1:6379",
        "c2:6379",
        "c3:6379"},

    ReadOnly:       false,
    RouteRandomly:  false,
    RouteByLatency: false,
    PoolSize:       10,
    MinIdleConns:   10,

    DialTimeout:  5 * time.Second,
    ReadTimeout:  3 * time.Second,
    WriteTimeout: 3 * time.Second,
    PoolTimeout:  4 * time.Second,

    //IdleCheckFrequency: 60 * time.Second,
    //IdleTimeout:        5 * time.Minute,
    //MaxConnAge:         0 * time.Second,

    MaxRetries:      10,
    MinRetryBackoff: 8 * time.Millisecond,
    MaxRetryBackoff: 512 * time.Millisecond,

    TLSConfig: &tls.Config{
        InsecureSkipVerify: true,
    },
})

func InitializeStatusMessageSubscriber(statusTopic string, cm *ConnectionManager) {
    subscriber := redisClient.Subscribe(ctx, statusTopic)
    log.Printf("Subscribing to status subscriber in go-routine %s", statusTopic)
    var socketEventTypingStatus SocketEventTypingMessage

    channel := subscriber.Channel()

    for msg := range channel {
        if marshalError := json.Unmarshal([]byte(msg.Payload), &socketEventTypingStatus); marshalError != nil {
            log.Println("Error unmarshalling message to SocketEventTypingStatus message", marshalError)
            return
        }
        messageId := socketEventTypingStatus.Id
        expired := CheckIfKeyIsExpired(messageId)
        if expired {
            continue
        }

        id := socketEventTypingStatus.ChatId

        users := GetAllUsersForAGivenChat(id)

        for _, user := range users {
            cm.WriteMessageToUserSocket(user, msg.Payload)
        }
    }
}

注：InitializeStatusMessageSubscriber仅初始化一次

核心原因 #

Redis集群的Pub/Sub机制中，主节点收到的消息会同步到该分片的所有从节点。而默认的集群客户端订阅逻辑会在集群的每个节点上建立订阅连接，导致同一条消息从主节点和多个从节点被重复接收，最终推送给客户端时出现重复。

解决方法 #

1. 单独创建订阅专用的集群客户端，强制路由到主节点 #

创建一个专门用于订阅的客户端，通过自定义路由策略，让订阅请求只发送到集群中的一个主节点（集群内主节点会自动同步Pub/Sub消息到其他主节点，无需订阅多个）。

修改客户端初始化代码：

import "errors"

// 自定义路由函数：让订阅请求只发往任意一个主节点
func subscribeRoute(cmd redis.Cmder, topology *redis.ClusterTopology) ([]string, error) {
    // 筛选所有主节点地址
    var masters []string
    for _, node := range topology.Nodes {
        if node.IsMaster() {
            masters = append(masters, node.Addr)
        }
    }
    // 返回第一个主节点地址即可
    if len(masters) > 0 {
        return []string{masters[0]}, nil
    }
    return nil, errors.New("no master nodes available in cluster")
}

// 初始化订阅专用客户端
var redisSubClient = redis.NewClusterClient(&redis.ClusterOptions{
    Addrs: []string{"c1:6379", "c2:6379", "c3:6379"},
    Route: subscribeRoute, // 应用自定义路由策略
    ReadOnly: false,
    RouteRandomly: false,
    RouteByLatency: false,
    PoolSize:       10,
    MinIdleConns:   10,
    // 其他超时、重试、TLS配置保持和原客户端一致
    DialTimeout:  5 * time.Second,
    ReadTimeout:  3 * time.Second,
    WriteTimeout: 3 * time.Second,
    PoolTimeout:  4 * time.Second,
    MaxRetries:      10,
    MinRetryBackoff: 8 * time.Millisecond,
    MaxRetryBackoff: 512 * time.Millisecond,
    TLSConfig: &tls.Config{
        InsecureSkipVerify: true,
    },
})

2. 修改订阅逻辑使用专用客户端 #

将原来的redisClient.Subscribe替换为redisSubClient.Subscribe：

func InitializeStatusMessageSubscriber(statusTopic string, cm *ConnectionManager) {
    // 使用订阅专用客户端
    subscriber := redisSubClient.Subscribe(ctx, statusTopic)
    log.Printf("Subscribing to status subscriber in go-routine %s", statusTopic)
    var socketEventTypingStatus SocketEventTypingMessage

    channel := subscriber.Channel()

    for msg := range channel {
        if marshalError := json.Unmarshal([]byte(msg.Payload), &socketEventTypingStatus); marshalError != nil {
            log.Println("Error unmarshalling message to SocketEventTypingStatus message", marshalError)
            return
        }
        messageId := socketEventTypingStatus.Id
        expired := CheckIfKeyIsExpired(messageId)
        if expired {
            continue
        }

        id := socketEventTypingStatus.ChatId

        users := GetAllUsersForAGivenChat(id)

        for _, user := range users {
            cm.WriteMessageToUserSocket(user, msg.Payload)
        }
    }
}

3. 额外的幂等保障（可选但推荐） #

即使解决了订阅重复的问题，也建议保留或完善现有的CheckIfKeyIsExpired(messageId)逻辑，通过消息ID做幂等处理。比如用Redis缓存已处理的消息ID，设置合理的过期时间（比如和输入状态的有效期一致），彻底避免任何场景下的重复消息推送。

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