java线程池

July 17, 2022

线程池工具类由 java.util.concurrent 包提供

线程池

线程池用于管理和分配线程

ExecutorService 接口

java.util.concurrent.Executors 类提供静态方法，返回不同类型的线程池，这些线程池用 ExecutorService 接口操作

https://docs.oracle.com/en/java/javase/17/docs/api/java.base/java/util/concurrent/ExecutorService.html
https://docs.oracle.com/en/java/javase/17/docs/api/java.base/java/util/concurrent/Executors.html

java

import java.util.concurrent.*;

public class Main {
    public static void main(String[] args) throws InterruptedException {
        ExecutorService es = Executors.newFixedThreadPool(4); // 创建一个固定大小的线程池

        for (int i = 1; i <= 6; i++) {
            int id = i;
            es.submit(() -> System.out.println("Running Task: " + id));
        }
        es.shutdown(); // 立即关闭线程池
//        es.awaitTermination(5, TimeUnit.SECONDS); //  等待5s后再关闭线程池
    }
}

上面的线程池管理固定的 4 个线程，所以 1 2 3 4 号线程同时运行，5 6 号线程总是在前面的线程执行完毕后运行
因此，输出结果应该是 1 2 3 4 之间任意排序，然后是 5 6 之间任意排序

java

// 该线程池会动态调整线程数
ExecutorService es = Executors.newCachedThreadPool();

java

// 指定动态调整的线程数范围
int min = 4;
int max = 10;
ExecutorService es = new ThreadPoolExecutor(min, max, 60L, TimeUnit.SECONDS, new SynchronousQueue<Runnable>());

ScheduledThreadPoolExecutor （ExecutorService 实现）

java

ScheduledExecutorService ses = Executors.newScheduledThreadPool(4);
// 1秒后执行一次性任务:
ses.schedule(new Task("one-time"), 1, TimeUnit.SECONDS);
// 2秒后开始执行定时任务，每3秒执行:
ses.scheduleAtFixedRate(new Task("fixed-rate"), 2, 3, TimeUnit.SECONDS);
// 2秒后开始执行定时任务，以3秒为间隔执行:
ses.scheduleWithFixedDelay(new Task("fixed-delay"), 2, 3, TimeUnit.SECONDS);

ForkJoinPool （ExecutorService 实现）

ForkJoin 线程池用于将大任务拆成多个小任务并行执行。

java

import java.util.*;
import java.util.concurrent.*;

public class Main {
    public static void main(String[] args) throws Exception {
        long[] array = new Random(233).longs(1, 100)
                .limit(2000).toArray();
        long sum = Arrays.stream(array).sum(); // 求和

        // 执行 ForkJoin 任务
        ForkJoinTask<Long> task = new SumTask(array, 0, array.length);
        Long result = ForkJoinPool.commonPool().invoke(task);
        System.out.println("expect: " + sum + " got: " + result);
        // expect: 100732 got: 100732
    }
}

// 递归任务实现
class SumTask extends RecursiveTask<Long> {
    static final int THRESHOLD = 500;
    long[] array;
    int start;
    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) {
            // 如果任务足够小，直接计算
            return Arrays.stream(array).skip(start).limit(end - start).sum(); // 数组指定范围求和
        } else {
            // 任务太大，一分为二
            int middle = (end + start) / 2;
            SumTask subtask1 = new SumTask(this.array, start, middle);
            SumTask subtask2 = new SumTask(this.array, middle, end);
            invokeAll(subtask1, subtask2);
            long subresult1 = subtask1.join();
            long subresult2 = subtask2.join();
            return subresult1 + subresult2;
        }
    }
}

ThreadPoolExecutor （ExecutorService 实现）

ThreadPoolExecutor 是通用的 ExecutorService 实现
能够实现相较之 Executors 静态方法提供的线程池 newFixedThreadPool()、newSingleThreadExecutor()、newCachedThreadPool() 更灵活的线程池

https://docs.oracle.com/en/java/javase/17/docs/api/java.base/java/util/concurrent/ThreadPoolExecutor.html

←Back to home