【Java】缓存击穿解决方案

什么是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.Group

func 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 或其他分布式锁机制的实现

    @Override
    protected boolean acquireDistributedLock(K key) {
        return false;
    }

    @Override
    protected void releaseDistributedLock(K key) {

    }

    // 如果需要，可以添加特定于 Redis 的其他方法或逻辑
}