package com.example.cache;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CompletionException;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Supplier;
public class SingleFlightLoader<K, V> {
private final ConcurrentHashMap<K, CompletableFuture<V>> inFlight = new ConcurrentHashMap<>();
public V load(K key, Supplier<V> loader) {
AtomicBoolean weAreLeader = new AtomicBoolean(false);
CompletableFuture<V> future = inFlight.computeIfAbsent(key, k -> {
weAreLeader.set(true);
return new CompletableFuture<>();
});
if (weAreLeader.get()) {
runLoad(key, future, loader);
}
try {
return future.join();
} catch (CompletionException e) {
Throwable cause = e.getCause();
if (cause instanceof RuntimeException) {
throw (RuntimeException) cause;
}
throw e;
}
}
private void runLoad(K key, CompletableFuture<V> future, Supplier<V> loader) {
try {
future.complete(loader.get());
} catch (Throwable t) {
future.completeExceptionally(t);
} finally {
// Only remove if this exact future is still published.
inFlight.remove(key, future);
}
}
}
package com.example.cache;
import com.google.common.cache.Cache;
import com.google.common.cache.CacheBuilder;
import java.util.concurrent.TimeUnit;
import org.springframework.stereotype.Service;
@Service
public class CachedProductService {
private final ProductRepository repository;
private final Cache<Long, Product> cache = CacheBuilder.newBuilder()
.maximumSize(10_000)
.expireAfterWrite(5, TimeUnit.MINUTES)
.build();
private final SingleFlightLoader<Long, Product> singleFlight = new SingleFlightLoader<>();
public CachedProductService(ProductRepository repository) {
this.repository = repository;
}
public Product getProduct(long id) {
Product cached = cache.getIfPresent(id);
if (cached != null) {
return cached;
}
return singleFlight.load(id, () -> {
Product fresh = repository.findById(id)
.orElseThrow(() -> new ProductNotFoundException(id));
cache.put(id, fresh);
return fresh;
});
}
public void invalidate(long id) {
cache.invalidate(id);
}
}
package com.example.cache;
import static org.junit.jupiter.api.Assertions.assertEquals;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import org.junit.jupiter.api.Test;
class SingleFlightLoaderTest {
@Test
void coalescesConcurrentLoadsForSameKey() throws Exception {
SingleFlightLoader<String, Integer> loader = new SingleFlightLoader<>();
AtomicInteger loadCount = new AtomicInteger();
int threads = 50;
ExecutorService pool = Executors.newFixedThreadPool(threads);
CountDownLatch start = new CountDownLatch(1);
AtomicInteger results = new AtomicInteger();
for (int i = 0; i < threads; i++) {
pool.submit(() -> {
await(start);
int value = loader.load("user:42", () -> {
loadCount.incrementAndGet();
sleep(100); // simulate slow I/O
return 7;
});
results.addAndGet(value);
});
}
start.countDown();
pool.shutdown();
pool.awaitTermination(5, TimeUnit.SECONDS);
assertEquals(1, loadCount.get(), "underlying load must run exactly once");
assertEquals(7 * threads, results.get());
}
private static void await(CountDownLatch latch) {
try {
latch.await();
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
private static void sleep(long ms) {
try {
Thread.sleep(ms);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
}
When a hot cache key expires under load, many threads can miss simultaneously and all run the same expensive load — hitting the database or an upstream service N times for one value. This is the thundering-herd (cache stampede) problem, and the single-flight pattern solves it by letting exactly one load run per key while every other concurrent caller waits for and shares that single result.
The SingleFlightLoader tab implements the guard. It keeps a ConcurrentHashMap<K, CompletableFuture<V>> of in-flight loads. The core trick is computeIfAbsent: for a given key it either finds an existing future or atomically installs a brand-new incomplete one. computeIfAbsent runs its mapping function under the map bin's lock, so only the winning thread receives true from the AtomicBoolean and becomes the loader; all losers get the same future reference and simply call future.join(). The loader then runs the supplier outside the map lock to avoid holding it during slow I/O, completing the shared future with either the value or the exception.
Cleanup matters: the finally block removes the entry with inFlight.remove(key, future), the two-argument form that only deletes if the value is still the exact future it published. That prevents a race where a newer load for the same key gets clobbered by a stale cleanup. Failures propagate to all waiters through completeExceptionally, so a broken load does not leave callers hung, and the entry is still removed so the next call retries fresh rather than caching the error.
The CachedProductService tab shows the read-through wiring. It checks a Guava Cache first, and only on a miss delegates to the loader, storing the resolved value back into the cache. Because the loader coalesces, a burst of misses for one product id triggers a single repository.findById call. The SingleFlightLoaderTest tab demonstrates the guarantee: 50 threads request the same key against a supplier that increments an AtomicInteger, and the assertion confirms the underlying load ran exactly once. This approach shines for expensive, idempotent reads with high key contention; the trade-off is that a slow load makes all waiters wait, so pairing it with a timeout on join or a stale-while-revalidate policy is wise for latency-sensitive paths.
Related snips
module Api
module V1
class UsersController < BaseController
def show
user = User.includes(:profile).find(params[:id])ETags for conditional requests and caching
use crossbeam::channel::unbounded;
use std::thread;
fn main() {
let (tx, rx) = unbounded();Crossbeam for advanced concurrent data structures
// Creating a Promise
const myPromise = new Promise((resolve, reject) => {
const success = true;
setTimeout(() => {
if (success) {Promises and async/await patterns for asynchronous JavaScript
use std::sync::mpsc;
use std::thread;
fn main() {
let (tx, rx) = mpsc::channel();Channels (mpsc) for message passing between threads
export type Settled<R> =
| { status: 'fulfilled'; value: R }
| { status: 'rejected'; reason: unknown };
export interface ConcurrencyOptions {
limit: number;Simple concurrency limiter for batch operations
json.array! @posts do |post|
json.cache! ['v1', post], expires_in: 1.hour do
json.id post.id
json.title post.title
json.excerpt post.excerpt
json.published_at post.published_atFragment caching for expensive JSON serialization
Share this code
Here's the card — post it anywhere.
