Java并發(fā)編程實戰(zhàn)之從線程池到CompletableFuture詳解
更新時間：2026年04月18日 14:30:45 作者：程序員鴨梨
文章總結(jié)了Java并發(fā)編程的關(guān)鍵點,包括線程池的最佳實踐和監(jiān)控、CompletableFuture的異步編程、并發(fā)集合與工具類的使用、原子類與CAS、Fork/Join框架等內(nèi)容,強調(diào)了理解happens-before規(guī)則和內(nèi)存模型的重要性
一、線程池最佳實踐
1.1 自定義線程池
@Configuration
public class ThreadPoolConfig {
@Bean("taskExecutor")
public Executor taskExecutor() {
ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
// 核心線程數(shù)
executor.setCorePoolSize(10);
// 最大線程數(shù)
executor.setMaxPoolSize(50);
// 隊列容量
executor.setQueueCapacity(200);
// 線程活躍時間
executor.setKeepAliveSeconds(60);
// 線程名前綴
executor.setThreadNamePrefix("task-");
// 拒絕策略
executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
// 優(yōu)雅關(guān)閉
executor.setWaitForTasksToCompleteOnShutdown(true);
executor.setAwaitTerminationSeconds(60);
executor.initialize();
return executor;
}
@Bean("ioExecutor")
public Executor ioExecutor() {
// IO 密集型任務(wù)使用更多線程
return Executors.newFixedThreadPool(
Runtime.getRuntime().availableProcessors() * 2,
new ThreadFactoryBuilder().setNameFormat("io-pool-%d").build()
);
}
@Bean("cpuExecutor")
public Executor cpuExecutor() {
// CPU 密集型任務(wù)使用核心數(shù)線程
return Executors.newFixedThreadPool(
Runtime.getRuntime().availableProcessors(),
new ThreadFactoryBuilder().setNameFormat("cpu-pool-%d").build()
);
}
}
1.2 線程池監(jiān)控
@Component
public class ThreadPoolMonitor {
@Autowired
private ThreadPoolTaskExecutor taskExecutor;
@Scheduled(fixedRate = 60000)
public void monitor() {
ThreadPoolExecutor executor = taskExecutor.getThreadPoolExecutor();
log.info("Thread Pool Status - " +
"Active: {}, " +
"Pool Size: {}, " +
"Core Pool Size: {}, " +
"Max Pool Size: {}, " +
"Task Count: {}, " +
"Completed Task Count: {}, " +
"Queue Size: {}",
executor.getActiveCount(),
executor.getPoolSize(),
executor.getCorePoolSize(),
executor.getMaximumPoolSize(),
executor.getTaskCount(),
executor.getCompletedTaskCount(),
executor.getQueue().size()
);
}
}
二、CompletableFuture 異步編程
2.1 基礎(chǔ)用法
@Service
public class AsyncOrderService {
@Autowired
private InventoryService inventoryService;
@Autowired
private PaymentService paymentService;
@Autowired
private NotificationService notificationService;
// 異步創(chuàng)建訂單
public CompletableFuture<Order> createOrderAsync(OrderRequest request) {
return CompletableFuture.supplyAsync(() -> validateRequest(request))
.thenApplyAsync(this::reserveInventory)
.thenApplyAsync(this::processPayment)
.thenApplyAsync(this::saveOrder)
.whenComplete((order, ex) -> {
if (ex != null) {
log.error("Order creation failed", ex);
// 補償操作
compensate(order);
} else {
notificationService.sendOrderConfirmation(order);
}
});
}
// 并行處理
public CompletableFuture<OrderDetails> getOrderDetailsAsync(String orderId) {
CompletableFuture<Order> orderFuture = CompletableFuture
.supplyAsync(() -> orderRepository.findById(orderId).orElseThrow());
CompletableFuture<List<OrderItem>> itemsFuture = CompletableFuture
.supplyAsync(() -> orderItemRepository.findByOrderId(orderId));
CompletableFuture<ShippingInfo> shippingFuture = CompletableFuture
.supplyAsync(() -> shippingService.getShippingInfo(orderId));
// 等待所有完成
return CompletableFuture.allOf(orderFuture, itemsFuture, shippingFuture)
.thenApply(v -> {
OrderDetails details = new OrderDetails();
details.setOrder(orderFuture.join());
details.setItems(itemsFuture.join());
details.setShipping(shippingFuture.join());
return details;
});
}
// 帶超時的異步操作
public CompletableFuture<Order> createOrderWithTimeout(OrderRequest request) {
return createOrderAsync(request)
.orTimeout(30, TimeUnit.SECONDS)
.exceptionally(ex -> {
log.error("Order creation timeout", ex);
throw new OrderTimeoutException("Order creation timed out");
});
}
}
2.2 異常處理
@Service
public class AsyncExceptionHandlingService {
public CompletableFuture<Result> processWithFallback(String input) {
return CompletableFuture
.supplyAsync(() -> riskyOperation(input))
.exceptionally(ex -> {
log.error("Primary operation failed", ex);
return fallbackOperation(input);
})
.thenApply(this::transformResult);
}
// 組合異常處理
public CompletableFuture<Result> processWithMultipleFallbacks(String input) {
return CompletableFuture
.supplyAsync(() -> primaryOperation(input))
.handle((result, ex) -> {
if (ex != null) {
log.warn("Primary failed, trying fallback 1", ex);
return fallbackOperation1(input);
}
return result;
})
.thenCompose(result -> {
if (result == null) {
return CompletableFuture.supplyAsync(() -> fallbackOperation2(input));
}
return CompletableFuture.completedFuture(result);
});
}
}
三、并發(fā)集合與工具類
3.1 ConcurrentHashMap 使用
@Service
public class CacheService {
private final ConcurrentHashMap<String, CacheEntry> cache = new ConcurrentHashMap<>();
public Object get(String key, Supplier<Object> loader) {
CacheEntry entry = cache.get(key);
if (entry != null && !entry.isExpired()) {
return entry.getValue();
}
// 使用 computeIfAbsent 保證原子性
return cache.computeIfAbsent(key, k -> {
Object value = loader.get();
return new CacheEntry(value, System.currentTimeMillis() + 60000);
}).getValue();
}
public void put(String key, Object value, long ttlMillis) {
cache.put(key, new CacheEntry(value, System.currentTimeMillis() + ttlMillis));
}
// 批量操作
public Map<String, Object> getAll(Set<String> keys) {
return cache.entrySet().stream()
.filter(e -> keys.contains(e.getKey()) && !e.getValue().isExpired())
.collect(Collectors.toMap(
Map.Entry::getKey,
e -> e.getValue().getValue()
));
}
// 定期清理
@Scheduled(fixedRate = 60000)
public void cleanup() {
cache.entrySet().removeIf(e -> e.getValue().isExpired());
}
}
3.2 其他并發(fā)工具
@Component
public class ConcurrentToolsDemo {
// CountDownLatch - 等待多個任務(wù)完成
public void processWithLatch(List<Task> tasks) throws InterruptedException {
CountDownLatch latch = new CountDownLatch(tasks.size());
for (Task task : tasks) {
CompletableFuture.runAsync(() -> {
try {
task.execute();
} finally {
latch.countDown();
}
});
}
latch.await(5, TimeUnit.MINUTES);
}
// CyclicBarrier - 多階段任務(wù)同步
public void multiPhaseProcessing(List<Worker> workers) throws Exception {
CyclicBarrier barrier = new CyclicBarrier(workers.size());
for (Worker worker : workers) {
new Thread(() -> {
try {
// 階段 1
worker.phase1();
barrier.await();
// 階段 2
worker.phase2();
barrier.await();
// 階段 3
worker.phase3();
} catch (Exception e) {
log.error("Worker failed", e);
}
}).start();
}
}
// Semaphore - 限流
public void limitedConcurrentOperation(List<Task> tasks, int maxConcurrent) {
Semaphore semaphore = new Semaphore(maxConcurrent);
for (Task task : tasks) {
CompletableFuture.runAsync(() -> {
try {
semaphore.acquire();
task.execute();
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
} finally {
semaphore.release();
}
});
}
}
// Exchanger - 線程間數(shù)據(jù)交換
public void exchangeData() {
Exchanger<Data> exchanger = new Exchanger<>();
// 生產(chǎn)者
new Thread(() -> {
try {
Data data = produceData();
Data processed = exchanger.exchange(data);
// 使用處理后的數(shù)據(jù)
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}).start();
// 消費者
new Thread(() -> {
try {
Data data = exchanger.exchange(null);
Data processed = processData(data);
exchanger.exchange(processed);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}).start();
}
}
四、原子類與 CAS
@Component
public class AtomicOperationsDemo {
// 原子計數(shù)器
private final AtomicLong counter = new AtomicLong(0);
// 原子累加器（高并發(fā)下性能更好）
private final LongAdder longAdder = new LongAdder();
// 原子引用
private final AtomicReference<Config> configRef = new AtomicReference<>(new Config());
public long incrementAndGet() {
return counter.incrementAndGet();
}
public void add(long value) {
longAdder.add(value);
}
public long getSum() {
return longAdder.sum();
}
// CAS 更新配置
public boolean updateConfig(Config newConfig) {
Config current;
do {
current = configRef.get();
if (!shouldUpdate(current, newConfig)) {
return false;
}
} while (!configRef.compareAndSet(current, newConfig));
return true;
}
// LongAccumulator - 自定義累加邏輯
private final LongAccumulator maxAccumulator = new LongAccumulator(Long::max, 0);
public void updateMax(long value) {
maxAccumulator.accumulate(value);
}
public long getMax() {
return maxAccumulator.get();
}
}
五、Fork/Join 框架
@Component
public class ForkJoinDemo {
// 遞歸任務(wù)示例
public long parallelSum(long[] array) {
ForkJoinPool pool = new ForkJoinPool();
try {
return pool.invoke(new SumTask(array, 0, array.length));
} finally {
pool.shutdown();
}
}
// 遞歸任務(wù)
private static class SumTask extends RecursiveTask<Long> {
private static final int THRESHOLD = 10000;
private final long[] array;
private final int start;
private final int end;
SumTask(long[] array, int start, int end) {
this.array = array;
this.start = start;
this.end = end;
}
@Override
protected Long compute() {
if (end - start <= THRESHOLD) {
// 直接計算
long sum = 0;
for (int i = start; i < end; i++) {
sum += array[i];
}
return sum;
}
// 拆分任務(wù)
int mid = (start + end) / 2;
SumTask left = new SumTask(array, start, mid);
SumTask right = new SumTask(array, mid, end);
left.fork();
long rightResult = right.compute();
long leftResult = left.join();
return leftResult + rightResult;
}
}
// 遞歸動作示例
public void parallelProcess(List<Task> tasks) {
ForkJoinPool.commonPool().invoke(new ProcessTask(tasks, 0, tasks.size()));
}
private static class ProcessTask extends RecursiveAction {
private static final int THRESHOLD = 100;
private final List<Task> tasks;
private final int start;
private final int end;
ProcessTask(List<Task> tasks, int start, int end) {
this.tasks = tasks;
this.start = start;
this.end = end;
}
@Override
protected void compute() {
if (end - start <= THRESHOLD) {
for (int i = start; i < end; i++) {
tasks.get(i).execute();
}
return;
}
int mid = (start + end) / 2;
ProcessTask left = new ProcessTask(tasks, start, mid);
ProcessTask right = new ProcessTask(tasks, mid, end);
invokeAll(left, right);
}
}
}
六、總結(jié)
Java 并發(fā)編程的關(guān)鍵點：
- 線程池：合理配置核心參數(shù)，做好監(jiān)控
- 異步編程：CompletableFuture 讓異步代碼更優(yōu)雅
- 并發(fā)集合：選擇合適的線程安全集合
- 原子操作：CAS 實現(xiàn)無鎖編程
- 任務(wù)分解：Fork/Join 處理大數(shù)據(jù)量任務(wù)
這其實可以更優(yōu)雅一點。并發(fā)編程的核心是理解 happens-before 規(guī)則和內(nèi)存模型。
參考資源：
以上為個人經(jīng)驗，希望能給大家一個參考，也希望大家多多支持腳本之家。
相關(guān)文章
這篇文章主要介紹了Java線程阻塞和喚醒的幾種方式詳解,線程阻塞是指當(dāng)一個線程無法繼續(xù)執(zhí)行時,它會進(jìn)入阻塞狀態(tài),直到某個條件滿足后才能繼續(xù)執(zhí)行,線程阻塞可以通過多種方式實現(xiàn),如等待鎖、等待IO操作、等待其他線程的完成等,需要的朋友可以參考下2023-10-10
SpringBoot中的五種對靜態(tài)資源的映射規(guī)則的實現(xiàn)
這篇文章主要介紹了SpringBoot中的五種對靜態(tài)資源的映射規(guī)則的實現(xiàn)，文中通過示例代碼介紹的非常詳細(xì)，對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值，需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧2019-12-12
這篇文章主要介紹了java 文件鎖的簡單實現(xiàn)的相關(guān)資料,需要的朋友可以參考下2017-07-07
Spring Boot 3.4.x中@Bean新增defaultCandidate=false屬性,用于在存在相同類型Bean時忽略非默認(rèn)候選者,確保僅實例化一個Bean,下面就來介紹一下如何使用2025-08-08
java開發(fā)AOP基礎(chǔ)JdkDynamicAopProxy
這篇文章主要為大家介紹了java開發(fā)AOP基礎(chǔ)JdkDynamicAopProxy源碼示例解析，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進(jìn)步，早日升職加薪2023-07-07
Java讀取Excel文件并寫入數(shù)據(jù)庫的示例代碼
這篇文章主要介紹了Java讀取Excel文件并寫入數(shù)據(jù)庫的示例代碼,本文通過實例代碼給大家介紹的非常詳細(xì),對大家的學(xué)習(xí)或工作具有一定的參考借鑒價值,需要的朋友參考下吧2025-06-06
SpringBoot+Spring Security無法實現(xiàn)跨域的解決方案
這篇文章主要介紹了SpringBoot+Spring Security無法實現(xiàn)跨域的解決方案，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教2021-07-07
如何從官網(wǎng)下載Hibernate jar包的方法示例
這篇文章主要介紹了如何從官網(wǎng)下載Hibernate jar包的方法示例，小編覺得挺不錯的，現(xiàn)在分享給大家，也給大家做個參考。一起跟隨小編過來看看吧2019-04-04
java處理轉(zhuǎn)義字符↑ → ↓ 保存后的展示還原操作
這篇文章主要介紹了java處理轉(zhuǎn)義字符↑ → ↓ 保存后的展示還原操作，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教2021-06-06
java的main方法中調(diào)用spring的service方式
這篇文章主要介紹了在java的main方法中調(diào)用spring的service方式，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教2021-12-12