深入理解 SingleFlight：从单机到分布式的请求合并方案全解析

作者：逆境不可逃

技术永无止境

希望我的内容可以帮助到你！！！！

大家吼 ! 我是 逆境不可逃 今天给大家带来文章

《 深入理解 SingleFlight：从单机到分布式的请求合并方案全解析》

近期文章 欢迎阅读

一篇速成 汇编程序语言设计之 8086 汇编核心指令-CSDN博客

下面进入正题

前言

在高并发后端系统中，大量相同请求同时访问同一资源是再常见不过的场景：热点商品详情、缓存失效瞬间的回源、同参数的 RPC 调用…… 这些重复请求会无端放大下游负载，引发缓存击穿乃至雪崩。

SingleFlight（单飞 / 请求合并） 正是对这类问题最经典的优化：让同一份业务逻辑在并发到达时只执行一次，其余请求共享结果。本文从原理讲到分布式落地，并给出真正生产可用的实现。

一、核心认知

1.1 定义

SingleFlight 是一种并发控制模式：多个并发请求访问同一资源时，仅其中一个真正执行业务逻辑，其余阻塞等待并复用结果。

它本质是同维度请求合并，避免相同逻辑被重复执行，最典型用途就是解决缓存击穿。

1.2 工作原理

请求 A ──┐
请求 B ──┼──→ [按 Key 抢占] ──→ A 抢到，B/C 等待 ──→ A 执行 ──→ 广播结果给 B/C
请求 C ──┘

三个关键步骤：

1.3 适用场景

• 热点数据查询（商品详情、内容流首屏）
• 缓存击穿防护（热 Key 失效瞬间的回源）
• 相同入参的重复 RPC / DB 调用
• 计算密集型任务的结果复用

1.4 固有缺陷（必须正视）

• 单点阻塞：执行节点慢，所有等待方都被拖
• 异常批量传导：执行报错时所有等待方都拿到错误
• 冷热失衡：超低频请求合并意义不大，反而增加协调开销

二、单机版实现

2.1 Go 官方实现

golang.org/x/sync/singleflight 是工业界最经典的单机实现，核心方法 Do(key, fn) 保证同 Key 并发下 fn 仅执行一次。

package main

import (
"fmt"
"golang.org/x/sync/singleflight"
"time"
)

var g singleflight.Group

// 模拟耗时的数据库查询
func queryData(key string) (interface{}, error) {
time.Sleep(1 * time.Second)
return fmt.Sprintf("data-%s", key), nil
}

func main() {
key := "user_1001"
for i := 0; i < 5; i++ {
go func(idx int) {
// 第三个返回值 shared 表示该结果是否被多个调用方共享
val, err, shared := g.Do(key, func() (interface{}, error) {
return queryData(key)
})
fmt.Printf("请求%d 结果=%v err=%v shared=%v\\n", idx, val, err, shared)
}(i)
}
time.Sleep(2 * time.Second)
}

几个容易踩的点：

• Do 是阻塞调用；若执行函数卡死，所有等待方都被卡。需要无阻塞兜底时用 DoChan + select { case <-ctx.Done() }。
• Go 官方实现对 panic 做了 recover，但只在执行 goroutine 内 recover，并通过 panic 二次抛出给所有等待方，调用方仍需自行兜住。
• Forget(key) 可以提前丢弃记录，防止极端长尾任务后续合并到错误结果。

2.2 Java 单机实现

WARNING

Java 没有官方工具，手撸版常见漏洞有 4 个：泛型不安全、await 无超时、Error 不隔离、清理时使用单参 remove。下面是修复后的版本：

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

