Java并发编程(六)——线程池

最新推荐文章于 2024-11-25 21:37:57 发布

程勇uestc

最新推荐文章于 2024-11-25 21:37:57 发布

阅读量619

点赞数

分类专栏： Java 文章标签：线程池任务调度线程池工程调度

本文链接：https://blog.csdn.net/qq_40438165/article/details/115959700

版权

Java 专栏收录该内容

51 篇文章

订阅专栏

一、自定义线程池

$\quad$ 线程池跟连接池一样，维护线程池可以减少线程创建和关闭的时间。首先实现一个自定义线程池。
在这里插入图片描述
首先是创建一个管理任务的阻塞队列，防止任务突然变多，线程忙不过来而造成任务丢失的情况。Thread pool维护一堆线程。

class BlockingQueue<T>{
    // 1. 任务队列(双向链表，先进先出)
    private Deque<T> queue = new ArrayDeque<>();
    // 2. 锁(多个线程获取一个任务的时候需要加锁)保护队列头元素
    private ReentrantLock lock = new ReentrantLock();
    // 3. 生产者条件变量
    private Condition fullWaitSet = lock.newCondition();
    // 4. 消费者条件变量
    private Condition emptyWaitSet = lock.newCondition();
    // 5. 容量
    private int capacity;
    // 带有超时放弃的阻塞获取
    public T poll(long timeout, TimeUnit unit){
        lock.lock();
        try {
            long nanos = unit.toNanos(timeout);  // 将timeout统一转换成纳秒
            while(queue.isEmpty()){  // 队列为空时等待
                try {
                    // awaitNanos返回剩余等待时间
                    if(nanos <= 0) return null;  // 超时直接返回null
                    nanos = emptyWaitSet.awaitNanos(nanos);  // 让此线程进入等待
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            T t = queue.removeFirst();  // 取出头部任务返回
            // 已经消费一个任务，此时队列容量肯定小于capacity，需要唤醒fullWaitSet
            fullWaitSet.signal();
            return t;
        }finally {
            lock.unlock();
        }
    }
    // 阻塞获取
    public T take(){
        lock.lock();
        try {
            while(queue.isEmpty()){  // 队列为空时等待
                try {
                    emptyWaitSet.await();  // 让此线程进入等待
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            T t = queue.removeFirst();  // 取出头部任务返回
            // 已经消费一个任务，此时队列容量肯定小于capacity，需要唤醒fullWaitSet
            fullWaitSet.signal();
            return t;
        }finally {
            lock.unlock();
        }
    }
    // 阻塞添加
    public void put(T element){
        lock.lock();
        try {
            while(queue.size() == capacity){  // 容量满进入阻塞
                try {
                    fullWaitSet.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            queue.addLast(element);  // 添加元素
            // 唤醒emptyWaitSet中线程，表示有新任务可以处理了
            emptyWaitSet.signal();
        }finally {
            lock.unlock();
        }
    }
    // 获取大小
    public int size(){
        lock.lock();
        try{
            return queue.size();
        }finally {
            lock.unlock();
        }
    }
}

class ThreadPoll{
    // 1. 任务队列
    private BlockingQueue<Runnable> taskQueue;
    // 2. 线程集合
    private HashSet<Worker> workers = new HashSet<>();
    // 3. 线程数
    private int coreSize;
    // 4. 获取任务的超时时间
    private long timeout;
    // 5. 时间单位
    private TimeUnit unit;
    // 执行任务
    public void execute(Runnable task){
        synchronized (workers) {
            if (workers.size() < coreSize) {  // 线程数目没有超过限制
                Worker worker = new Worker(task);
                log.debug("new worker={}, task={}", worker, task);
                workers.add(worker);
                worker.start();
            } else {
                taskQueue.put(task);
//                taskQueue.offer(task, 1000, TimeUnit.MILLISECONDS);
            }
        }
    }

    public ThreadPoll(int coreSize, long timeout, TimeUnit unit, int queueCapacity) {
        this.coreSize = coreSize;
        this.timeout = timeout;
        this.unit = unit;
        this.taskQueue = new BlockingQueue<>(queueCapacity);
    }

    class Worker extends Thread{
        private Runnable task;
        public Worker(Runnable target) {
            this.task = task;
        }

        @Override
        public void run() {
            // 执行任务
            // 1. 当task不为空，执行任务
            // 2. task为空，则从任务队列中获取任务
            while(task != null || (task = taskQueue.poll(timeout, unit)) != null){
                try {
                    log.debug("running task={}", task);
                    task.run();
                }catch (Exception e){
                    e.printStackTrace();
                }finally {
                    task = null;
                }
            }
            synchronized (workers){
                log.debug("worker={} remove", this);
                workers.remove(this);
            }
        }
    }
}

二、线程池`ThreadPoolExecutor`

$\quad$ ThreadPoolExecutor使用int的高三位表示线程池状态，低29位表示线程数量。
在这里插入图片描述
从数值上比较：011>010>001>000>111(最高位是符号位)。

构造方法

    public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue,
                              RejectedExecutionHandler handler)

corePoolSize：核心线程数目(正常情况下线程数目)
maximumPoolSIze：最大线程数目(开启救急线程时线程最大数目)
keepAliveTime：救急线程生存时间
unit：救急线程时间单位
workQueue：阻塞队列
threadFactory：线程工厂，可以给线程起名字
handler：拒绝策略
当线程数达到 corePoolSize 并没有线程空闲，这时再加入任务，新加的任务会被加入workQueue 队列排队，直到有空闲的线程。
如果队列选择了有界队列，那么任务超过了队列大小时，会创建 maximumPoolSize - corePoolSize 数目的线程来救急。
如果线程到达 maximumPoolSize 仍然有新任务这时会执行拒绝策略
当高峰过去后，超过corePoolSize 的救急线程如果一段时间没有任务做，需要结束节省资源，这个时间由keepAliveTime 和 unit 来控制。

创建固定大小的线程池`newFixedThreadPool`

public static ExecutorService newFixedThreadPool(int nThreads) {
    return new ThreadPoolExecutor(nThreads, nThreads,
                                  0L, TimeUnit.MILLISECONDS,
                                  new LinkedBlockingQueue<Runnable>());
}

核心线程数=最大线程数，因此没有救急线程，也就没有超时时间
阻塞队列是无界的，可以放任意数量的任务
适用于任务量已知，相对耗时的任务

ExecutorService pool = Executors.newFixedThreadPool(2);
pool.execute(()->{
    log.debug("1");
});
pool.execute(()->{
    log.debug("2");
});
pool.execute(()->{
    log.debug("3");
});

带缓冲功能的线程池`newCachedThreadPool`

public static ExecutorService newCachedThreadPool() {
    return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                  60L, TimeUnit.SECONDS,
                                  new SynchronousQueue<Runnable>());
}

核心线程数是 0，最大线程数是 Integer.MAX_VALUE，救急线程的空闲生存时间是 60s，意味着
- 全部都是救急线程（60s 后可以回收）
- 救急线程可以无限创建
队列采用了 SynchronousQueue 实现特点是，它没有容量，没有线程来取是放不进去的（也就是只有当前队列中任务被线程取走后才能放入新的任务，相当于容量为1）
适合任务数密集但每个任务耗时短的情况

单个线程的线程池`newSingleThreadExecutor`

public static ExecutorService newSingleThreadExecutor() {
    return new FinalizableDelegatedExecutorService
        (new ThreadPoolExecutor(1, 1,
                                0L, TimeUnit.MILLISECONDS,
                                new LinkedBlockingQueue<Runnable>()));
}

希望多个任务排队执行。线程数固定为 1，任务数多于 1 时，会放入无界队列排队。任务执行完毕，这唯一的线程也不会被释放。
与自己创建一个线程池的区别：保证线程池中始终有一个线程处理任务，即使该线程中途处理某个任务抛出异常，线程池就会新建一个线程处理之后的任务

ExecutorService pool = Executors.newSingleThreadExecutor();
pool.execute(()->{
    log.debug("1");
    int i = 1 / 0;
});
pool.execute(()->{
    log.debug("2");
});
pool.execute(()->{
    log.debug("3");
});

线程提交任务的方法

void execute(Ruunable command); 执行任务
<T> Future<T> submit(Callable<T> task); 提交任务 task，用返回值 Future 获得任务执行结果

ExecutorService pool = Executors.newFixedThreadPool(2);
Future<String> future = pool.submit(new Callable<String>() {
    @Override
    public String call() throws Exception {
        log.debug("running");
        Thread.sleep(1000);
        return "ok";
    }
});
log.debug("{}", future.get());
11:14:43.482 c.TestThreadPoolExecutors [pool-1-thread-1] - running
11:14:44.496 c.TestThreadPoolExecutors [main] - ok

<T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks); 提交 tasks 中所有任务

ExecutorService pool = Executors.newFixedThreadPool(2);
List<Future<String>> futures = pool.invokeAll(Arrays.asList(
        () -> {
            log.debug("running1");
            Thread.sleep(1000);
            return "1";
        },
        () -> {
            log.debug("running2");
            Thread.sleep(2000);
            return "2";
        },
        () -> {
            log.debug("running3");
            Thread.sleep(3000);
            return "3";
        }
));
for (Future<String> future : futures) {
    log.debug("{}", future.get());
}

关闭线程池

void shutdown(); 线程池状态变为SHUTDOWN，不会接受新任务，但是已提交的任务会执行完毕

ExecutorService pool = Executors.newFixedThreadPool(2);
Future<String> res1 = pool.submit(() -> {
    log.debug("running1");
    Thread.sleep(1000);
    log.debug("finish1");
    return "1";
});
Future<String> res2 = pool.submit(() -> {
    log.debug("running2");
    Thread.sleep(1000);
    log.debug("finish2");
    return "2";
});
Future<String> res3 = pool.submit(() -> {
    log.debug("running3");
    Thread.sleep(1000);
    log.debug("finish3");
    return "3";
});
pool.shutdown();

16:11:50.948 c.TestThreadPoolExecutors [pool-1-thread-1] - running1
16:11:50.971 c.TestThreadPoolExecutors [pool-1-thread-2] - running2
16:11:51.950 c.TestThreadPoolExecutors [pool-1-thread-1] - finish1
16:11:51.950 c.TestThreadPoolExecutors [pool-1-thread-1] - running3
16:11:51.977 c.TestThreadPoolExecutors [pool-1-thread-2] - finish2
16:11:52.960 c.TestThreadPoolExecutors [pool-1-thread-1] - finish3

List<Runnable> shotdownNow(); 线程池状态变为STOP，不会接受新任务，会将队列中的任务返回

pool.shutdownNow();  // 会取消正在执行的任务

16:13:36.012 c.TestThreadPoolExecutors [pool-1-thread-1] - running1
16:13:36.012 c.TestThreadPoolExecutors [pool-1-thread-2] - running2

三、工作线程

$\quad$ 让有限的工作线程来轮流异步处理无限多的任务。设置多个线程池，每个线程池完成的工作内容不一。不同的任务类型应该使用不同的线程池，不然容易发生饥饿，相互等待。

线程池大小为多少比较合适

CPU密集型运算：线程池大小=cpu核数+1。+1是为了保证当线程故障时额外的线程可以顶上去，保证cpu时钟不浪费
I/O密集型：线程数=核数*总时间(cpu计算时间+等待时间)/cpu计算时间，例如对于单核，对于某类任务，cpu计算时间占50%，I/O时间占50%，则可以开两个线程，让cpu别闲着。

四、任务调度线程池

$\quad$ 假设要完成定时任务，在『任务调度线程池』功能加入之前，可以使用 java.util.Timer 来实现定时功能，Timer 的优点在于简单易用，但由于所有任务都是由同一个线程来调度，因此所有任务都是串行执行的，同一时间只能有一个任务在执行，前一个任务的延迟或异常都将会影响到之后的任务。假设要1s后同时执行task1和task2，可以使用newScheduledThreadPool(int coreSize)来实现:

ScheduledExecutorService pool = Executors.newScheduledThreadPool(2);
pool.schedule(()->{
    log.debug("task1");
    sleep(2);
}, 1, TimeUnit.SECONDS);
pool.schedule(()->{
    log.debug("task2");
}, 1, TimeUnit.SECONDS);

两个任务1s后同时进行，且一个任务出现异常不会影响其他任务：

17:08:04.443 c.TestTimer [pool-1-thread-2] - task2
17:08:04.453 c.TestTimer [pool-1-thread-1] - task1

$\quad$ 假设要执行定时任务，延时1s后每隔1s就同时执行任务1和任务2：

ScheduledExecutorService pool = Executors.newScheduledThreadPool(2);
log.debug("start...");
pool.scheduleAtFixedRate(()->{
    log.debug("task1");
    sleep(2);
}, 1, 1, TimeUnit.SECONDS);
pool.scheduleAtFixedRate(()->{
    log.debug("task2");
}, 1, 1, TimeUnit.SECONDS);

17:13:43.666 c.TestTimer [main] - start...
17:13:44.670 c.TestTimer [pool-1-thread-2] - task2
17:13:44.670 c.TestTimer [pool-1-thread-1] - task1
17:13:45.680 c.TestTimer [pool-1-thread-1] - task1
17:13:45.680 c.TestTimer [pool-1-thread-2] - task2

任务：每周四18:00:00定时执行任务

public static void main(String[] args) throws ExecutionException, InterruptedException {
    // 获取当前时间
    LocalDateTime now = LocalDateTime.now();
    // 获取每周时间
    LocalDateTime time = now.withHour(18).withMinute(0).withSecond(0).withNano(0).with(DayOfWeek.MONDAY);
    if(now.compareTo(time) > 0){
        time = time.plusWeeks(1);
    }
    long delay = Duration.between(now, time).toMillis();

    ScheduledExecutorService pool = Executors.newScheduledThreadPool(1);
    pool.scheduleAtFixedRate(()->{
        log.debug("task1");
    }, delay, 1000 * 60 * 60 * 24 * 7, TimeUnit.MILLISECONDS);
}