Java并发——线程池

ゞ浴缸里的玫瑰 2024-03-30 12:48 110阅读 0赞

## 线程池 ##

### 一、线程池的作用 ###

线程的创建和销毁需要占用CPU资源，若频繁的进行创建和销毁会产生很大的开销，影响性能和系统稳定性。

**线程池的优点：**

*  线程池可以保存创建好的线程随用随取，降低资源消耗（重复利用线程池中的线程）
 *  提高响应速度（无需创建线程，任务到达后直接可以执行）。
 *  使用线程池可以对线程进行统一分配、监控和调优。（线程管理）

### 二、ThreadPoolExecutor 介绍 ###

实现原理，一个存放线程的set集合，每一个线程都是一个任务的执行者（worker），一个存放任务的阻塞队列，当任务提交后放入阻塞队列，线程（worker）不断从阻塞队列中获取任务然后执行，如果阻塞队列中没有任务，线程就阻塞等待。

![在这里插入图片描述][9198e17349c44d8197e0592190a71729.png]

#### 2.1参数 ####

`ThreadPoolExecutor`构造方法源码

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

*  `corePoolSize` 核心线程数，线程池维护线程的最小数量（如果线程池中线程数小于核心线程数，在新任务到来时，会创建一个新的线程执行任务，执行结束存放在线程池中。如果线程数等于核心线程数，任务放入阻塞队列中）
 *  `maximumPoolSize` 线程池中允许的最大线程数（如果阻塞队列已满，再提交任务会判断当前线程数是否大于最大线程数，如果不大于就创建新线程，否则执行拒绝策略）
 *  `keepAliveTime` 线程池中除**核心线程**外其他线程的最长空闲时间，超过此时间线程就会被销毁
 *  `unit` 最长空闲时间`keepAliveTime` 的时间单位
 *  `workQueue` 保存等待被执行的任务的阻塞队列
 *  `threadFactory` 线程工厂，可以给创建的线程指定属性，比如：线程名。一般使用默认
 *  `handler` 线程池拒绝任务的策略，当阻塞队列已满，且没有空闲的线程，此时提交任务会触发任务拒绝策略

**线程池任务提交流程**

![在这里插入图片描述][39fccc80e3b249d7ae2728b662326dcc.png]

#### 2.2阻塞队列 ####

在JDK中提供了四种阻塞队列

1.  `ArrayBlockingQueue` 由数组实现的固定大小的有界阻塞队列，按FIFO处理任务
2.  `LinkedBlockingQueue` 由单链表实现的默认无界的阻塞队列（可以指定大小），默认大小`Integer.MAX_VALUE` ，按FIFO处理任务
3.  `SynchronousQueue` 不存储元素的阻塞队列，在添加任务时必须等另一个线程调用移除操作，否则会一直处于阻塞状态。
4.  `PriorityBlockingQueue` 具有优先级的阻塞队列，按优先级处理任务

`LinkedBlockingQueue`比`ArrayBlockingQueue`在添加和删除节点上性能方面更优，但是二者在`put()`, `take()`任务时均需要加锁，`SynchronousQueue`使用无锁算法，根据节点的状态判断执行，不需要用到锁，其核心是`Transfer.transfer()`

#### 2.3拒绝策略 ####

线程池提供四种拒绝策略

1.  `AbortPolicy` 不处理任务，直接抛异常（默认策略）
2.  `CallerRunsPolicy` 调用者所在的线程来执行任务
3.  `DiscardPolicy` 丢弃该任务
4.  `DiscardOldestPolicy` 丢弃阻塞队列中靠前的任务，执行当前任务

#### 2.4三种类型 ####

通过`Executors`类创建的线程池

**1.newFixedThreadPool**

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

核心线程数 = 最大线程数 且为固定传入的参数 `nThreads` 大小，固定线程数不会释放，不需要空闲时间。

使用 `LinkedBlockingQueue` 无界阻塞队列，不会拒绝任务，不使用拒绝策略

