装饰网站开发背景小程序运营推广公司

什么是SingleFlight？

SingleFlight是go语言中sync包中的一个东西。它用于确保在并发环境下某个操作（例如，函数调用）即使被多个goroutine同时请求，也只会被执行一次。这对于防止重复的、昂贵的操作（如数据库查询、HTTP请求等）被不必要地多次执行是非常有用的。
使用 sync.SingleFlight，可以确保对于同一个键的并发请求，在缓存失效的情况下，只有一个请求会去加载数据（例如从数据库中），而其他并发的请求会等待这个加载操作完成，并共享相同的结果。这样，即便缓存失效，也不会因为大量的并发请求而对数据库或后端服务产生压力。
具体来说，当缓存失效时，第一个到达的请求会触发数据加载的操作（如数据库查询），而其他同时到达的请求会等待这个操作的完成。一旦数据被加载，它会被返回给所有等待的请求，并重新被放入缓存中。这个过程中，sync.SingleFlight 保证了数据加载函数只被调用一次，避免了不必要的重复处理。

SingleFlight主要提供以下功能：
● Do(key string, fn func() (interface{}, error)): 这是SingleFlight最核心的方法。当多个goroutine同时调用Do方法时，只有一个会真正执行传入的fn函数，其它等待这个函数执行完成。执行完成后，返回的结果和错误将会被返回给所有调用Do方法的goroutine。这里的key是用来区分不同操作的唯一标识。
● DoChan(key string, fn func() (interface{}, error)): 与Do类似，但它返回一个channel，你可以从这个channel中读取执行结果。
● Forget(key string): 这个方法用于清除SingleFlight中缓存的结果，以便于同一个key对应的函数在未来可以再次被执行。
● DupSuppressed() int64: 返回被SingleFlight机制抑制的重复调用次数。
SingleFlight的一个常见用途是缓存层，避免在缓存失效时由于缓存击穿而导致大量请求直接落到数据库。
如下是在写go语言的时候的使用SingleFight解决缓存击穿的代码。

var g singleflight.Groupfunc getCachedData(key string) (data interface{}, err error) {// 使用Do方法确保对于同一个key的请求，函数只会被执行一次v, err, _ := g.Do(key, func() (interface{}, error) {// 这里是实际的获取数据的操作，比如从数据库读取return getDataFromDatabase(key)})return v, err
}func getDataFromDatabase(key string) (interface{}, error) {// 模拟数据库操作// ...return data, nil
}

优化

SingleFlight 是一种用于减少重复工作的工具，特别是在并发编程中处理类似缓存击穿这样的问题时。尽管它非常有用，但也有一些潜在的缺点和限制：

缺点

1. 资源锁定：如果用于一个长时间运行的操作，SingleFlight 会阻止其他所有相关的请求，直到这个操作完成。这可能导致长时间的等待，特别是在操作非常耗时的情况下。
2. 错误传播：如果共享的操作因为某些原因失败了，这个错误会被传播给所有等待的请求。在某些情况下，单独重试可能更合适。
3. 内存压力：在高并发情况下，如果许多不同的键被请求，SingleFlight 结构可能占用大量内存。
4. 不适合高变动数据：对于频繁变化的数据，使用 SingleFlight 可能不太有效，因为一旦数据被缓存，就需要等待旧数据失效才能获取新数据。
5. 分布式锁：缺乏对分布式场景下的解决方案

优化策略

1. 设置超时：为 SingleFlight 中的操作设置合理的超时时间，可以防止一个慢操作阻塞其他请求过长时间。
2. 错误重试机制：对于某些操作，特别是网络请求等，实现自动重试逻辑可能会有帮助，而不是直接将一个失败共享给所有请求。
3. 限制并发数量：可以对 SingleFlight 正在进行的操作数量设置上限，以减少内存压力。
4. 数据版本控制：对于频繁变化的数据，可以结合数据版本控制，确保即使在数据更新的时候也能获取到最新的数据。