public class SingleFlight {

private final Map<String, Call<?>> callMap = new ConcurrentHashMap<>();

private static class Call<T> {
final CountDownLatch latch = new CountDownLatch(1);
volatile T result;
volatile Throwable error;
}

@FunctionalInterface
public interface Task<T> {
T run() throws Exception;
}

@SuppressWarnings("unchecked")
public <T> T execute(String key, Task<T> task, long timeoutMs) throws Exception {
Call<T> myCall = new Call<>();
// putIfAbsent 返回旧值；为 null 代表本线程抢占成功
Call<T> existing = (Call<T>) callMap.putIfAbsent(key, myCall);

if (existing == null) {
try {
myCall.result = task.run();
} catch (Throwable t) {
// Error 也兜住，避免一个请求把所有等待方拖进未定义状态
myCall.error = t;
if (t instanceof Exception) throw (Exception) t;
throw new RuntimeException(t);
} finally {
myCall.latch.countDown();
// 双参 remove：只删自己写入的那个引用，避免误删
callMap.remove(key, myCall);
}
return myCall.result;
}

// 抢占失败：限时等待，避免无限阻塞
if (!existing.latch.await(timeoutMs, TimeUnit.MILLISECONDS)) {
throw new TimeoutException("singleflight wait timeout, key=" + key);
}
if (existing.error != null) {
// 把首发者的异常透传给所有等待方
Throwable e = existing.error;
if (e instanceof Exception) throw (Exception) e;
throw new RuntimeException(e);
}
return existing.result;
}
}

2.3 单机方案的局限

仅能合并单进程内的请求；集群下每台机器都会执行一次，下游压力仍随机器数线性增长。集群越大、热点越集中，单机方案越力不从心。

三、分布式 SingleFlight 设计

3.1 设计目标

跨机器、跨进程合并相同请求，让整个集群内同一 Key 全局仅执行一次。

3.2 通用流程

节点 1 ─┐ ┌── 抢占成功 → 执行业务 → 写结果缓存 → 广播唤醒
节点 2 ─┼─→ 抢全局锁 ┤
节点 3 ─┘ └── 抢占失败 → 订阅/等待 → 读取结果缓存

四个关键能力：

3.3 中间件选型

工业界几乎都用 Redis：性能好、原生支持锁 / 缓存 / Pub-Sub。ZooKeeper、etcd 更适合强一致场景但延迟更高、接入更重，本文不展开。

四、生产可用的三种分布式实现

以下三种方案都修复了原版常见的 "锁误删 / 异常不广播 / 结果只能存 String" 等问题。

方案 1：Redis 锁 + 轮询 + Lua 安全释放

适合快速落地、依赖最少的场景。

关键安全设计：

import com.fasterxml.jackson.databind.ObjectMapper;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.Collections;
import java.util.UUID;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;

@Component
public class DistributedSingleFlight {

private static final String LOCK_PREFIX = "sf:lock:";
private static final String RESULT_PREFIX = "sf:res:";
private static final String ERROR_PREFIX = "sf:err:";

private static final long LOCK_EXPIRE_MS = 10_000;
private static final long RESULT_TTL_MS = 5_000;
private static final long ERROR_TTL_MS = 1_000; // 异常缓存短，避免长时间影响
private static final long MAX_WAIT_MS = 3_000;
private static final long POLL_BASE_MS = 50;

// Lua：value 相等才删，原子操作
private static final DefaultRedisScript<Long> UNLOCK_SCRIPT = new DefaultRedisScript<>(
"if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end",
Long.class);

@Resource private StringRedisTemplate redis;
private final ObjectMapper mapper = new ObjectMapper();

public <T> T execute(String bizKey, Class<T> type, Task<T> task) throws Exception {
String lockKey = LOCK_PREFIX + bizKey;
String resKey = RESULT_PREFIX + bizKey;
String errKey = ERROR_PREFIX + bizKey;
String token = UUID.randomUUID().toString();
long deadline = System.currentTimeMillis() + MAX_WAIT_MS;

// 1. 先查结果缓存
T cached = readResult(resKey, type);
if (cached != null) return cached;

// 2. 抢锁
Boolean locked = redis.opsForValue()
.setIfAbsent(lockKey, token, LOCK_EXPIRE_MS, TimeUnit.MILLISECONDS);

if (Boolean.TRUE.equals(locked)) {
try {
T result = task.run();
redis.opsForValue().set(resKey,
mapper.writeValueAsString(result),
RESULT_TTL_MS, TimeUnit.MILLISECONDS);
return result;
} catch (Exception e) {
// 异常也广播：让等待方快速失败而非全员穿透下游
redis.opsForValue().set(errKey, e.getClass().getName() + ":" + e.getMessage(),
ERROR_TTL_MS, TimeUnit.MILLISECONDS);
throw e;
} finally {
// 安全释放：value 一致才删
redis.execute(UNLOCK_SCRIPT, Collections.singletonList(lockKey), token);
}
}

// 3. 未抢到锁：带抖动的轮询
while (System.currentTimeMillis() < deadline) {
long jitter = ThreadLocalRandom.current().nextLong(POLL_BASE_MS);
Thread.sleep(POLL_BASE_MS + jitter);

T result = readResult(resKey, type);
if (result != null) return result;

String err = redis.opsForValue().get(errKey);
if (err != null) throw new RuntimeException("upstream singleflight failed: " + err);
}
// 4. 超时降级：自己执行（宁可少量穿透，也不能业务整体不可用）
return task.run();
}

private <T> T readResult(String resKey, Class<T> type) throws Exception {
String raw = redis.opsForValue().get(resKey);
return raw == null ? null : mapper.readValue(raw, type);
}

@FunctionalInterface
public interface Task<T> {
T run() throws Exception;
}
}

