JUC并发编程 共享模式之工具 线程池 JDK 提供的线程池工具类 -- Executors 类创建线程池（创建ThreadPoolExecutor对象）

Posted 2021-10-21 Z && Y

1. Executors 类创建线程池

ThreadPoolExecutor类构造方法:

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

• corePoolSize 核心线程数目 (最多保留的线程数)
• maximumPoolSize 最大线程数目
• keepAliveTime 生存时间 - 针对救急线程
• unit 时间单位 - 针对救急线程
• workQueue 阻塞队列
• threadFactory 线程工厂 - 可以为线程创建时起个好名字
• handler 拒绝策略

根据这个构造方法，JDK Executors 类中提供了众多工厂方法来创建各种用途的线程池

---

1.1 newFixedThreadPool（固定大小线程池）

使用场景:

适用于任务量已知，相对耗时的任务

构造器:

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

特点:

• 核心线程数 = 最大线程数（没有救急线程被创建），因此也无需超时时间
• 阻塞队列是无界的，可以放任意数量的任务

示例代码:

package com.tian;

import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

@Slf4j(topic = "TestThreadPoolExecutors")
public class TestThreadPoolExecutors {
    public static void main(String[] args) {
        ExecutorService pool = Executors.newFixedThreadPool(2);

        pool.execute(() -> {
            log.debug("线程01执行了.");
        });

        pool.execute(() -> {
            log.debug("线程02执行了.");
        });

        pool.execute(() -> {
            log.debug("线程03执行了.");
        });

        pool.execute(() -> {
            log.debug("线程04执行了.");
        });
    }
}

可以为线程起名字:

package com.tian;

import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.atomic.AtomicInteger;

@Slf4j(topic = "TestThreadPoolExecutors")
public class TestThreadPoolExecutors {
    public static void main(String[] args) {
        ExecutorService pool = Executors.newFixedThreadPool(2, new ThreadFactory() {
            // 自增的原子整数
            private AtomicInteger number = new AtomicInteger(1);

            @Override
            public Thread newThread(Runnable r) {
                return new Thread(r, "my_pool_thread: " + number.getAndIncrement());
            }
        });

        pool.execute(() -> {
            log.debug("线程01执行了.");
        });

        pool.execute(() -> {
            log.debug("线程02执行了.");
        });

        pool.execute(() -> {
            log.debug("线程03执行了.");
        });

        pool.execute(() -> {
            log.debug("线程04执行了.");
        });
    }
}

---

1.2 newCachedThreadPool（带缓冲功能线程池）

使用场景:

整个线程池表现为线程数会根据任务量不断增长，没有上限，当任务执行完毕，空闲 1分钟后释放线程。 适合任务数比较密集，但每个任务执行时间较短的情况

构造器:

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

特点:

• 核心线程数是 0， 最大线程数是 Integer.MAX_VALUE，救急线程的空闲生存时间是 60s，意味着:
  • 全部都是救急线程（60s 后可以回收）
  • 救急线程可以无限创建
• 队列采用了 SynchronousQueue 实现特点是，它没有容量，没有线程来取是放不进去的（一手交钱、一手交货）

示例代码:

package com.tian;

import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.SynchronousQueue;


@Slf4j(topic = "c.TestSynchronousQueue")
public class TestSynchronousQueue {
    public static void main(String[] args) throws InterruptedException {
        SynchronousQueue<Integer> integers = new SynchronousQueue<>();
        new Thread(() -> {
            try {
                log.debug("putting {} ", 1);
                integers.put(1);
                log.debug("{} putted...", 1);

                log.debug("putting...{} ", 2);
                integers.put(2);
                log.debug("{} putted...", 2);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }, "t1").start();

        Thread.sleep(1000);

        new Thread(() -> {
            try {
                log.debug("taking {}", 1);
                integers.take();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }, "t2").start();

        Thread.sleep(1000);

        new Thread(() -> {
            try {
                log.debug("taking {}", 2);
                integers.take();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }, "t3").start();
    }
}

运行结果:

---

1.3 newSingleThreadExecutor（单线程线程池）

使用场景:

希望多个任务排队执行。线程数固定为 1，任务数多于 1 时，会放入无界队列排队。任务执行完毕，这唯一的线程也不会被释放。

构造器:

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

和自己创建一个线程的区别:

• 自己创建一个单线程串行执行任务，如果任务执行失败而终止那么没有任何补救措施，而线程池还会新建一个线程，保证池的正常工作
• Executors.newSingleThreadExecutor() 线程个数始终为1，不能修改
  • FinalizableDelegatedExecutorService 应用的是装饰器模式，只对外暴露了 ExecutorService 接口，因此不能调用 ThreadPoolExecutor 中特有的方法
• Executors.newFixedThreadPool(1) 初始时为1，以后还可以修改
  • 对外暴露的是 ThreadPoolExecutor 对象，可以强转后调用 setCorePoolSize 等方法进行修改

测试代码:

package com.tian;

import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.atomic.AtomicInteger;

@Slf4j(topic = "c.TestExecutors")
public class TestExecutors {
    public static void main(String[] args) throws InterruptedException {
        test2();
    }

    public static void test2() {
        ExecutorService pool = Executors.newSingleThreadExecutor();
        pool.execute(() -> {
            log.debug("1");
            int i = 1 / 0;
        });

        pool.execute(() -> {
            log.debug("2");
        });

        pool.execute(() -> {
            log.debug("3");
        });
    }

    private static void test1() {
        ExecutorService pool = Executors.newFixedThreadPool(2, new ThreadFactory() {
            private AtomicInteger t = new AtomicInteger(1);

            @Override
            public Thread newThread(Runnable r) {
                return new Thread(r, "mypool_t" + t.getAndIncrement());
            }
        });

        pool.execute(() -> {
            log.debug("1");
        });

        pool.execute(() -> {
            log.debug("2");
        });

        pool.execute(() -> {
            log.debug("3");
        });
    }
}

运行结果:

---

---