Redis 配置与高可用

Asynq

Redis 配置与高可用

1 连接配置总览

Asynq 提供三种 Redis 连接模式：

模式	配置类型	适用场景
单机	`RedisClientOpt`	开发环境、单实例部署
Sentinel 哨兵	`RedisFailoverClientOpt`	生产环境高可用、自动故障转移
Cluster 集群	`RedisClusterClientOpt`	大规模数据、水平扩展

2 单机模式（RedisClientOpt）

2.1 全部字段

type RedisClientOpt struct {
    Addr         string          // Redis 地址，格式 "host:port"
    Password     string          // Redis 密码
    DB           int             // Redis 数据库编号（0-15，默认 0）
    Username     string          // Redis ACL 用户名（Redis 6.0+）
    DialTimeout  time.Duration   // 连接超时（默认 5s）
    ReadTimeout  time.Duration   // 读超时（默认 3s，-1 表示无超时）
    WriteTimeout time.Duration   // 写超时（默认等于 ReadTimeout，-1 表示无超时）
    TLSConfig    *tls.Config     // TLS 配置
    PoolSize     int             // 连接池大小
}

2.2 基础连接

// 最简配置
client := asynq.NewClient(asynq.RedisClientOpt{
    Addr: "localhost:6379",
})

// 带密码和 DB
client := asynq.NewClient(asynq.RedisClientOpt{
    Addr:     "localhost:6379",
    Password: os.Getenv("REDIS_PASSWORD"),
    DB:       0,
})

// 完整配置
client := asynq.NewClient(asynq.RedisClientOpt{
    Addr:         "redis-prod.internal:6379",
    Password:     os.Getenv("REDIS_PASSWORD"),
    Username:     "asynq_user",            // ACL 用户名
    DB:           1,                        // 使用 DB 1
    DialTimeout:  5 * time.Second,
    ReadTimeout:  3 * time.Second,
    WriteTimeout: 3 * time.Second,
    PoolSize:     50,                       // 连接池大小
})

2.3 连接池配置

asynq.RedisClientOpt{
    Addr:     "localhost:6379",
    PoolSize: 50, // 建议 >= Concurrency 值
}

⚡ 性能提示：PoolSize 应至少等于 Concurrency。如果多个 Client/Server 共享同一个 Redis，连接池大小需要适当放大。

3 Sentinel 哨兵模式（RedisFailoverClientOpt）

3.1 配置结构

type RedisFailoverClientOpt struct {
    MasterName       string          // 主节点名称（必填）
    SentinelAddrs    []string        // Sentinel 节点地址列表
    SentinelPassword string          // Sentinel 密码（如设置）
    Password         string          // Redis 主节点密码
    Username         string          // Redis ACL 用户名
    DB               int             // 数据库编号
    DialTimeout      time.Duration   // 连接超时
    ReadTimeout      time.Duration   // 读超时
    WriteTimeout     time.Duration   // 写超时
    TLSConfig        *tls.Config     // TLS 配置
    PoolSize         int             // 连接池大小
}

3.2 基本用法

redisOpt := asynq.RedisFailoverClientOpt{
    MasterName:    "mymaster",
    SentinelAddrs: []string{
        "sentinel1:26379",
        "sentinel2:26379",
        "sentinel3:26379",
    },
    Password: os.Getenv("REDIS_PASSWORD"),
}

// Client 和 Server 使用相同的配置
client := asynq.NewClient(redisOpt)
srv := asynq.NewServer(redisOpt, cfg)

3.3 Sentinel 架构示意

┌──────────────┐  ┌──────────────┐  ┌──────────────┐
│ Sentinel 1   │  │ Sentinel 2   │  │ Sentinel 3   │
│ :26379       │  │ :26379       │  │ :26379       │
└──────┬───────┘  └──────┬───────┘  └──────┬───────┘
       │                 │                 │
       └─────────────────┼─────────────────┘
                         │ 监控 & 故障转移
                         ▼
       ┌─────────────────────────────────┐
       │         Redis 主从集群           │
       │  Master  :6379                  │
       │  Replica1:6379                  │
       │  Replica2:6379                  │
       └─────────────────────────────────┘

