Java Concurrency

Java Concurrency Patterns

Virtual Threads (Java 21)

// Per-request virtual thread — cheap, blocks are fine
try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
    executor.submit(() -> handleRequest(req));
}

// Spring Boot 3.2+ — enable globally
// spring.threads.virtual.enabled=true

// Structured concurrency (preview)
try (var scope = new StructuredTaskScope.ShutdownOnFailure()) {
    Future<User> user = scope.fork(() -> userApi.getUser(id));
    Future<List<Order>> orders = scope.fork(() -> orderApi.getOrders(id));
    scope.join().throwIfFailed();
    return new UserWithOrders(user.get(), orders.get());
}

CompletableFuture

// Chain async operations
CompletableFuture<UserDto> result = CompletableFuture
    .supplyAsync(() -> userRepo.findById(id), executor)
    .thenApply(User::toDto)
    .thenCompose(dto -> enrichWithOrders(dto))
    .exceptionally(ex -> {
        log.warn("Failed to load user {}", id, ex);
        return UserDto.empty();
    });

// Parallel fetch, combine results
var userFuture  = CompletableFuture.supplyAsync(() -> userApi.get(id), executor);
var orderFuture = CompletableFuture.supplyAsync(() -> orderApi.list(id), executor);

CompletableFuture.allOf(userFuture, orderFuture)
    .thenApply(v -> new Dashboard(userFuture.join(), orderFuture.join()))
    .get(5, TimeUnit.SECONDS);

// Timeout
CompletableFuture.supplyAsync(this::slowOp)
    .orTimeout(3, TimeUnit.SECONDS)
    .exceptionally(ex -> fallback());

ExecutorService

// Fixed pool for CPU-bound work
ExecutorService cpuPool = Executors.newFixedThreadPool(
    Runtime.getRuntime().availableProcessors());

// Bounded queue prevents OOM under load
ExecutorService bounded = new ThreadPoolExecutor(
    4, 16,
    60L, TimeUnit.SECONDS,
    new ArrayBlockingQueue<>(1000),
    new ThreadFactory() {
        private final AtomicInteger count = new AtomicInteger();
        public Thread newThread(Runnable r) {
            Thread t = new Thread(r, "worker-" + count.incrementAndGet());
            t.setDaemon(true);
            return t;
        }
    },
    new ThreadPoolExecutor.CallerRunsPolicy() // backpressure
);

// Always shut down
Runtime.getRuntime().addShutdownHook(new Thread(() -> {
    cpuPool.shutdown();
    try { cpuPool.awaitTermination(30, TimeUnit.SECONDS); }
    catch (InterruptedException e) { Thread.currentThread().interrupt(); }
}));

Locks & Conditions

public class BoundedBuffer<T> {
    private final Lock lock = new ReentrantLock();
    private final Condition notFull  = lock.newCondition();
    private final Condition notEmpty = lock.newCondition();
    private final Queue<T> queue;
    private final int capacity;

    public void put(T item) throws InterruptedException {
        lock.lock();
        try {
            while (queue.size() == capacity) notFull.await();
            queue.add(item);
            notEmpty.signal();
        } finally { lock.unlock(); }
    }

    public T take() throws InterruptedException {
        lock.lock();
        try {
            while (queue.isEmpty()) notEmpty.await();
            T item = queue.poll();
            notFull.signal();
            return item;
        } finally { lock.unlock(); }
    }
}

Atomic Operations

// AtomicLong for counters — no synchronized needed
private final AtomicLong requestCount = new AtomicLong();
requestCount.incrementAndGet();

// AtomicReference for lock-free state updates
private final AtomicReference<Config> configRef = new AtomicReference<>(Config.DEFAULT);
configRef.updateAndGet(old -> old.withTimeout(newTimeout));

// CAS pattern
AtomicInteger state = new AtomicInteger(0);
boolean acquired = state.compareAndSet(0, 1); // only one thread succeeds

ConcurrentCollections

// Thread-safe map — prefer over Collections.synchronizedMap
ConcurrentHashMap<String, User> cache = new ConcurrentHashMap<>();
cache.computeIfAbsent(key, k -> loadUser(k));   // atomic
cache.compute(key, (k, v) -> v == null ? new User() : v.withUpdatedAt(now())); // atomic update

// CopyOnWriteArrayList — fast reads, slow writes, iteration never throws CME
List<EventListener> listeners = new CopyOnWriteArrayList<>();

// BlockingQueue for producer-consumer
BlockingQueue<Task> queue = new LinkedBlockingQueue<>(1000);
queue.put(task);     // blocks if full
queue.poll(1, TimeUnit.SECONDS); // waits up to 1s

Key Rules

• Prefer virtual threads (Java 21) for I/O-bound work — they make blocking code cheap; no need for reactive frameworks for most use cases
• Never call Future.get() without a timeout — deadlocks are invisible without one
• CallerRunsPolicy is the simplest backpressure mechanism — caller slows down naturally
• ConcurrentHashMap.computeIfAbsent is atomic; get + putIfAbsent is not — always use the atomic compound operations
• Always release locks in finally — any exception before unlock() leaves the lock permanently held