优缺点：

• ✅ 实现简单、零额外依赖
• ✅ 锁安全、异常可广播、支持任意对象
• ⚠️ 轮询有少量 CPU/Redis 开销，但抖动后可接受

方案 2：Redis 锁 + Pub/Sub（无轮询低延迟）

针对方案 1 的轮询开销，引入 Pub/Sub 让等待方被动唤醒。

关键安全设计：先订阅，再 double-check 结果缓存，最后才进入等待，避免唤醒消息丢失。

// 关键流程（伪代码，省略与方案 1 重复的释放/序列化逻辑）

public <T> T execute(String bizKey, Class<T> type, Task<T> task) throws Exception {
String resKey = RESULT_PREFIX + bizKey;
String chKey = "sf:ch:" + bizKey;
String token = UUID.randomUUID().toString();

// 先查缓存
T cached = readResult(resKey, type);
if (cached != null) return cached;

// 抢锁
if (tryLock(bizKey, token)) {
try {
T result = task.run();
writeResult(resKey, result);
return result;
} finally {
// 先发布唤醒，再安全释放锁
redis.convertAndSend(chKey, "done");
safeUnlock(bizKey, token);
}
}

// === 安全订阅：必须先订阅再 double-check，否则消息可能丢失 ===
CountDownLatch latch = new CountDownLatch(1);
MessageListener listener = (msg, pattern) -> latch.countDown();
container.addMessageListener(listener, new ChannelTopic(chKey));
try {
// double-check：订阅之前可能消息已经发布完了
T result = readResult(resKey, type);
if (result != null) return result;

if (!latch.await(MAX_WAIT_MS, TimeUnit.MILLISECONDS)) {
return task.run(); // 降级
}
result = readResult(resKey, type);
return result != null ? result : task.run();
} finally {
container.removeMessageListener(listener);
}
}

优缺点：

• ✅ 无轮询开销，延迟最低
• ⚠️ Pub/Sub 不持久化，断线期间消息会丢；必须保留超时降级兜底
• ⚠️ 实现复杂度高于方案 1

方案 3：Redisson 分布式锁 + 看门狗（生产首选）

Redisson 的 RLock 自带 owner 校验 + watchdog 自动续期，是生产环境最稳的选择。

关键安全设计：

import org.redisson.api.RTopic;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

