go-redis v9接收Redis客户端跟踪失效消息报错：unsupported pubsub message: "invalidate"

我尝试通过Redis的Client Tracking功能从本地Redis部署接收失效消息——Redis会在__redis__:invalidate Pub/Sub频道发布已更新的key，用于从客户端缓存中驱逐对应key。但使用go-redis v9时会抛出**unsupported pubsub message: "invalidate"**错误，v8版本则可正常运行。

原测试代码 #

package main

import (
    "context"
    "fmt"
    "time"

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

func main() {
    ctx := context.Background()
    redisOptions := &redis.Options{
        Addr: "localhost:6379",
        OnConnect: func(ctx context.Context, cn *redis.Conn) error {
            cid := cn.ClientID(ctx).Val()
            return cn.Process(ctx, redis.NewBoolCmd(ctx, "CLIENT", "TRACKING", "ON", "REDIRECT", fmt.Sprintf("%v", cid), "BCAST", "PREFIX", "my"))
        },
    }

    redisClient := redis.NewClient(redisOptions)

    pubsub := redisClient.Subscribe(ctx, "__redis__:invalidate")

    go func() {
        defer pubsub.Close()
        for {
            fmt.Println("Listening...")
            msg, err := pubsub.ReceiveMessage(ctx)
            if err != nil {
                fmt.Println(err) // 此处抛出错误
            }
            fmt.Println(msg)
        }
    }()

    time.Sleep(30 * time.Second)
}

复现步骤 #

• 启动本地Redis服务器
• 运行上述脚本
• 在Redis CLI中执行命令 SET my-key "val"

问题原因 #

go-redis v9对Pub/Sub消息的处理逻辑做了调整：默认的ReceiveMessage方法仅支持解析标准的Pub/Sub消息，而Client Tracking发送的invalidate属于特殊控制消息，v9未默认兼容这种类型的消息解析，v8版本则对此做了内置支持。

解决方案 #

使用Receive方法替代ReceiveMessage，该方法可以接收所有类型的Redis消息，再通过类型断言手动处理InvalidatedMessage类型的失效消息：

package main

import (
    "context"
    "fmt"
    "time"

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

func main() {
    ctx := context.Background()
    redisOptions := &redis.Options{
        Addr: "localhost:6379",
        OnConnect: func(ctx context.Context, cn *redis.Conn) error {
            cid := cn.ClientID(ctx).Val()
            return cn.Process(ctx, redis.NewBoolCmd(ctx, "CLIENT", "TRACKING", "ON", "REDIRECT", fmt.Sprintf("%v", cid), "BCAST", "PREFIX", "my"))
        },
    }

    redisClient := redis.NewClient(redisOptions)

    pubsub := redisClient.Subscribe(ctx, "__redis__:invalidate")

    go func() {
        defer pubsub.Close()
        for {
            fmt.Println("Listening...")
            msg, err := pubsub.Receive(ctx)
            if err != nil {
                fmt.Println(err)
                continue
            }

            switch m := msg.(type) {
            case *redis.Message:
                fmt.Printf("标准消息: 频道=%s, 内容=%s\n", m.Channel, m.Payload)
            case *redis.Subscription:
                fmt.Printf("订阅状态变更: 频道=%s, 类型=%s, 数量=%d\n", m.Channel, m.Kind, m.Count)
            case *redis.InvalidatedMessage:
                // 处理Client Tracking的失效消息
                fmt.Printf("失效key列表: %v\n", m.Keys)
            default:
                fmt.Printf("未知消息类型: %T\n", m)
            }
        }
    }()

    time.Sleep(30 * time.Second)
}

修改说明 #

1. 替换ReceiveMessage为Receive，获取全类型Redis消息
2. 通过类型断言区分消息类型：
   • *redis.InvalidatedMessage对应Client Tracking的失效消息，Keys字段包含被更新/删除的key列表
   • 保留对标准Pub/Sub消息和订阅状态变更消息的处理逻辑

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