concurrency-review

并发审查技能

审查 Java 并发代码的正确性、安全性和现代最佳实践。

为什么重要

近 60% 的多线程应用会因共享资源管理不当而遇到问题。- ACM 研究

并发错误：

• 难以复现 - 取决于时序
• 难以测试 - 可能只在负载下出现
• 难以调试 - 行为不确定

此技能有助于在问题进入生产环境之前发现它们。

何时使用

• 审查包含 synchronized、volatile、Lock 的代码
• 检查 @Async、CompletableFuture、ExecutorService
• 验证共享状态的线程安全性
• 审查虚拟线程 / 结构化并发代码
• 任何由多个线程访问的代码

---

现代 Java（21/25）：虚拟线程

何时使用虚拟线程

// ✅ Perfect for I/O-bound tasks (HTTP, DB, file I/O)
try (ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor()) {
    for (Request request : requests) {
        executor.submit(() -> callExternalApi(request));
    }
}

// ❌ Not beneficial for CPU-bound tasks
// Use platform threads / ForkJoinPool instead

经验法则：如果应用从未同时运行 10,000 个以上的任务，虚拟线程可能不会带来显著收益。

Java 25：修复同步固定问题

在 Java 21-23 中，虚拟线程在进入包含阻塞操作的 synchronized 代码块时会变为“固定”状态。Java 25 修复了这个问题（JEP 491）。

// In Java 21-23: ⚠️ Could cause pinning
synchronized (lock) {
    blockingIoCall();  // Virtual thread pinned to carrier
}

// In Java 25: ✅ No longer an issue
// But consider ReentrantLock for explicit control anyway

使用 ScopedValue 替代 ThreadLocal

// ❌ ThreadLocal problematic with virtual threads
private static final ThreadLocal<User> currentUser = new ThreadLocal<>();

// ✅ ScopedValue (Java 21+ preview, improved in 25)
private static final ScopedValue<User> CURRENT_USER = ScopedValue.newInstance();

ScopedValue.where(CURRENT_USER, user).run(() -> {
    // CURRENT_USER.get() available here and in child virtual threads
    processRequest();
});

结构化并发（Java 25 预览版）

// ✅ Structured concurrency - tasks tied to scope lifecycle
try (StructuredTaskScope.ShutdownOnFailure scope = new StructuredTaskScope.ShutdownOnFailure()) {
    Subtask<User> userTask = scope.fork(() -> fetchUser(id));
    Subtask<Orders> ordersTask = scope.fork(() -> fetchOrders(id));

    scope.join();            // Wait for all
    scope.throwIfFailed();   // Propagate exceptions

    return new Profile(userTask.get(), ordersTask.get());
}
// All subtasks automatically cancelled if scope exits

---

Spring @Async 常见问题

1. 忘记添加 @EnableAsync

// ❌ @Async silently ignored
@Service
public class EmailService {
    @Async
    public void sendEmail(String to) { }
}

// ✅ Enable async processing
@Configuration
@EnableAsync
public class AsyncConfig { }

2. 在同一个类中调用异步方法

@Service
public class OrderService {

    // ❌ Bypasses proxy - runs synchronously!
    public void processOrder(Order order) {
        sendConfirmation(order);  // Direct call, not async
    }

    @Async
    public void sendConfirmation(Order order) { }
}

// ✅ Inject self or use separate service
@Service
public class OrderService {
    @Autowired
    private EmailService emailService;  // Separate bean

    public void processOrder(Order order) {
        emailService.sendConfirmation(order);  // Proxy call, async works
    }
}

3. 在非公共方法上使用 @Async

// ❌ Non-public methods - proxy can't intercept
@Async
private void processInBackground() { }

@Async
protected void processInBackground() { }

// ✅ Must be public
@Async
public void processInBackground() { }

4. 默认执行器会为每个任务创建一个线程

// ❌ Default SimpleAsyncTaskExecutor - creates new thread each time!
// Can cause OutOfMemoryError under load