@Component
public class RedissonSingleFlight {

private static final String LOCK_PREFIX = "sf:lock:";
private static final String RES_PREFIX = "sf:res:";
private static final String TOPIC_PREFIX = "sf:ch:";
private static final long MAX_WAIT_MS = 3_000;

@Resource private RedissonClient redisson;
private final ObjectMapper mapper = new ObjectMapper();

public <T> T execute(String bizKey, Class<T> type, Task<T> task) throws Exception {
String resKey = RES_PREFIX + bizKey;
String topic = TOPIC_PREFIX + bizKey;

// 1. 先查缓存
T cached = readResult(resKey, type);
if (cached != null) return cached;

RLock lock = redisson.getLock(LOCK_PREFIX + bizKey);
// waitTime=0：抢不到立即返回；leaseTime=-1：启用看门狗自动续期
boolean locked = lock.tryLock(0, -1, TimeUnit.MILLISECONDS);

if (locked) {
try {
T result = task.run();
writeResult(resKey, result);
return result;
} finally {
redisson.getTopic(topic).publish("done");
lock.unlock(); // Redisson 自带 owner 校验
}
}

// 2. 等待方：先订阅 → double-check → 阻塞等待
CountDownLatch latch = new CountDownLatch(1);
RTopic rTopic = redisson.getTopic(topic);
int listenerId = rTopic.addListener(String.class, (ch, msg) -> latch.countDown());
try {
T result = readResult(resKey, type);
if (result != null) return result;

if (!latch.await(MAX_WAIT_MS, TimeUnit.MILLISECONDS)) {
return task.run();
}
result = readResult(resKey, type);
return result != null ? result : task.run();
} finally {
rTopic.removeListener(listenerId);
}
}

private <T> T readResult(String resKey, Class<T> type) { /* JSON 读 */ return null; }
private <T> void writeResult(String resKey, T v) { /* JSON 写 */ }

@FunctionalInterface public interface Task<T> { T run() throws Exception; }
}

为什么不用 RCountDownLatch？

常见教学版用 latch.trySetCount(1) 配合 countDown / await 实现，但这里有个致命 race：

• 节点 A 执行完 countDown 把 count 归 0；
• 节点 C 进来 trySetCount(1) 把 count 又设回 1；
• 节点 B 此时仍在 await，会拿到节点 C 的执行结果（语义已偏离）。

结果就是看似 "单飞"，实际每一波都重新跑一次，根本没合并跨批次的并发。所以生产更推荐用 RTopic（Pub/Sub）做唤醒。

五、方案对比与选型

选型口诀：

• 已用 Redisson → 直接方案 3
• 新项目快速验证 → 方案 1
• 追求极致延迟且能容忍降级 → 方案 2

六、生产环境必踩的 6 个坑

6.1 分布式锁误删

坑：finally 里 del lockKey，业务超时后 Redis 自动过期，新节点拿到锁，旧节点把别人的锁删了 → 并发执行。

正解：锁 value 写 UUID，用 Lua get + del 原子判断。

6.2 锁过期但业务未完

坑：固定 leaseTime，业务跑得比锁久 → 锁失效 → 并发执行。

正解：用 Redisson 看门狗（leaseTime=-1）或自己起协程续期。

6.3 异常黑洞

坑：执行节点报错不广播，等待方一直轮询到超时，最后全员降级穿透下游。

正解：异常也短期写入 "错误缓存"，等待方读到后快速失败。

6.4 结果只能存 String

坑：String.valueOf(obj) 写入、强转读出，复杂对象成乱码。

正解：JSON / Protobuf 序列化，传 Class<T> 或 TypeReference 反序列化。

6.5 Pub/Sub 消息丢失

坑：先抢锁失败、再订阅；此时发布者已经 publish 完了 → 永远等不到。

正解：先订阅再 double-check 缓存；超时降级永远要保留。

6.6 Redis 不可用

坑：Redis 抖动时所有请求挂在 Redis 调用上。

正解：Redis 调用包熔断器（Sentinel / Resilience4j），熔断后直接降级到本地执行。

七、最佳实践清单

•  Key 规范：业务域:资源:唯一ID，例如 goods:detail:1001，长 Key 做 SHA-1
•  多级缓存 + 单飞：本地缓存承接绝大部分流量，缓存失效用单飞防击穿
•  超时分层：业务超时 < 锁过期 < 等待超时
•  监控指标：合并率、执行次数、超时次数、降级次数、异常率
•  压测：必须模拟 Redis 抖动、网络分区、执行节点 OOM 三种故障

八、总结

SingleFlight 是一种轻量但高效的并发优化手段，核心思想始终是合并重复请求、复用执行结果。

• 单机靠进程内同步工具即可，注意泛型安全和限时等待
• 分布式靠 Redis 协调，关键看锁安全、续期、异常广播、序列化这四件事
• 生产环境强烈推荐 Redisson + 看门狗 + Pub/Sub 唤醒 的组合

在真实工程里，SingleFlight 是缓存体系的补充而非独立方案。合理搭配「多级缓存 + 请求合并 + 限流降级」，才能构建稳定的高并发数据链路。