网站建设目的意义,网页制作的视频教程,专业建设网站的公司,媒体网络推广价格优惠文章目录 什么是SingleFlight#xff1f;优化缺点优化策略 什么是SingleFlight#xff1f;
SingleFlight是go语言中sync包中的一个东西。它用于确保在并发环境下某个操作#xff08;例如#xff0c;函数调用#xff09;即使被多个goroutine同时请求#xff0c;也只会被执… 文章目录 什么是SingleFlight优化缺点优化策略 什么是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 是一种用于减少重复工作的工具特别是在并发编程中处理类似缓存击穿这样的问题时。尽管它非常有用但也有一些潜在的缺点和限制
缺点
资源锁定如果用于一个长时间运行的操作SingleFlight 会阻止其他所有相关的请求直到这个操作完成。这可能导致长时间的等待特别是在操作非常耗时的情况下。错误传播如果共享的操作因为某些原因失败了这个错误会被传播给所有等待的请求。在某些情况下单独重试可能更合适。内存压力在高并发情况下如果许多不同的键被请求SingleFlight 结构可能占用大量内存。不适合高变动数据对于频繁变化的数据使用 SingleFlight 可能不太有效因为一旦数据被缓存就需要等待旧数据失效才能获取新数据。分布式锁缺乏对分布式场景下的解决方案
优化策略
设置超时为 SingleFlight 中的操作设置合理的超时时间可以防止一个慢操作阻塞其他请求过长时间。错误重试机制对于某些操作特别是网络请求等实现自动重试逻辑可能会有帮助而不是直接将一个失败共享给所有请求。限制并发数量可以对 SingleFlight 正在进行的操作数量设置上限以减少内存压力。数据版本控制对于频繁变化的数据可以结合数据版本控制确保即使在数据更新的时候也能获取到最新的数据。
对上面的缺点进行优化得到如下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 SingleFlightK, V {private final ConcurrentHashMapK, CompletableFutureV ongoingOperations new ConcurrentHashMap();/*** 版本号控制使得每次进行更新的时候一定都是对最新数据进行更新*/private final ConcurrentHashMapK, 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默认3000ms3s* return*/public CompletableFutureV doOrTimeout(K key, FunctionK, V function, long timeout, boolean useDistributedLock) {if (useDistributedLock !acquireDistributedLock(key)) {CompletableFutureV 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) {CompletableFutureV 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 CompletableFutureV doOrRetry(K key, FunctionK, V function, int retries, long timeout,long delayBetweenRetries, boolean useDistributedLock) {if (useDistributedLock !acquireDistributedLock(key)) {CompletableFutureV 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) {CompletableFutureV 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 CompletableFutureV doOrCompensate(K key, FunctionK, V function, FunctionK, V compensation,int retries, long timeout, long delayBetweenRetries,boolean useDistributedLock) {if (useDistributedLock !acquireDistributedLock(key)) {CompletableFutureV 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) {CompletableFutureV 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(CompletableFutureV future,K key, FunctionK, V function, FunctionK, 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) {}FunctionK, 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) {FunctionK, 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 CompletableFutureCompletableFutureV doChan(K key, FunctionK, V function) {return CompletableFuture.completedFuture(ongoingOperations.computeIfAbsent(key, k - {CompletableFutureV 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 RedisSingleFlightK, V extends SingleFlightK, V {// Redis 或其他分布式锁机制的实现Overrideprotected boolean acquireDistributedLock(K key) {return false;}Overrideprotected void releaseDistributedLock(K key) {}// 如果需要可以添加特定于 Redis 的其他方法或逻辑
}
