标题：Go 语言批量导入 2 亿 Redis Key 的高性能实践与排障指南

本文针对使用 go（redigo）向 redis 批量写入海量键（如 2 亿）时频繁出现连接重置、EOF、拒绝连接等错误的问题，从内存瓶颈、协议优化、连接管理、批量策略四方面提供可落地的调优方案。

在实际生产中，用 Go 向 redis 导入 2 亿级 Key 是常见但极具挑战性的任务。你遇到的 connection reset by peer、connection refused 和 EOF 并非单纯网络或代码逻辑错误，而往往是 redis 实例因内存耗尽被 OOM Killer 终止、或进入假死状态后拒绝新连接 的典型表现——这正是问题的根本症结。

? 根本原因分析

Redis 官方明确指出：单实例可支持 ≥2.5 亿 keys，但前提是有足够内存。假设每个 key 平均占用 100 字节（含 key 名、value、过期字段、内部结构开销），2 亿 keys 将消耗至少 20 GB 内存（未计 Redis 自身开销及内存碎片）。若服务器物理内存不足或未配置 maxmemory + 合理淘汰策略，Redis 极易被 linux OOM Killer 杀死，导致后续连接全部失败——这正是你每次卡在 3100 万 keys 左右的深层原因（可能对应某次内存临界点）。

✅ 正确实践方案

1. 优先压缩存储结构：改用 Hash 替代独立 Key

避免为每个 key 创建独立的 SET + EXPIRE 开销（每个 key 都有 dictEntry、redisObject 等元数据）。推荐使用 Redis Hash：

// ✅ 推荐：将 1000 个 key 归组到一个 hash 中（例如按前缀分桶） func batchLoadToHash(conn redis.Conn, bucketName string, keys []string) error {     pipe := conn.Send("MULTI")     for _, key := range keys {         // field = key, value = maxCount，TTL 单独设置（hash 本身不支持 per-field TTL）         conn.Send("HSET", bucketName, key, maxCount)     }     conn.Send("EXPIRE", bucketName, numSecondsExpire) // 整个 hash 共享 TTL     _, err := conn.Do("EXEC")     return err }

? 提示：Hash 可降低约 40–60% 内存占用（官方实测），且 HSET 原子性优于多次 SET。

2. 严格控制批量大小 & 连接生命周期

你当前的 RedisServerBatchLoadKeys 存在严重隐患：

• defer conn.Close() 在函数退出时才关闭，但 for 循环内反复调用 GetConnOrPanic 会快速耗尽连接池；
• MULTI/EXEC 包裹全部 200 万 keys？Redis 单次 EXEC 不宜超过 1 万命令（否则阻塞主线程、OOM 风险陡增）。

✅ 正确做法（分片 + 流式提交）：

const batchSize = 5000 // 每批 5000 条命令，平衡吞吐与内存安全  func RedisServerBatchLoadKeys(rtbExchange string, allKeys []string) {     pool := GetPool(rtbExchange) // 复用全局 pool，非每次新建     for i := 0; i < len(allKeys); i += batchSize {         batch := allKeys[i:min(i+batchSize, len(allKeys))]          conn := pool.Get()         defer conn.Close() // ✅ 每批获取/释放连接          if err := conn.Send("MULTI"); err != nil {             log.Fatal("MULTI failed:", err)         }         for _, key := range batch {             conn.Send("SET", key, maxCount)             conn.Send("EXPIRE", key, numSecondsExpire)         }         _, err := conn.Do("EXEC")         if err != nil {             log.Printf("Batch %d-%d failed: %v", i, i+len(batch)-1, err)             // 指数退避重试（最多 3 次），避免雪崩             time.Sleep(time.Second * time.Duration(1<

return &redis.Pool{     MaxIdle:     20,           // 提高空闲连接复用率     MaxActive:   50,           // 允许更多并发写入     IdleTimeout: 300 * time.Second,     Dial: func() (redis.Conn, error) {         c, err := redis.Dial("tcp", server,             redis.DialConnectTimeout(5*time.Second),             redis.DialReadTimeout(10*time.Second),             redis.DialWriteTimeout(10*time.Second),         )         if err != nil {             return nil, err         }         // ✅ 强制设置 DB（避免跨 DB 混淆）         if _, err := c.Do("SELECT", 0); err != nil {             c.Close()             return nil, err         }         return c, nil     },     TestOnBorrow: func(c redis.Conn, t time.Time) error {         _, err := c.Do("PING")         return err     }, }