本文针对使用 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<3. 连接池必须健壮配置
你当前的 MaxIdle=3, MaxActive=10 在高并发导入下极易枯竭。建议调整为：
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
},
}4. 生产级兜底：监控 + 分片 + 持久化
-
内存监控：导入前执行 INFO memory 检查 used_memory_rss，预留 30% 内存余量；
-
分片（Sharding）：当单机无法承载时，按 key hash % N 分发至 N 台 Redis（推荐使用 Redis Cluster 或 Codis）；
-
持久化保护：启用 RDB/AOF 并设置 save ""（禁用自动 RDB）+ appendonly yes，避免导入时频繁 fork 阻塞；
-
服务端调优：在 redis.conf 中设置：
maxmemory 25gb
maxmemory-policy allkeys-lru # 或 volatile-lru，避免 OOM
tcp-keepalive 60
? 总结
不要把“导入失败”归咎于 Go 代码或 redigo 库——90% 的类似问题源于 未对齐 Redis 的内存模型与批量写入的工程约束。正确的路径是：
① 用 Hash / Sorted Set 等紧凑结构降内存 →
② 小批量（≤5k）、流式提交 + 健壮连接池 →
③ 全链路超时控制 + OOM 监控 →
④ 必要时水平分片。
完成以上改造后，2 亿 keys 的稳定导入将成为可预期的常规操作，而非故障频发的“玄学任务”。