**2.newSingleThreadExecutor**

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

创建的线程池中只有一个线程，保证所提交的任务顺序执行

使用 `LinkedBlockingQueue` 无界阻塞队列，不会拒绝任务，不使用拒绝策略

**3.newCachedThreadPool**

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

核心线程数为0，最大允许线程数`Integer.MAX_VALUE`。因此所有线程超过空闲时间 60s 就会释放，当有任务提交此时没有空闲线程或者没有线程时会 先创建一个线程 (可能会导致短时间内创建大量线程造成OOM)

使用`SynchronousQueue` 阻塞队列通过`offer()`方法放入任务，`poll()`方法获取任务

### 三、ThreadPoolExecutor源码解析 ###

#### 3.1核心属性 ####

// 存放 当前允许的worker数量以及线程池的状态    
       // int类型数据是4字节32位，高三位表示线程池状态的标志位，后29位表示当前运行worker的数量
        private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
        private static final int COUNT_BITS = Integer.SIZE - 3;
        private static final int COUNT_MASK = (1 << COUNT_BITS) - 1;
    
    
        // 高3位为 111 该状态的线程池会接收新的任务，处理阻塞队列中的任务
        private static final int RUNNING    = -1 << COUNT_BITS;
        // 高3位为 000 该状态的线程池 不会接收新任务，会处理阻塞队列中的任务，中断所有没有执行的线程
        private static final int SHUTDOWN   =  0 << COUNT_BITS;
        // 高3位为 001 该状态的线程不会接收新任务，不会处理阻塞队列中的任务，中断正在运行的线程
        private static final int STOP       =  1 << COUNT_BITS;
        // 高3位为 010 表示所有任务都已经终止
        private static final int TIDYING    =  2 << COUNT_BITS;
        // 高3位为 011  terminated()方法执行完成 线程池终止
        private static final int TERMINATED =  3 << COUNT_BITS;
    
        // Packing and unpacking ctl
        private static int runStateOf(int c)     {
         return c & ~COUNT_MASK; }
        private static int workerCountOf(int c)  {
         return c & COUNT_MASK; }
        private static int ctlOf(int rs, int wc) {
         return rs | wc; }
    
        // 存放任务的阻塞队列
        private final BlockingQueue<Runnable> workQueue;
    
        // worker 的set集合存放线程
        private final HashSet<Worker> workers = new HashSet<>();

**线程池状态的切换流程**  
![在这里插入图片描述][6889cea016384c2b8bf7b8c772264c20.png]

#### 3.2任务的提交 ####

**1.submit()方法**

在`AbstractExecutorService`抽象类中`submit()`的实现，`AbstractExecutorService`类是实现了`ExecutorService`接口，而`ThreadPoolExecutor`是`AbstractExecutorService`的子类

public <T> Future<T> submit(Callable<T> task) {
        
            if (task == null) throw new NullPointerException();
            // 将通过submit方法提交的callable任务封装成一个 futureTask对象可以获取执行的结果
            RunnableFuture<T> ftask = newTaskFor(task);
            // execute()执行任务
            execute(ftask);
            // 返回结果
            return ftask;
        }

通过`submit()`方法提交的`Callable`或`Runnable`任务会被封装成了一个`FutureTask`对象。通过`Executor.execute()`方法提交`FutureTask`到线程池中等待被执行，最终执行的是`FutureTask`中的`run()`方法

**2.FutureTask类**

任务的七种状态

private volatile int state;
        // 新建
        private static final int NEW          = 0;
        // 完成中
        private static final int COMPLETING   = 1;
        // 正常
        private static final int NORMAL       = 2;
        // 异常
        private static final int EXCEPTIONAL  = 3;
        // 取消
        private static final int CANCELLED    = 4;
        // 正在打断
        private static final int INTERRUPTING = 5;
        // 已经被打断
        private static final int INTERRUPTED  = 6;

