自定义线程池的使用

现在有一个使用多线程进行调优的案例：模拟使用多线程并发处理数据。

1.自定义一个线程池

//自定义一个线程池
class ThreadPoolManger {
    private final ThreadPoolExecutor executor;
    private final AtomicBoolean isShutdown;

    /**
     * 构造方法，用于初始化线程池。
     *
     * @param corePoolSize    核心线程池的大小
     * @param maximumPoolSize 最大线程池的大小
     * @param keepAliveTime   非核心线程空闲存活时间
     * @param unit            存活时间单位
     * @param queueCapacity   任务队列容量
     */
    public ThreadPoolManger(int corePoolSize, int maximumPoolSize, long keepAliveTime,
                            TimeUnit unit, int queueCapacity) {
        //防止参数为负数
        if (corePoolSize <= 0 || maximumPoolSize <= 0 || queueCapacity < 0) {
            throw new IllegalArgumentException("Pool size and capacity must be positive.");
        }

        executor = new ThreadPoolExecutor(
                corePoolSize,
                maximumPoolSize,
                keepAliveTime,
                unit,
                new LinkedBlockingQueue<>(queueCapacity)
        );
        //初始线程池关闭状态为false
        this.isShutdown = new AtomicBoolean(false);
        configThreadPool();
    }

    /**
     * 配置线程池的线程工厂和拒绝策略。
     */
    private void configThreadPool() {
        //设置自定义的线程工厂
        executor.setThreadFactory(new ThreadFactory() {
            @Override
            //这个参数r，是该线程要执行的任务
            public Thread newThread(Runnable r) {
                Thread thread = new Thread(r);
                thread.setName("CustomThreadPool" + thread.getId());
                thread.setDaemon(false);//是否设置为守护线程
                thread.setPriority(Thread.NORM_PRIORITY);
                return thread;
            }
        });
        //自定义拒绝策略
        executor.setRejectedExecutionHandler(new RejectedExecutionHandler() {
            @Override
            /**
             * Runnable r：被拒绝的任务。
             * ThreadPoolExecutor executor：调用执行任务的线程池。
             */
            public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
                System.err.println("任务 " + r.toString() + " 被拒绝, 线程池 " + executor.toString() + " 已满");
            }
        });
    }

    //把任务提交给线程池
    public void submit(Runnable task) {
        if (isShutdown.get()) {
            throw new IllegalStateException("ThreadPool is shut down");
        }
        executor.execute(task); // 使用execute方法提交任务
    }

    //关闭线程池
    public void shutdown() {
        //CAS操作确保线程池只被关闭一次
        //如果false，说明有其他线程竞争关闭了
        if (isShutdown.compareAndSet(false, true)) {
            executor.shutdown();
            try {
                // 等待线程池关闭，这段时间留给其他线程执行完任务，最多等待30秒
                if (!executor.awaitTermination(30, TimeUnit.SECONDS)) {
                    // 如果超时，则尝试立即关闭
                    executor.shutdownNow();
                }
            } catch (InterruptedException e) {
                // 如果等待时被中断，重新设置线程中断标志，并尝试立即关闭线程池
                Thread.currentThread().interrupt();
                executor.shutdownNow();
            }
        }
    }

}

自定义线程池包含7个参数，这里我们全部自定义。

2.使用线程池并发处理数据

然后在main方法中，使用线程池中的线程，达到目的效果。

class ThreadDemo {

    //模拟项目中多线程的使用案例：并发处理数据
    public static void main(String[] args) {

        //假设这个集合就是我们要处理的数据集合，假设有10000条数据
        // 假设这是我们要处理的10000条数据
        List<Integer> data = new ArrayList<>();
        for (int i = 1; i <= 10000; i++) {
            data.add(i);
        }

        ThreadPoolManger threadPool = new ThreadPoolManger(
                5,10,20,TimeUnit.SECONDS,10
        );

        // 计算每个线程处理的数据量
        int totalData = data.size();
        int chunkSize = totalData / 10; // 每个线程处理1000条数据

        // 提交任务
        for (int i = 0; i < 10; i++) {
            int start = i * chunkSize;
            int end = start + chunkSize;

            // 确保最后一个线程处理所有剩余数据
            if (i == 9) {
                end = totalData; // 处理最后一部分数据
            }

            int finalEnd = end;
            threadPool.submit(new Runnable() {
                List<Integer> subList = data.subList(start, finalEnd);
                @Override
                public void run() {
                    for (Integer integer : subList) {
                        System.out.println("处理数据："+integer);

                    }
                }
            });
        }
        // 关闭线程池
        threadPool.shutdown();
    }
}

3.碰到的问题

3.1

注意，这里我说两个我当时觉得奇怪的点：

两个问题：第一个，为什么这个end报错，start不报错

第二个问题：为什么这样把end复制一份，也没有加final修饰，finalEnd就不报错了

gpt的回答：

第一个问题：为什么第一段代码中 end 报错，而 start 不报错？

在Java中，匿名内部类（如 Runnable）可以访问外部类的变量，但这些变量必须是final或者是effectively final（即在赋值后没有被修改）。在第一段代码中，end 是一个循环变量，在每次循环迭代中都会被重新赋值。因此，当匿名内部类尝试使用 end 时，编译器无法确认 end 是 final 或 effectively final，所以会报错。

而 start 由于是局部变量，且在每次迭代中重新赋值，所以它也不能被访问，实际上它也会报错。但这里你可能没有注意到它的作用域问题，两个变量都是存在类似的问题。

第二个问题：为什么第二段代码中这样把 end 复制一份，也没有加 final 修饰，finalEnd 就不报错了？

在第二段代码中，finalEnd 是在 end 的基础上创建的一个新的局部变量。Java 的编译器能够确定 finalEnd 的值在定义后不会被修改，因此它被视为 effectively final。

当你将 end 的值赋给 finalEnd 时，finalEnd 不再依赖于外部变量 end 的生命周期。由于 finalEnd 的值在定义后不再改变，所以它符合可以在匿名内部类中使用的条件。

总结

1. 第一个问题：end 在循环中会被重新赋值，导致编译器认为它不是 final 或 effectively final，因此报错。
2. 第二个问题：finalEnd 是新的局部变量，且在赋值后不再修改，所以它符合匿名内部类的使用条件。

---

注意：如果把这段代码删掉，end变量没有再次赋值，就不会报错。

            if (i == 9) {
                end = totalData; // 处理最后一部分数据
            }

3.2

问题2，volatile关键字修饰的变量

volatile static int count = 0;

Java 中，count++ 这个操作实际上是由多个步骤组成的：读取当前值、加一、再写回。这些步骤在多线程环境中是非原子的，意味着两个线程可能会同时读取到相同的值，然后都加一并写回，最终只会增加一次计数，导致丢失更新。

但是，每个线程对于自己要处理的subList集合是不会出现数据丢失。因为每个线程都是处理自己的一段subList，没有像这个count一样，出现多线程竞争同一个变量的情况。