Java并发编程最佳实践之构建高并发、可伸缩的应用实例

// 好的做法
ThreadPoolExecutor executor = new ThreadPoolExecutor(
    corePoolSize,          // 核心线程数
    maximumPoolSize,       // 最大线程数
    keepAliveTime,         // 线程存活时间
    TimeUnit.SECONDS,      // 时间单位
    new LinkedBlockingQueue<>(queueCapacity), // 任务队列
    new ThreadFactoryBuilder().setNameFormat("worker-%d").build(), // 线程工厂
    new ThreadPoolExecutor.CallerRunsPolicy() // 拒绝策略
);

// 不好的做法
ExecutorService executor = Executors.newFixedThreadPool(10); // 可能导致 OOM

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

// 好的做法
public class Counter {
    private final AtomicInteger count = new AtomicInteger(0);
    public void increment() {
        count.incrementAndGet();
    }
    public int getCount() {
        return count.get();
    }
}
// 不好的做法
public class Counter {
    private int count = 0;
    public synchronized void increment() {
        count++;
    }
    public synchronized int getCount() {
        return count;
    }
}

// 高并发读写
private final Map<String, User> userMap = new ConcurrentHashMap<>();
// 读多写少
private final List<User> userList = new CopyOnWriteArrayList<>();
// 生产者-消费者
private final BlockingQueue<Message> messageQueue = new LinkedBlockingQueue<>();

// 最佳实践
public enum Singleton {
    INSTANCE;
    private final UserService userService;
    Singleton() {
        userService = new UserService();
    }
    public UserService getUserService() {
        return userService;
    }
}
// 使用
UserService userService = Singleton.INSTANCE.getUserService();

public class Singleton {
    private static volatile Singleton instance;
    private Singleton() {}
    public static Singleton getInstance() {
        if (instance == null) {
            synchronized (Singleton.class) {
                if (instance == null) {
                    instance = new Singleton();
                }
            }
        }
        return instance;
    }
}

// 原子更新
AtomicInteger counter = new AtomicInteger(0);
// 递增
int value = counter.incrementAndGet();
// 比较并交换
boolean updated = counter.compareAndSet(expectedValue, newValue);

// 等待多个线程完成
CountDownLatch latch = new CountDownLatch(3);
for (int i = 0; i < 3; i++) {
    executor.submit(() -> {
        try {
            // 执行任务
        } finally {
            latch.countDown();
        }
    });
}
// 等待所有任务完成
latch.await();

// 等待多个线程到达屏障
CyclicBarrier barrier = new CyclicBarrier(3, () -> {
    System.out.println("所有线程已到达屏障");
});
for (int i = 0; i < 3; i++) {
    executor.submit(() -> {
        try {
            // 执行任务
            barrier.await();
            // 继续执行
        } catch (Exception e) {
            e.printStackTrace();
        }
    });
}

// 控制并发访问数
Semaphore semaphore = new Semaphore(5);
for (int i = 0; i < 10; i++) {
    executor.submit(() -> {
        try {
            semaphore.acquire();
            try {
                // 执行任务
            } finally {
                semaphore.release();
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    });
}

// 异步执行
CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> {
    // 执行任务
    return "Result 1";
});
CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> {
    // 执行任务
    return "Result 2";
});
// 组合结果
CompletableFuture<String> combined = future1.thenCombine(future2, (result1, result2) -> {
    return result1 + ", " + result2;
});
// 获取结果
String result = combined.join();

// 好的做法：按固定顺序获取锁
public void transferMoney(Account from, Account to, int amount) {
    int fromHash = System.identityHashCode(from);
    int toHash = System.identityHashCode(to);
    if (fromHash < toHash) {
        synchronized (from) {
            synchronized (to) {
                from.debit(amount);
                to.credit(amount);
            }
        }
    } else if (fromHash > toHash) {
        synchronized (to) {
            synchronized (from) {
                from.debit(amount);
                to.credit(amount);
            }
        }
    } else {
        // 使用额外的锁
        synchronized (lock) {
            synchronized (from) {
                synchronized (to) {
                    from.debit(amount);
                    to.credit(amount);
                }
            }
        }
    }
}

public class InventoryService {
    private final ConcurrentHashMap<String, AtomicInteger> inventory = new ConcurrentHashMap<>();
    public boolean decreaseStock(String productId, int quantity) {
        AtomicInteger stock = inventory.computeIfAbsent(productId, k -> new AtomicInteger(0));
        while (true) {
            int currentStock = stock.get();
            if (currentStock < quantity) {
                return false; // 库存不足
            }
            if (stock.compareAndSet(currentStock, currentStock - quantity)) {
                return true; // 成功减少库存
            }
            // 竞争失败，重试
        }
    }
    public int getStock(String productId) {
        AtomicInteger stock = inventory.get(productId);
        return stock != null ? stock.get() : 0;
    }
    public void increaseStock(String productId, int quantity) {
        AtomicInteger stock = inventory.computeIfAbsent(productId, k -> new AtomicInteger(0));
        stock.addAndGet(quantity);
    }
}