**3.get()方法**

获取任务的执行结果

public V get() throws InterruptedException, ExecutionException {
        
            int s = state;
            if (s <= COMPLETING)
             // 任务未完成  就调用awaitDone方法阻塞主线程 等待执行结果
                s = awaitDone(false, 0L);
            return report(s);
        }

**4.awaitDone()方法**

private int awaitDone(boolean timed, long nanos) throws InterruptedException {
        
           
            long startTime = 0L;    
            WaitNode q = null;
            boolean queued = false;
            for (;;) {
        
                int s = state;
               // 如果任务处于完成之后的状态 直接返回
                if (s > COMPLETING) {
        
                    if (q != null)
                        q.thread = null;
                    return s;
                }
                // 如果任务正在完成中等待任务完成
                else if (s == COMPLETING)
                   // yield释放cpu时间片等待下一次cpu执行
                    Thread.yield();
                else if (Thread.interrupted()) {
        
                    // 如果主线程中断，抛出异常
                    removeWaiter(q);
                    throw new InterruptedException();
                }
                // q为null 新建一个WaitNode
                else if (q == null) {
        
                    if (timed && nanos <= 0L)
                        return s;
                    q = new WaitNode();
                }
                // q没有入队，通过CAS将其入队
                else if (!queued)
                    queued = WAITERS.weakCompareAndSet(this, q.next = waiters, q);
                else if (timed) {
        
                    
                    final long parkNanos;
                    if (startTime == 0L) {
         // 首次执行
                        startTime = System.nanoTime();
                        if (startTime == 0L)
                            startTime = 1L;
                        parkNanos = nanos;
                    } else {
        
                        long elapsed = System.nanoTime() - startTime;
                        if (elapsed >= nanos) {
        
                            removeWaiter(q);
                            return state;
                        }
                        parkNanos = nanos - elapsed;
                    }
                    // 停止前再次检查状态
                    if (state < COMPLETING)
                        // 挂起线程
                        LockSupport.parkNanos(this, parkNanos);
                }
                else
                    LockSupport.park(this);
            }
    }

**5.run()方法**

执行任务的call()方法，如果执行成功，保存结果。如果执行有异常，抛出异常并记录。

public void run() {
        
        // 检查任务状态和执行的线程，如果状态不是NEW,或者通过CAS设置当前线程为任务执行线程失败，直接返回
            if (state != NEW ||
                !RUNNER.compareAndSet(this, null, Thread.currentThread()))
                return;
            try {
        
                Callable<V> c = callable;
                if (c != null && state == NEW) {
        
                    // 判断任务状态
                    V result;
                    boolean ran;
                    try {
        
                        // 执行任务的call() 方法
                        result = c.call();
                        // 执行完成
                        ran = true;
                    } catch (Throwable ex) {
        
                        // 抛出异常并记录
                        result = null;
                        ran = false;
                        setException(ex);
                    }
                    if (ran)
                        set(result);
                }
            } finally {
        
                // runner 设置为null，保证其他线程可以再次执行此任务
                
                runner = null;
                // 重读状态
                
                int s = state;
                if (s >= INTERRUPTING)
                    // 线程中断
                    handlePossibleCancellationInterrupt(s);
            }
        }

#### 3.3任务的执行 ####

**1.execute()方法**

`ThreadPoolExecutor`中的`execute()`实现了`Executor`中的`execute()`

