Java线程池高效并发编程实战技巧

import java.util.concurrent.*;
import java.util.Random;
public class ThreadPoolDemo {
    public static void main(String[] args) {
        // 1. 创建线程池
        // 核心参数：核心线程数5，最大线程数10，空闲时间60秒，任务队列容量100
        ThreadPoolExecutor executor = new ThreadPoolExecutor(
            5,  // corePoolSize
            10, // maximumPoolSize
            60, // keepAliveTime
            TimeUnit.SECONDS,
            new LinkedBlockingQueue<>(100), // 任务队列
            Executors.defaultThreadFactory(),
            new ThreadPoolExecutor.AbortPolicy() // 拒绝策略
        );
        // 2. 提交任务
        for (int i = 1; i <= 20; i++) {
            int taskId = i;
            executor.execute(() -> {
                System.out.println("处理任务" + taskId + 
                    ", 线程: " + Thread.currentThread().getName());
                try {
                    Thread.sleep(1000); // 模拟业务处理
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            });
        }
        // 3. 优雅关闭
        executor.shutdown();
        try {
            if (!executor.awaitTermination(60, TimeUnit.SECONDS)) {
                executor.shutdownNow();
            }
        } catch (InterruptedException e) {
            executor.shutdownNow();
        }
    }
}

import java.util.concurrent.*;
import java.util.List;
import java.util.ArrayList;
public class OrderProcessor {
    // 使用单例模式创建线程池
    private static final ThreadPoolExecutor orderExecutor = new ThreadPoolExecutor(
        3, 8, 30, TimeUnit.SECONDS,
        new ArrayBlockingQueue<>(1000),
        new ThreadFactory() {
            private int count = 0;
            @Override
            public Thread newThread(Runnable r) {
                Thread thread = new Thread(r, "order-process-" + (++count));
                thread.setDaemon(false);
                return thread;
            }
        },
        new ThreadPoolExecutor.CallerRunsPolicy() // 拒绝策略：调用者线程执行
    );
    /**
     * 异步处理订单
     */
    public CompletableFuture<Void> processOrderAsync(Order order) {
        return CompletableFuture.runAsync(() -> {
            try {
                // 1. 验证订单
                validateOrder(order);
                // 2. 扣减库存
                reduceInventory(order);
                // 3. 生成发货单
                generateShipping(order);
                // 4. 发送通知
                sendNotification(order);
                System.out.println("订单处理完成: " + order.getId());
            } catch (Exception e) {
                // 记录异常，进行补偿
                handleOrderException(order, e);
            }
        }, orderExecutor);
    }
    /**
     * 批量处理订单
     */
    public CompletableFuture<Void> batchProcessOrders(List<Order> orders) {
        List<CompletableFuture<Void>> futures = new ArrayList<>();
        for (Order order : orders) {
            CompletableFuture<Void> future = processOrderAsync(order);
            futures.add(future);
        }
        // 等待所有任务完成
        return CompletableFuture.allOf(
            futures.toArray(new CompletableFuture[0])
        );
    }
    // 业务方法（模拟实现）
    private void validateOrder(Order order) {
        // 验证逻辑
    }
    private void reduceInventory(Order order) {
        // 扣减库存逻辑
    }
    private void generateShipping(Order order) {
        // 生成发货单逻辑
    }
    private void sendNotification(Order order) {
        // 发送通知
    }
    private void handleOrderException(Order order, Exception e) {
        // 异常处理
    }
    // 优雅关闭
    public void shutdown() {
        orderExecutor.shutdown();
    }
    // 订单类
    static class Order {
        private String id;
        // 其他字段
        public String getId() { return id; }
    }
}

import java.util.concurrent.*;
import java.util.List;
import java.io.File;
public class DataExportService {
    // 专门用于导出任务的线程池
    private static final ThreadPoolExecutor exportExecutor = new ThreadPoolExecutor(
        2, 4, 5, TimeUnit.MINUTES,
        new LinkedBlockingQueue<>(50),
        new ThreadFactory() {
            private int count = 0;
            @Override
            public Thread newThread(Runnable r) {
                Thread thread = new Thread(r, "export-thread-" + (++count));
                thread.setPriority(Thread.NORM_PRIORITY);
                return thread;
            }
        },
        new ThreadPoolExecutor.DiscardOldestPolicy() // 拒绝策略：丢弃最老任务
    );
    /**
     * 异步导出Excel
     */
    public CompletableFuture<File> exportExcelAsync(String exportId, 
                                                     List<?> dataList) {
        return CompletableFuture.supplyAsync(() -> {
            System.out.println("开始导出数据，任务ID: " + exportId);
            try {
                // 模拟大数据量处理
                File excelFile = generateExcelFile(dataList);
                // 模拟上传到云存储
                String url = uploadToCloudStorage(excelFile);
                // 记录导出日志
                saveExportLog(exportId, url, "SUCCESS");
                return excelFile;
            } catch (Exception e) {
                saveExportLog(exportId, null, "FAILED");
                throw new RuntimeException("导出失败", e);
            }
        }, exportExecutor);
    }
    /**
     * 带进度的数据导出
     */
    public CompletableFuture<File> exportWithProgress(String exportId, 
                                                      List<?> dataList,
                                                      ProgressCallback callback) {
        return CompletableFuture.supplyAsync(() -> {
            int total = dataList.size();
            int batchSize = 1000;
            int processed = 0;
            for (int i = 0; i < total; i += batchSize) {
                int end = Math.min(i + batchSize, total);
                List<?> batchData = dataList.subList(i, end);
                // 处理批次数据
                processBatchData(batchData);
                processed = end;
                float progress = (float) processed / total;
                // 回调更新进度
                if (callback != null) {
                    callback.onProgress(progress);
                }
                // 模拟处理时间
                try {
                    Thread.sleep(100);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }
            return generateExcelFile(dataList);
        }, exportExecutor);
    }
    // 业务方法（模拟实现）
    private File generateExcelFile(List<?> dataList) {
        // 生成Excel文件
        return new File("export.xlsx");
    }
    private String uploadToCloudStorage(File file) {
        // 上传到云存储
        return "https://oss.example.com/" + file.getName();
    }
    private void saveExportLog(String exportId, String url, String status) {
        // 保存日志
    }
    private void processBatchData(List<?> batchData) {
        // 处理批次数据
    }
    // 进度回调接口
    public interface ProgressCallback {
        void onProgress(float progress);
    }
    // 获取线程池状态
    public void printThreadPoolStatus() {
        System.out.println("核心线程数: " + exportExecutor.getCorePoolSize());
        System.out.println("活动线程数: " + exportExecutor.getActiveCount());
        System.out.println("任务队列大小: " + exportExecutor.getQueue().size());
        System.out.println("已完成任务数: " + exportExecutor.getCompletedTaskCount());
    }
}

// 模拟处理时间
try {
    Thread.sleep(100);
} catch (InterruptedException e) {
    // 就是这一句！重新设置中断状态
    Thread.currentThread().interrupt();
}

import java.util.concurrent.*;
import java.time.LocalDateTime;
public class ScheduledTaskService {
    private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(3);
    /**
     * 初始化定时任务
     */
    public void initScheduledTasks() {
        // 1. 每天凌晨执行数据清理
        scheduleDailyCleanup();
        // 2. 每5分钟执行一次数据同步
        schedulePeriodicSync();
        // 3. 延迟执行一次性任务
        scheduleOneTimeTask();
    }
    /**
     * 每天凌晨2点执行数据清理
     */
    private void scheduleDailyCleanup() {
        long initialDelay = calculateInitialDelay(2, 0); // 凌晨2点
        long period = 24 * 60 * 60; // 24小时
        scheduler.scheduleAtFixedRate(() -> {
            try {
                System.out.println("开始数据清理: " + LocalDateTime.now());
                cleanUpOldData();
                System.out.println("数据清理完成: " + LocalDateTime.now());
            } catch (Exception e) {
                System.err.println("数据清理失败: " + e.getMessage());
            }
        }, initialDelay, period, TimeUnit.SECONDS);
    }
    /**
     * 每5分钟执行数据同步
     */
    private void schedulePeriodicSync() {
        scheduler.scheduleWithFixedDelay(() -> {
            try {
                syncDataWithExternalSystem();
            } catch (Exception e) {
                // 记录异常，下次继续执行
                System.err.println("数据同步失败: " + e.getMessage());
            }
        }, 0, 5, TimeUnit.MINUTES);
    }
    /**
     * 延迟10秒执行一次性任务
     */
    private void scheduleOneTimeTask() {
        scheduler.schedule(() -> {
            System.out.println("执行一次性任务: " + LocalDateTime.now());
        }, 10, TimeUnit.SECONDS);
    }
    /**
     * 提交可取消的定时任务
     */
    public ScheduledFuture<?> submitCancellableTask(Runnable task, 
                                                    long initialDelay, 
                                                    long period, 
                                                    TimeUnit unit) {
        return scheduler.scheduleAtFixedRate(task, initialDelay, period, unit);
    }
    // 工具方法：计算到指定时间的延迟
    private long calculateInitialDelay(int targetHour, int targetMinute) {
        LocalDateTime now = LocalDateTime.now();
        LocalDateTime targetTime = now.withHour(targetHour)
                                     .withMinute(targetMinute)
                                     .withSecond(0);
        if (now.isAfter(targetTime)) {
            targetTime = targetTime.plusDays(1);
        }
        return java.time.Duration.between(now, targetTime).getSeconds();
    }
    // 业务方法
    private void cleanUpOldData() {
        // 清理过期数据
    }
    private void syncDataWithExternalSystem() {
        // 同步数据
    }
    public void shutdown() {
        scheduler.shutdown();
    }
}

@Component
@Slf4j
public class DataCleanupTask {
    // 从配置文件读取 Cron 表达式，例如：0 0 2 * * ? (每天凌晨2点)
    @Scheduled(cron = "${task.cleanup.cron}")
    public void dailyCleanup() {
        log.info("开始数据清理...");
        try {
            // 业务逻辑
        } catch (Exception e) {
            log.error("清理失败", e);
        }
    }
}

@Scheduled(cron = "0 0 2 * * ?")
@SchedulerLock(name = "dataCleanupTask", lockAtMostFor = "10m", lockAtLeastFor = "1m")
public void scheduledTask() {
    // 只有抢到锁的机器才会执行
}