// ✅ Configure proper thread pool
@Configuration
@EnableAsync
public class AsyncConfig {

    @Bean
    public Executor taskExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(10);
        executor.setMaxPoolSize(50);
        executor.setQueueCapacity(100);
        executor.setThreadNamePrefix("async-");
        executor.setRejectedExecutionHandler(new CallerRunsPolicy());
        executor.initialize();
        return executor;
    }
}

5. SecurityContext 不会传播

// ❌ SecurityContextHolder is ThreadLocal-bound
@Async
public void auditAction() {
    // SecurityContextHolder.getContext() is NULL here!
    String user = SecurityContextHolder.getContext().getAuthentication().getName();
}

// ✅ Use DelegatingSecurityContextAsyncTaskExecutor
@Bean
public Executor taskExecutor() {
    ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
    // ... configure ...
    return new DelegatingSecurityContextAsyncTaskExecutor(executor);
}

---

CompletableFuture 模式

错误处理

// ❌ Exception silently swallowed
CompletableFuture.supplyAsync(() -> riskyOperation());
// If riskyOperation throws, nobody knows

// ✅ Always handle exceptions
CompletableFuture.supplyAsync(() -> riskyOperation())
    .exceptionally(ex -> {
        log.error("Operation failed", ex);
        return fallbackValue;
    });

// ✅ Or use handle() for both success and failure
CompletableFuture.supplyAsync(() -> riskyOperation())
    .handle((result, ex) -> {
        if (ex != null) {
            log.error("Failed", ex);
            return fallbackValue;
        }
        return result;
    });

超时处理（Java 9+）

// ✅ Fail after timeout
CompletableFuture.supplyAsync(() -> slowOperation())
    .orTimeout(5, TimeUnit.SECONDS);  // Throws TimeoutException

// ✅ Return default after timeout
CompletableFuture.supplyAsync(() -> slowOperation())
    .completeOnTimeout(defaultValue, 5, TimeUnit.SECONDS);

组合 Future

// ✅ Wait for all
CompletableFuture.allOf(future1, future2, future3)
    .thenRun(() -> log.info("All completed"));

// ✅ Wait for first
CompletableFuture.anyOf(future1, future2, future3)
    .thenAccept(result -> log.info("First result: {}", result));

// ✅ Combine results
future1.thenCombine(future2, (r1, r2) -> merge(r1, r2));

使用合适的执行器

// ❌ CPU-bound task in ForkJoinPool.commonPool (default)
CompletableFuture.supplyAsync(() -> cpuIntensiveWork());

// ✅ Custom executor for blocking/I/O operations
ExecutorService ioExecutor = Executors.newFixedThreadPool(20);
CompletableFuture.supplyAsync(() -> blockingIoCall(), ioExecutor);

// ✅ In Java 21+, virtual threads for I/O
ExecutorService virtualExecutor = Executors.newVirtualThreadPerTaskExecutor();
CompletableFuture.supplyAsync(() -> blockingIoCall(), virtualExecutor);

---

经典并发问题

竞态条件：先检查后执行

// ❌ Race condition
if (!map.containsKey(key)) {
    map.put(key, computeValue());  // Another thread may have added it
}

// ✅ Atomic operation
map.computeIfAbsent(key, k -> computeValue());

// ❌ Race condition with counter
if (count < MAX) {
    count++;  // Read-check-write is not atomic
}

// ✅ Atomic counter
AtomicInteger count = new AtomicInteger();
count.updateAndGet(c -> c < MAX ? c + 1 : c);

可见性：缺少 volatile

// ❌ Other threads may never see the update
private boolean running = true;

public void stop() {
    running = false;  // May not be visible to other threads
}

public void run() {
    while (running) { }  // May loop forever
}

// ✅ Volatile ensures visibility
private volatile boolean running = true;

long/double 非原子性

// ❌ 64-bit read/write is non-atomic on 32-bit JVMs
private long counter;