public void execute(Runnable command) {
        
            if (command == null)
                throw new NullPointerException();
          
            int c = ctl.get();
         // workerCountOf获取线程池中当前线程数，小于corePoolSize 执行addWorker()方法创建新线程        //   执行command任务
            if (workerCountOf(c) < corePoolSize) {
        
                if (addWorker(command, true))
                    return;
                c = ctl.get();
            }
         //  线程池中当前线程数，大于corePoolSize
         // 线程池处于Running状态，且提交的任务成功放入阻塞队列
            if (isRunning(c) && workQueue.offer(command)) {
        
                int recheck = ctl.get();
      // 再次获取ctl的值 判断线程池的状态 如果线程池不处于Running状态，就remove(command)
    
                if (! isRunning(recheck) && remove(command))
              // 线程池不在Running状态 从阻塞队列中移除任务，执行reject处理任务
                    reject(command);
              // 线程池处于Running状态 但是没有线程，创建线程
                else if (workerCountOf(recheck) == 0)
                    addWorker(null, false);
            }
            // 线程池中创建新线程失败，执行reject处理任务
            else if (!addWorker(command, false))
                reject(command);
        }

两次获取`ctl的值`检测线程池的状态的原因

在多线程环境下，线程池的状态时刻变化，而`ctl.get()`是非原子操作，在判断线程池是否是`Running`状态前获取的`ctl.get()` 可能在此期间发生了变化。如果没有两次检测，而线程池又恰好处于非`Running`状态，那么`command`将永远不会执行

**2.addWorker() 方法**

addWorker()主要负责创建新的线程并执行任务

// 定义的全局 可重入锁
       private final ReentrantLock mainLock = new ReentrantLock();
      
       private boolean addWorker(Runnable firstTask, boolean core) {
        
            // 尝试通过CAS更新线程池的数量
            retry:
            for (int c = ctl.get();;) {
        
                // Check if queue empty only if necessary.
                if (runStateAtLeast(c, SHUTDOWN)
                    && (runStateAtLeast(c, STOP)
                        || firstTask != null
                        || workQueue.isEmpty()))
                    return false;
    
                for (;;) {
        
                    if (workerCountOf(c)
                        >= ((core ? corePoolSize : maximumPoolSize) & COUNT_MASK))
                        return false;
                    if (compareAndIncrementWorkerCount(c))
                        break retry;
                    c = ctl.get();  // Re-read ctl
                    if (runStateAtLeast(c, SHUTDOWN))
                        continue retry;
                    // CAS失败 重试
                }
            }
    
            boolean workerStarted = false;
            boolean workerAdded = false;
            Worker w = null;
            try {
        
                w = new Worker(firstTask);
                final Thread t = w.thread;
                if (t != null) {
        
                    final ReentrantLock mainLock = this.mainLock;
                    mainLock.lock();
                    try {
        
                        // Recheck while holding lock.
                        // Back out on ThreadFactory failure or if
                        // shut down before lock acquired.
                        int c = ctl.get();
    
                        if (isRunning(c) ||
                            (runStateLessThan(c, STOP) && firstTask == null)) {
        
                            // 线程不是创建状态抛出异常
                            if (t.getState() != Thread.State.NEW)
                                throw new IllegalThreadStateException();
                            // 添加任务worker 到workers中
                            workers.add(w);
                            // worker添加成功
                            workerAdded = true;
               // 获取workers的大小 与曾经的worker的最大数比较 如果超过就修改最大数为当前worker数量
                            int s = workers.size();
                            if (s > largestPoolSize)
                                largestPoolSize = s;
                        }
                    } finally {
        
                        // 释放锁
                        mainLock.unlock();
                    }
                    if (workerAdded) {
        
                       // worker添加成功意味着线程创建成功且可以处理任务
                       // 启动线程执行任务
                        t.start();
                        workerStarted = true;
                    }
                }
            } finally {
        
                if (! workerStarted)
                    addWorkerFailed(w);
            }
            return workerStarted;
        }

**3.worker类**

`ThreadPoolExecutor`的核心类`worker`

*  继承`AQS`方便对线程进行操作
 *  实现`Runnable`接口，可以将自身作为一个任务在线程中执行