对上面的缺点进行优化，得到如下Java代码。

package blossom.project.towelove.common.utils;import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.Function;/*** @author: ZhangBlossom* @date: 2024/1/9 22:06* @contact: QQ:4602197553* @contact: WX:qczjhczs0114* @blog: https://blog.csdn.net/Zhangsama1* @github: https://github.com/ZhangBlossom* @description: 对于go语言中解决缓存击穿工具SingleFlight的实现*/
public abstract class SingleFlight<K, V> {private final ConcurrentHashMap<K, CompletableFuture<V>> ongoingOperations = new ConcurrentHashMap<>();/*** 版本号控制，使得每次进行更新的时候，一定都是对最新数据进行更新*/private final ConcurrentHashMap<K, AtomicLong> versions = new ConcurrentHashMap<>();// 设置默认超时时间，例如3秒private final long defaultTimeout = 3 * 1000;protected abstract boolean acquireDistributedLock(K key);protected abstract void releaseDistributedLock(K key);/*** 更新指定键的版本号** @param key 键*/public void updateVersion(K key) {versions.compute(key, (k, version) -> {if (version == null) {return new AtomicLong(1);} else {version.incrementAndGet();return version;}});}/*** 获取指定键的当前版本号** @param key 键* @return 版本号*/private long getVersion(K key) {return versions.getOrDefault(key, new AtomicLong(0)).get();}/*** 确保对于同一个键，相关的操作只会被执行一次，并且其结果将被所有调用者共享.* 如果超时了没有complete，将会返回TimeoutException** @param key      唯一key* @param function 要执行的方法函数* @param timeout  超时时间，单位为ms，默认3000ms=3s* @return*/public CompletableFuture<V> doOrTimeout(K key, Function<K, V> function, long timeout, boolean useDistributedLock) {if (useDistributedLock && !acquireDistributedLock(key)) {CompletableFuture<V> future = new CompletableFuture<>();future.completeExceptionally(new IllegalStateException("Unable to acquire distributed lock"));return future;}try {long versionAtCallTime = getVersion(key);return ongoingOperations.compute(key, (k, existingFuture) -> {if (existingFuture == null || getVersion(k) != versionAtCallTime) {CompletableFuture<V> future = new CompletableFuture<>();CompletableFuture.runAsync(() -> {try {V result = function.apply(k);future.complete(result);} catch (Exception e) {future.completeExceptionally(e);} finally {ongoingOperations.remove(k);}});// 应用超时设置return future.orTimeout(timeout, TimeUnit.MILLISECONDS);}return existingFuture;});} finally {if (useDistributedLock) {releaseDistributedLock(key);}}}/*** 当前方法会异步执行任务，并保证只有一个key能执行function任务，其他任务进行等待* 同时，如果执行失败，那么允许设定重试次数。并且再次执行function方法。** @param key                 执行方法唯一key* @param function            要执行的任务* @param retries             重试次数* @param timeout             超时时间* @param delayBetweenRetries 重试前延迟时间* @return*/public CompletableFuture<V> doOrRetry(K key, Function<K, V> function, int retries, long timeout,long delayBetweenRetries, boolean useDistributedLock) {if (useDistributedLock && !acquireDistributedLock(key)) {CompletableFuture<V> future = new CompletableFuture<>();future.completeExceptionally(new IllegalStateException("Unable to acquire distributed lock"));return future;}try {long versionAtCallTime = getVersion(key);return ongoingOperations.compute(key, (k, existingFuture) -> {if (existingFuture == null || getVersion(k) != versionAtCallTime) {CompletableFuture<V> future = new CompletableFuture<>();executeWithRetriesOrCompensate(future, key, function, null, retries, timeout, delayBetweenRetries, versionAtCallTime);return future;}return existingFuture;});} finally {if (useDistributedLock) {releaseDistributedLock(key);}}}/*** 当前方法会异步执行任务，并保证只有一个key能执行function任务，其他任务进行等待* 同时，如果执行失败，那么允许设定重试次数。并且执行compensation补偿方法。** @param key                 执行方法唯一key* @param function            原有方法* @param compensation        补偿方法 在执行失败的时候执行* @param retries             重试次数* @param timeout             超时时间* @param delayBetweenRetries 重试前延迟时间* @return*/public CompletableFuture<V> doOrCompensate(K key, Function<K, V> function, Function<K, V> compensation,int retries, long timeout, long delayBetweenRetries,boolean useDistributedLock) {if (useDistributedLock && !acquireDistributedLock(key)) {CompletableFuture<V> future = new CompletableFuture<>();future.completeExceptionally(new IllegalStateException("Unable to acquire distributed lock"));return future;}try {long versionAtCallTime = getVersion(key);return ongoingOperations.compute(key, (k, existingFuture) -> {if (existingFuture == null || getVersion(k) != versionAtCallTime) {CompletableFuture<V> future = new CompletableFuture<>();executeWithRetriesOrCompensate(future, key, function, compensation, retries, timeout,delayBetweenRetries, versionAtCallTime);return future;}return existingFuture;});} finally {if (useDistributedLock) {releaseDistributedLock(key);}}}/*** @param future* @param key* @param function* @param compensation* @param retries* @param timeout* @param delayBetweenRetries* @param versionAtCallTime*/private void executeWithRetriesOrCompensate(CompletableFuture<V> future,K key, Function<K, V> function, Function<K, V> compensation,int retries, long timeout, long delayBetweenRetries,long versionAtCallTime) {CompletableFuture.runAsync(() -> {try {if (getVersion(key) != versionAtCallTime) {throw new IllegalStateException("Data version changed");}V result = function.apply(key);future.complete(result);} catch (Exception e) {if (retries > 0 && getVersion(key) == versionAtCallTime) {try {TimeUnit.MILLISECONDS.sleep(delayBetweenRetries);} catch (InterruptedException ignored) {}Function<K, V> nextFunction = (compensation != null) ? compensation : function;executeWithRetriesOrCompensate(future, key, nextFunction, compensation, retries - 1,timeout, delayBetweenRetries, versionAtCallTime);} else {future.completeExceptionally(e);}}}).orTimeout(timeout, TimeUnit.MILLISECONDS).exceptionally(ex -> {if (retries > 0 && ex instanceof TimeoutException && getVersion(key) == versionAtCallTime) {Function<K, V> nextFunction = (compensation != null) ? compensation : function;executeWithRetriesOrCompensate(future, key, nextFunction, compensation, retries - 1, timeout,delayBetweenRetries, versionAtCallTime);} else {future.completeExceptionally(ex);}return null;});}/*** 提供一个方式来异步地执行操作，并返回一个 CompletableFuture，* 该 CompletableFuture 可以让调用者在未来某个时刻获取操作的结果** @param key* @param function* @return*/public CompletableFuture<CompletableFuture<V>> doChan(K key, Function<K, V> function) {return CompletableFuture.completedFuture(ongoingOperations.computeIfAbsent(key, k -> {CompletableFuture<V> future = new CompletableFuture<>();CompletableFuture.runAsync(() -> {try {V result = function.apply(k);future.complete(result);} catch (Exception e) {future.completeExceptionally(e);} finally {ongoingOperations.remove(k);}});return future;}));}/*** 从 ongoingOperations 映射中移除了给定的键** @param key*/public void forget(K key) {ongoingOperations.remove(key);}}/*** 假设一个基于Redis的SingleFlight分布式锁实现* 从而使得SingleFlight支持分布式锁* @param <K>* @param <V>*/
class RedisSingleFlight<K, V> extends SingleFlight<K, V> {// Redis 或其他分布式锁机制的实现@Overrideprotected boolean acquireDistributedLock(K key) {return false;}@Overrideprotected void releaseDistributedLock(K key) {}// 如果需要，可以添加特定于 Redis 的其他方法或逻辑
}