3.4 Sentinel 相关错误处理

srv := asynq.NewServer(redisOpt, asynq.Config{
    HealthCheckFunc: func(err error) {
        if err != nil {
            log.Printf("Redis health check failed: %v", err)
            // Sentinel 故障转移期间可能出现短暂连接失败
            // 属于正常行为，Asynq 会自动重连
        }
    },
})

4 Cluster 集群模式（RedisClusterClientOpt）

4.1 配置结构

type RedisClusterClientOpt struct {
    Addrs        []string        // 集群节点地址列表
    Password     string          // 集群密码
    Username     string          // Redis ACL 用户名
    DialTimeout  time.Duration   // 连接超时
    ReadTimeout  time.Duration   // 读超时
    WriteTimeout time.Duration   // 写超时
    TLSConfig    *tls.Config     // TLS 配置
}

4.2 基本用法

redisOpt := asynq.RedisClusterClientOpt{
    Addrs: []string{
        "node1:7000",
        "node2:7001",
        "node3:7002",
        "node4:7003",
    },
    Password: os.Getenv("REDIS_CLUSTER_PASSWORD"),
}

client := asynq.NewClient(redisOpt)
srv := asynq.NewServer(redisOpt, cfg)

4.3 Cluster 兼容性注意事项

🚨 Redis Cluster 限制：

Asynq 的部分 Lua 脚本涉及跨 slot 操作，在 Cluster 模式下可能存在兼容性问题
Cluster 模式下，所有 key 需要属于同一个 hash slot
生产环境使用 Cluster 前，务必在测试环境充分验证

💡 推荐：对于大多数场景，Sentinel 模式即可满足高可用需求（避免了 Cluster 的跨 slot 问题）。仅在单实例 Redis 无法承载数据量时再考虑 Cluster。

4.4 Cluster 架构示意

┌──────────────────────────────────────────┐
│            Redis Cluster                 │
│  ┌─────────┐ ┌─────────┐ ┌─────────┐   │
│  │ Master A│ │ Master B│ │ Master C│   │
│  │ :7000   │ │ :7001   │ │ :7002   │   │
│  └────┬────┘ └────┬────┘ └────┬────┘   │
│       │           │           │         │
│  ┌────┴────┐ ┌────┴────┐ ┌────┴────┐   │
│  │Slave A1 │ │Slave B1 │ │Slave C1 │   │
│  │ :7003   │ │ :7004   │ │ :7005   │   │
│  └─────────┘ └─────────┘ └─────────┘   │
└──────────────────────────────────────────┘

5 TLS / SSL 安全连接

5.1 基础 TLS 配置

tlsConfig := &tls.Config{
    // 单向 TLS：验证服务端
    InsecureSkipVerify: false, // 生产环境必须为 false
}

redisOpt := asynq.RedisClientOpt{
    Addr:      "redis-prod.internal:6380",
    TLSConfig: tlsConfig,
}

5.2 双向 TLS（mTLS）

// 加载客户端证书
cert, err := tls.LoadX509KeyPair("client.crt", "client.key")
if err != nil {
    log.Fatal(err)
}

tlsConfig := &tls.Config{
    Certificates: []tls.Certificate{cert},
    RootCAs:      loadCA("ca.crt"),
    MinVersion:   tls.VersionTLS12,
}

redisOpt := asynq.RedisClientOpt{
    Addr:      "redis-prod.internal:6380",
    TLSConfig: tlsConfig,
}

5.3 Sentinel + TLS

redisOpt := asynq.RedisFailoverClientOpt{
    MasterName:    "mymaster",
    SentinelAddrs: []string{"sentinel1:26379"},
    TLSConfig: &tls.Config{
        MinVersion: tls.VersionTLS12,
    },
}

6 Redis ACL 认证

Redis 6.0+