private final class Worker extends AbstractQueuedSynchronizer implements Runnable
     {
         
            // 运行任务的线程，通过启动线程来实现worker对任务的处理
              final Thread thread;
            
            // 定义一个任务，当一个worker创建时，尝试执行这个任务
            Runnable firstTask;
           // 记录完成任务的数量
            volatile long completedTasks;
            
           Worker(Runnable firstTask) {
        
                setState(-1); // inhibit interrupts until runWorker
               // worker的有参构造函数，参数就是传入的任务
                this.firstTask = firstTask;
               // 创建一个新的线程  即thread的实例化 
                this.thread = getThreadFactory().newThread(this);
            }
    
            // 线程启动 本质上是执行worker的run()方法
            public void run() {
        
                // 线程执行任务就是在运行worker
                runWorker(this);
            }
            
            // ......
    }

**4.runWorker()方法**

work类的核心方法，线程启动运行run()实质上就是在调用runWorker()方法，主要作用是执行任务

final void runWorker(Worker w) {
        
            Thread wt = Thread.currentThread();
            Runnable task = w.firstTask;
            w.firstTask = null;
         // 释放锁 unlock()会设置AQS的state为0 表示运行可以中断 
         // 如果线程池要进入stop状态，确保线程可以被中断
            w.unlock(); 
            boolean completedAbruptly = true;
            try {
        
                // worker执行firstTask或者从workQueue中获取任务
                while (task != null || (task = getTask()) != null) {
        
                    // 加锁 确保线程不被其他线程 中断
                    w.lock();
                    
                   // 检测线程池的状态，如果处于stop状态 那么中断线程
                    if ((runStateAtLeast(ctl.get(), STOP) ||
                         (Thread.interrupted() &&
                          runStateAtLeast(ctl.get(), STOP))) &&
                        !wt.isInterrupted())
                        wt.interrupt();
                    try {
        
                        // 执行前准备阶段，其实就是创建worker
                        beforeExecute(wt, task);
                        try {
        
                            // 执行任务的 run()方法
                            task.run();
                            // 执行任务后阶段
                            afterExecute(task, null);
                        } catch (Throwable ex) {
        
                            // 出现异常就捕获异常 并记录
                            afterExecute(task, ex);
                            throw ex;
                        }
                    } finally {
        
                        task = null;
                        // 任务执行完成 完成任务数量+1 并解锁
                        w.completedTasks++;
                        w.unlock();
                    }
                }
                completedAbruptly = false;
            } finally {
        
                processWorkerExit(w, completedAbruptly);
            }
        }

**5.getTask()方法**

主要的负责从阻塞队列中获取任务 和 销毁非核心线程的空闲线程

private Runnable getTask() {
        
            boolean timedOut = false; // Did the last poll() time out?
    
            for (;;) {
        
                int c = ctl.get();
    
                // 检测线程池的状态 和 任务队列是否为空
                if (runStateAtLeast(c, SHUTDOWN)
                    && (runStateAtLeast(c, STOP) || workQueue.isEmpty())) {
        
                    decrementWorkerCount();
                    return null;
                }
               // 获取worker的数量
                int wc = workerCountOf(c);
    
           // 通过判断 核心线程是否会被销毁标识 或 worker数量是否大于核心线程数 来决定是否进行线程销毁
                boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
    
                if ((wc > maximumPoolSize || (timed && timedOut))
                    && (wc > 1 || workQueue.isEmpty())) {
        
                    if (compareAndDecrementWorkerCount(c))
                        return null;
                    continue;
                }
    
                try {
        
                    // 如果当前空闲的线程不需要被销毁，timed = false 调用take()执行任务，
                    // 如果此时队列为空，则当前线程被挂起，等待被唤醒
                    Runnable r = timed ?
            // 当前空闲线程需要销毁 就在keepAliveTime时间内获取队列中的任务，
                    //    如果一直没有获取到(超过空闲时间) 就销毁该线程
                        workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                        workQueue.take();
                    if (r != null)
                        return r;
                    timedOut = true;
                } catch (InterruptedException retry) {
        
                    timedOut = false;
                }
            }
        }

