Java并发

---
name: java-concurrency
description: "When to activate: Java concurrency, CompletableFuture, ExecutorService, virtual threads, synchronized, ReentrantLock, atomic, thread safety"
origin: zakr
metadata:
  namespace: zakr/devtools-pack
  version: 1.0.0
  category: engineering
  language: java-kotlin
  platforms: [claude-code, codex, gemini-cli, cursor, aider, windsurf]
---
# Java Concurrency Patterns

## Virtual Threads (Java 21)

```java
// 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

```java
// 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

```java
// 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

```java
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

```java
// 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

```java
// 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