public void increment() {
    counter++;  // Not atomic!
}

// ✅ Use AtomicLong or synchronization
private AtomicLong counter = new AtomicLong();

// ✅ Or volatile (for single-writer scenarios)
private volatile long counter;

双重检查锁定

// ❌ Broken without volatile
private static Singleton instance;

public static Singleton getInstance() {
    if (instance == null) {
        synchronized (Singleton.class) {
            if (instance == null) {
                instance = new Singleton();  // May be seen partially constructed
            }
        }
    }
    return instance;
}

// ✅ Correct with volatile
private static volatile Singleton instance;

// ✅ Or use holder class idiom
private static class Holder {
    static final Singleton INSTANCE = new Singleton();
}

public static Singleton getInstance() {
    return Holder.INSTANCE;
}

死锁：锁定顺序

// ❌ Potential deadlock
// Thread 1: lock(A) -> lock(B)
// Thread 2: lock(B) -> lock(A)

public void transfer(Account from, Account to, int amount) {
    synchronized (from) {
        synchronized (to) {
            // Transfer logic
        }
    }
}

// ✅ Consistent lock ordering
public void transfer(Account from, Account to, int amount) {
    Account first = from.getId() < to.getId() ? from : to;
    Account second = from.getId() < to.getId() ? to : from;

    synchronized (first) {
        synchronized (second) {
            // Transfer logic
        }
    }
}

---

线程安全集合

选择合适的集合

使用场景	错误选择	正确选择
并发读写	HashMap	ConcurrentHashMap
频繁迭代	ConcurrentHashMap	CopyOnWriteArrayList
生产者-消费者	ArrayList	BlockingQueue
并发排序	TreeMap	ConcurrentSkipListMap

ConcurrentHashMap 常见问题

// ❌ Non-atomic compound operation
if (!map.containsKey(key)) {
    map.put(key, value);
}

// ✅ Atomic
map.putIfAbsent(key, value);
map.computeIfAbsent(key, k -> createValue());

// ❌ Nested compute can deadlock
map.compute(key1, (k, v) -> {
    return map.compute(key2, ...);  // Deadlock risk!
});

---

并发审查清单

🔴 高严重性（很可能是错误）

• 对共享状态执行先检查后操作时，必须进行同步
• synchronized 代码块中不得调用外部或未知代码（有死锁风险）
• 双重检查锁定中存在 volatile
• 循环等待更新时，不读取非 volatile 字段
• ConcurrentHashMap.compute() 不调用其他映射操作
• @Async 方法为公共方法，且从不同的 Bean 中调用

🟡 中等严重性（潜在问题）

• 线程池大小设置合理且已命名
• 已处理 CompletableFuture 异常（exceptionally/handle）
• 如有需要，已将 SecurityContext 传播到异步任务
• 已正确关闭 ExecutorService
• 在 finally 块中调用 Lock.unlock()
• 共享数据使用线程安全集合

🟢 现代模式（Java 21/25）

• 对 I/O 密集型并发任务使用虚拟线程
• 考虑使用 ScopedValue 替代 ThreadLocal
• 对相关子任务使用结构化并发
• 为 CompletableFuture 操作设置超时

📝 文档

• 在共享类上注明线程安全性
• 记录嵌套锁的加锁顺序
• 说明每处 volatile 用法的理由

---

分析命令

# Find synchronized blocks
grep -rn "synchronized" --include="*.java"

# Find @Async methods
grep -rn "@Async" --include="*.java"

# Find volatile fields
grep -rn "volatile" --include="*.java"

# Find thread pool creation
grep -rn "Executors\.\|ThreadPoolExecutor\|ExecutorService" --include="*.java"

# Find CompletableFuture without error handling
grep -rn "CompletableFuture\." --include="*.java" | grep -v "exceptionally\|handle\|whenComplete"

# Find ThreadLocal (consider ScopedValue in Java 21+)
grep -rn "ThreadLocal" --include="*.java"