`allowCoreThreadTimeOut` 核心线程空闲时的处理方案，默认为false表示核心线程空闲保存活跃状态，如果指定为true，那么核心线程空闲时也会被销毁

**总结：**

线程池的工作线程通过`Woker`类实现，在`ReentrantLock`锁的保证下，把创建的`Woker`对象添加到`HashSet`集合之后，启动`Woker`中的线程。 当执行`start()`方法启动线程`thread`时，本质是执行了`Worker`类的`runWorker()`方法。`runWorker()`方法中执行任务的`run()`方法。任务执行完成之后，通过`getTask()`方法从阻塞队列中获取等待的任务，如果队列中没有任务，`getTask()`方法会被阻塞并挂起，不会占用cpu资源；

#### 3.4任务的关闭 ####

**1.shutdown()方法**

`shutdown`方法会将线程池的状态设置为`SHUTDOWN`,线程池进入这个状态后,就拒绝再接受任务,然后会将剩余的任务全部执行完

public void shutdown() {
        
            final ReentrantLock mainLock = this.mainLock;
            mainLock.lock();
            try {
        
                // 检查是否可以关闭线程
                checkShutdownAccess();
                // 设置线程池的状态
                advanceRunState(SHUTDOWN);
                //  尝试中断worker
                interruptIdleWorkers();
                // 钩子方法，预留
                onShutdown(); 
            } finally {
        
                mainLock.unlock();
            }
            tryTerminate();
        }
    
    
    
     private void interruptIdleWorkers(boolean onlyOne) {
        
            final ReentrantLock mainLock = this.mainLock;
            mainLock.lock();
            try {
        
                // 遍历所有worker
                for (Worker w : workers) {
        
                    Thread t = w.thread;
                    // 尝试获取worker的锁，如果获取到说明worker是空闲的直接中断
                    // worker实现了AQS同步框架有锁的功能，实现的锁是不可重入的，
                   // worker在执行任务会先加锁 此时tryLock()返回false
                    if (!t.isInterrupted() && w.tryLock()) {
        
                        try {
        
                            t.interrupt();
                        } catch (SecurityException ignore) {
        
                        } finally {
        
                            w.unlock();
                        }
                    }
                    if (onlyOne)
                        break;
                }
            } finally {
        
                mainLock.unlock();
            }
    }

**2.shutdownNow()方法**

将线程池状态设置为`STOP`，然后拒绝所有提交的任务，中断所有`worker`包括正在执行的，最后清空任务队列

public List<Runnable> shutdownNow() {
        
            List<Runnable> tasks;
            final ReentrantLock mainLock = this.mainLock;
            mainLock.lock();
            try {
        
                // 检查是否可以关闭线程
                checkShutdownAccess();
                // 设置线程池状态
                advanceRunState(STOP);
                // 中断所有worker
                interruptWorkers();
                // 清空任务队列
                tasks = drainQueue();
            } finally {
        
                mainLock.unlock();
            }
            tryTerminate();
            return tasks;
        }
    
      private void interruptWorkers() {
        
            // 遍历所有worker,进行中断
            for (Worker w : workers)
                w.interruptIfStarted();
        }
            void interruptIfStarted() {
        
                Thread t;
                if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {
        
                    try {
        
                        t.interrupt();
                    } catch (SecurityException ignore) {
        
                    }
                }
            }
        }

### 四、使用线程池 ###

**1.使用方式**

避免使用`Executors`去创建，通过`ThreadPoolExecutor`创建线程池，根据使用的场景自定义线程池中的核心参数

*  `FixedThreadPool` 和 `SingleThreadPool`:
    
    允许阻塞队列长度为`Integer.MAX_VALUE`，可能会堆积大量的请求，从而导致 OOM。
 *  `CachedThreadPool` 和 `ScheduledThreadPool`:
    
    允许创建线程的数量为 `Integer.MAX_VALUE`， 可能会在一段时间内创建大量的线程，造成大量资源开销，甚至导致 OOM。