redisOpt := asynq.RedisClientOpt{
    Addr:     "localhost:6379",
    Username: "asynq_worker", // ACL 用户名
    Password: "secure_password",
}

Redis ACL 配置示例：

# redis.conf 或 ACL 文件
user asynq_worker on >secure_password ~asynq:* +@all -@dangerous
user asynq_client on >client_password ~asynq:* +@write +@read -@dangerous

7 连接参数调优

7.1 超时配置

asynq.RedisClientOpt{
    Addr:         "redis-prod.internal:6379",
    DialTimeout:  5 * time.Second,  // 建立连接的超时
    ReadTimeout:  5 * time.Second,  // 读取响应的超时（-1 禁用）
    WriteTimeout: 5 * time.Second,  // 发送命令的超时（-1 禁用）
}

场景	建议 ReadTimeout
短任务（<1s）	3s
中等任务（1~30s）	10s
长任务（30s+）	-1（禁用）或 60s+

🚨 陷阱：ReadTimeout 太短（如 1s）可能导致 Redis 命令被截断。如果任务处理时间较长，建议将 ReadTimeout 设置得足够大或设为 -1。

7.2 连接池调优

asynq.RedisClientOpt{
    PoolSize:     100,  // 连接池大小（建议 >= Concurrency × Server实例数）
}

⚡ 性能提示：连接池是复用 TCP 连接的关键。如果观察到大量 TIME_WAIT 或连接建立耗时，增大 PoolSize。

8 环境变量管理

func NewRedisOpt() asynq.RedisClientOpt {
    return asynq.RedisClientOpt{
        Addr:     os.Getenv("REDIS_ADDR"),
        Password: os.Getenv("REDIS_PASSWORD"),
        DB:       getEnvInt("REDIS_DB", 0),
        PoolSize: getEnvInt("REDIS_POOL_SIZE", 50),
    }
}

func getEnvInt(key string, defaultVal int) int {
    if v := os.Getenv(key); v != "" {
        n, _ := strconv.Atoi(v)
        return n
    }
    return defaultVal
}

9 模式选择决策树

需要高可用？
    │
    ├── 否 → 单机模式 (RedisClientOpt)
    │
    └── 是
        ├── 数据量 < 单机内存
        │   └── Sentinel 模式 (RedisFailoverClientOpt) ✅ 推荐
        │
        └── 数据量 > 单机内存
            └── Cluster 模式 (RedisClusterClientOpt)
                ⚠️ 需验证 Lua 脚本兼容性

10 完整高可用配置示例

func NewRedisConnOpt() asynq.RedisConnOpt {
    mode := os.Getenv("REDIS_MODE") // "standalone" | "sentinel" | "cluster"

    switch mode {
    case "sentinel":
        return asynq.RedisFailoverClientOpt{
            MasterName:    os.Getenv("REDIS_MASTER_NAME"),
            SentinelAddrs: strings.Split(os.Getenv("REDIS_SENTINEL_ADDRS"), ","),
            Password:      os.Getenv("REDIS_PASSWORD"),
            DB:            0,
            DialTimeout:   5 * time.Second,
            ReadTimeout:   10 * time.Second,
            WriteTimeout:  10 * time.Second,
            PoolSize:      100,
        }

    case "cluster":
        return asynq.RedisClusterClientOpt{
            Addrs:    strings.Split(os.Getenv("REDIS_CLUSTER_ADDRS"), ","),
            Password: os.Getenv("REDIS_PASSWORD"),
        }

    default:
        return asynq.RedisClientOpt{
            Addr:     os.Getenv("REDIS_ADDR"),
            Password: os.Getenv("REDIS_PASSWORD"),
            DB:       0,
        }
    }
}

// 使用
func main() {
    redisOpt := NewRedisConnOpt()

    client := asynq.NewClient(redisOpt)
    defer client.Close()

    srv := asynq.NewServer(redisOpt, asynq.Config{Concurrency: 20})

    // ...
}

并发控制与限流监控与运维