可以通过引如第三方工具包比如：`commons-lang3` 、`com.google.guava` 来实现线程池，当然最简单直接的还是自定义`ThreadPoolExecutor`的参数创建线程池。

在整合框架时，可以在配置文件中配置线程池的参数

**线程池使用例子1：**  
`execute()`执行任务，无返回值

public class ThreadPoolDemo {
        
        public static void main(String[] args) {
        
            // 核心线程数
            int coreThreadNum = 5;
            //最大线程数控制
            int maxThreadNum = 10;
            ExecutorService executor = new ThreadPoolExecutor(coreThreadNum, maxThreadNum, 10, TimeUnit.SECONDS,
                    new ArrayBlockingQueue<>(1000), Executors.defaultThreadFactory(), new ThreadPoolExecutor.CallerRunsPolicy());
            for (int i = 0; i < 5; i++) {
        
    
                executor.execute(new RunnableTask(i));
            }
            //关闭线程池
            executor.shutdown();
    
        }
    
    
        static class RunnableTask implements Runnable {
        
            private int idx;
    
            public RunnableTask(int i) {
        
                this.idx = i;
            }
    
            @Override
            public void run() {
        
                //业务处理
                System.out.println(Thread.currentThread().getName() + " " + idx);
            }
        }
    }
    /* 运行结果
    pool-1-thread-1 0
    pool-1-thread-3 2
    pool-1-thread-4 3
    pool-1-thread-2 1
    pool-1-thread-5 4
     */

**线程池使用例子2：**

`submit()`提交任务有返回值

public class ThreadPoolDemo {
        
        public static void main(String[] args) {
        
            // 核心线程数
            int coreThreadNum = 5;
            //最大线程数控制
            int maxThreadNum = 10;
            ExecutorService executor = new ThreadPoolExecutor(coreThreadNum, maxThreadNum, 10, TimeUnit.SECONDS,
                    new ArrayBlockingQueue<>(1000), Executors.defaultThreadFactory(), new ThreadPoolExecutor.CallerRunsPolicy());
    
           Future<String> future = executor.submit(new CallableTask());
    
    
                try {
        
                    final String s = future.get();
                    System.out.println(s);
                } catch (Exception e) {
        
                    e.printStackTrace();
                }
    
            //关闭线程池
            executor.shutdown();
    
        }
    
        static class CallableTask implements Callable<String> {
        
            @Override
            public String call() throws Exception {
        
                System.out.println(Thread.currentThread().getName() + " ");
                return "result";
            }
        }
    }

**2.配置线程池考虑因素**

*  **CPU密集型：** 尽可能少的线程，推荐cpu核数+1
 *  **IO密集型：** 尽可能多的线程，推荐cpu核数的2倍，例如数据库连接池
 *  **混合型：** 如果CPU密集型的任务与IO密集型任务的执行时间差别较小，拆分为两个线程池，否则就是用一个均衡线程数的线程池

**参考文章：**

*  《Java并发编程艺术》
 *  https://pdai.tech/md/java/thread/java-thread-x-juc-executor-ThreadPoolExecutor.html
 *  https://blog.csdn.net/programmer\_at/article/details/79799267
 *  https://blog.csdn.net/u013332124/article/details/79587436
 *  https://www.journaldev.com/1069/threadpoolexecutor-java-thread-pool-example-executorservice
 *  https://www.cnblogs.com/spec-dog/p/11149741.html
 *  https://blog.csdn.net/huangzhilin2015/article/details/116550409

[9198e17349c44d8197e0592190a71729.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/30/711e2b684a564f4eb43942dc6e9de361.png
[39fccc80e3b249d7ae2728b662326dcc.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/30/344284a3a1804b85b87a2ec4b518fceb.png
[6889cea016384c2b8bf7b8c772264c20.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/30/f04141b9341a4c809971b46dd659d52e.png