并发编程（第八章共享模型之工具 - 线程池）

女爷i 2024-04-08 11:10 69阅读 0赞

#### 文章目录 ####

*   *  线程池
     *   *  一、自定义线程池
         *  二、ThreadPoolExecutor
         *   *  1、线程池状态
             *  2、构造方法
             *  3、newFixedThreadPool
             *  4、newCachedThreadPool
             *  5、newSingleThreadExecutor
             *  6、提交任务
             *  7、关闭线程池
             *  8、模式之Worker Thread（重点）
             *  9、任务调度线程池
             *  10、正确处理执行任务异常
             *  11、应用之定时任务（重点）
             *  12、Tomcat线程池
         *  三、Fork/Join

### 线程池 ###

#### 一、自定义线程池 ####

![在这里插入图片描述][75388740ed954a6ea8baf44ed73f6c4f.png]

> 步骤1:自定义拒绝策略接口

@FunctionalInterface // 拒绝策略 
    interface RejectPolicy<T> {
        
    	void reject(BlockingQueue<T> queue, T task); 
    }

> 步骤2:自定义任务队列

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 capcity;
    	
    	public BlockingQueue(int capcity) {
         
    		this.capcity = capcity;
    	}
    	// 带超时阻塞获取
    	public T poll(long timeout, TimeUnit unit) {
        
    		lock.lock(); 
    		try {
        
    			// 将 timeout 统一转换为 纳秒
    			long nanos = unit.toNanos(timeout); 
    			while (queue.isEmpty()) {
        
    				try {
        
    					// 返回值是剩余时间 
    					if (nanos <= 0) {
        
    						return null; 
    					}
    					nanos = emptyWaitSet.awaitNanos(nanos); 
    				} catch (InterruptedException e) {
        
    					e.printStackTrace(); 
    				}
    			}
    			T t = queue.removeFirst(); 
    			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(); 
    			fullWaitSet.signal(); 
    			return t;
    		} finally {
         
    			lock.unlock();
    		} 
    	}
    	
    	// 阻塞添加
    	public void put(T task) {
        
    		lock.lock(); 
    		try {
        
    			while (queue.size() == capcity) {
         
    				try {
        
    					log.debug("等待加入任务队列 {} ...", task);
    					fullWaitSet.await();
    				} catch (InterruptedException e) {
        
    					e.printStackTrace(); 
    				}
    			}
    			log.debug("加入任务队列 {}", task); 
    			queue.addLast(task); 
    			emptyWaitSet.signal();
    		} finally {
         
    			lock.unlock();
    		} 
    	}
    	
    	// 带超时时间阻塞添加
    	public boolean offer(T task, long timeout, TimeUnit timeUnit) {
        
    		lock.lock(); 
    		try {
        
    			long nanos = timeUnit.toNanos(timeout); 
    			while (queue.size() == capcity) {
        
    				try {
        
    					if(nanos <= 0) {
        
    						return false; 
    					}
    					log.debug("等待加入任务队列 {} ...", task); 
    					nanos =	 fullWaitSet.awaitNanos(nanos);
    				} catch (InterruptedException e) {
         
    					e.printStackTrace();
    				} 
    			}
    			log.debug("加入任务队列 {}", task); 
    			queue.addLast(task); 
    			emptyWaitSet.signal();
    			return true;
    		} finally {
         
    			lock.unlock();
    		} 
    	}
    	
    	public int size() {
         
    		lock.lock();
    		try {
        
    			return queue.size();
    		} finally {
         
    			lock.unlock();
    		} 
    	}
    	
    	public void tryPut(RejectPolicy<T> rejectPolicy, T task) {
         
    		lock.lock();
    		try {
        
    			// 判断队列是否满
    			if(queue.size() == capcity) {
         
    				rejectPolicy.reject(this, task);
    			} else {
         
    				// 有空闲 
    				log.debug("加入任务队列 {}", task); 
    				queue.addLast(task); 
    				emptyWaitSet.signal();
    			}
    		} finally {
        
    			lock.unlock(); 
    		}
    	} 
    }

> 步骤3:自定义线程池

class ThreadPool {
        
        // 任务队列
        private BlockingQueue<Runnable> taskQueue;
    
        // 线程集合
        private HashSet<Worker> workers = new HashSet();
    
        // 核心线程数
        private int coreSize;
    
        // 获取任务的超时时间
        private long timeout;
    
        private TimeUnit timeUnit;
    
        private RejectPolicy<Runnable> rejectPolicy;
    
        // 执行任务
        public void execute(Runnable task) {
        
            // 当任务数没有超过 coreSize时，直接交给 worker 对象执行
            // 如果任务数超过coreSize时，加入任务队列暂存
            synchronized (workers) {
        
                if (workers.size() < coreSize) {
        
                    Worker worker = new Worker(task);
                    workers.add(worker);
                    worker.start();
                } else {
        
                    taskQueue.put(task);
                    // 1、死等
                    // 2、带超时等待
                    // 3、让调用者放弃任务执行
                    // 4、让调用者抛出异常
                    // 5、让调用者自己执行任务
                    taskQueue.tryPut(rejectPolicy, task);
                }
            }
        }
    
        public ThreadPool(int coreSize, long timeout, TimeUnit timeUnit, int queueCapcity, RejectPolicy<Runnable> rejectPolicy) {
        
            this.taskQueue = new BlockingQueue<>(queueCapcity);
            this.coreSize = coreSize;
            this.timeout = timeout;
            this.timeUnit = timeUnit;
            this.rejectPolicy = rejectPolicy;
        }
    
        class Worker extends Thread {
        
            private Runnable task;
    
            public Worker(Runnable task) {
        
                this.task = task;
            }
    
            @Override
            public void run() {
        
                // 执行任务
                // 1、当task不为空，执行任务
                // 2、当task执行完毕，再接着从任务队列获取任务并执行
                //while (task != null || (task = taskQueue.take()) != null) {
        
                while (task != null || (task = taskQueue.poll(timeout, timeUnit)) != null) {
        
                    try{
        
                        task.run();
                    } catch (Exception e) {
        
                        e.printStackTrace();
                    } finally {
        
                        task = null;
                    }
                }
                synchronized (workers) {
        
                    workers.remove(this);
                }
            }
        }
    }

> 步骤4:测试

public static void main(String[] args) {
         
    	ThreadPool threadPool = new ThreadPool(1,
    		1000, TimeUnit.MILLISECONDS, 1, (queue, task)->{
         
    			// 1. 死等
    			queue.put(task); 
    			// 2) 带超时等待
    			queue.offer(task, 1500, TimeUnit.MILLISECONDS); 
    			// 3) 让调用者放弃任务执行
    			log.debug("放弃{}", task); 
    			// 4) 让调用者抛出异常
    			throw new RuntimeException("任务执行失败 " + task); 
    			// 5) 让调用者自己执行任务
    			task.run();
    	});
    	for (int i = 0; i < 4; i++) {
        
    		int j = i; threadPool.execute(() -> {
        
    			try {
         
    				Thread.sleep(1000L);
    			} catch (InterruptedException e) {
         
    				e.printStackTrace();
    			}
    			log.debug("{}", j); 
    		});
    	} 
    }

#### 二、ThreadPoolExecutor ####

##### 1、线程池状态 #####

![在这里插入图片描述][dbf0199c8ce0473eb58345c81b28a95d.png]  
**1.1 线程池状态**

> ThreadPoolExecutor 使用int 的高3位来表示线程池状态，低29位表示线程数量

<table> 
 <thead> 
  <tr> 
   <th>状态名</th> 
   <th>高3位</th> 
   <th>接收新任务</th> 
   <th>处理阻塞队列任务</th> 
   <th>说明</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>RUNNING</td> 
   <td>111</td> 
   <td>Y</td> 
   <td>Y</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>SHUTDOWN</td> 
   <td>000</td> 
   <td>N</td> 
   <td>Y</td> 
   <td>不会接收新任务，但会处理阻塞队列剩余任务</td> 
  </tr> 
  <tr> 
   <td>STOP</td> 
   <td>001</td> 
   <td>N</td> 
   <td>N</td> 
   <td>会中断正在执行的任务，并抛弃阻塞队列任务</td> 
  </tr> 
  <tr> 
   <td>TIDYING</td> 
   <td>010</td> 
   <td>–</td> 
   <td>–</td> 
   <td>任务全执行完毕，活动线程位0即将进入终结</td> 
  </tr> 
  <tr> 
   <td>TERMINATED</td> 
   <td>011</td> 
   <td>–</td> 
   <td>–</td> 
   <td>终结状态</td> 
  </tr> 
 </tbody> 
</table>

> 从数字上比较，`TERMINATED > TIDYING > STOP > SHUTDOWN > RUNNING`

> 这些信息存储在一个原子变量ctl中，目的是将线程池状态与线程个数合二为一，这样就可以用一次cas原子操作进行赋值

// c 为旧值， ctlOf 返回结果为新值
    ctl.compareAndSet(c, ctlOf(targetState, workerCountOf(c))));
    
    // rs 为高 3 位代表线程池状态， wc 为低 29 位代表线程个数，ctl 是合并它们 
    private static int ctlOf(int rs, int wc) {
         return rs | wc; }

##### 2、构造方法 #####

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

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

> 工作方式：  
> ![在这里插入图片描述][d7f0615b323e4460898d8466f70badc6.png]
> 
>  *  线程池中刚开始没有线程，当一个任务提交给线程池后，线程池会创建一个新线程来执行任务。
>  *  当线程数达到corePoolSize并没有线程空闲，这时再加入任务，新加的任务会被加入workQueue队列排队，直到有空闲的线程。
>  *  如果队列选择了有界队列，那么任务超过了队列大小时，会创建maximumPoolSize - corePoolSize数目的线程来救急。
>  *  如果线程达到maximumPoolSize仍然有新任务这时会执行拒绝策略。拒绝策略jdk提供了4种实现，其它著名框架也提供了实现。  
>     – AbortPolicy让调用者抛出RejectedExecutionExeception异常，这是默认策略  
>     – CallRunsPolicy让调用者运行任务  
>     – DiscardPolicy放弃本次任务  
>     – DiscardOldestPolicy放弃队列中最早的任务，本任务取而代之  
>     – Dubbo的实现，在抛出RejectedExecutionException异常之前会记录日志，并dump线程栈信息，方便定位问题  
>     – Netty的实现，是创建一个新线程来执行任务  
>     – ActiveMQ的实现，带超时等待（60s）尝试放入队列，类似我们之前自定义的拒绝策略  
>     – PinPoint的实现，它使用了一个拒绝策略链，会逐一尝试策略链中每种拒绝策略
>  *  当高峰过去后，超过corePoolSize的救急线程如果一段时间没有任务做，需要结束节省资源，这个时间由keepAliveTime 和 unit来控制。

> ![在这里插入图片描述][a69c472079454628a309b46f1e2f37ba.png]  
> 根据这个构造方法，JDK Executors类中提供了众多工厂方法来创建各种用途的线程池

##### 3、newFixedThreadPool #####

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

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

> **评价**  
> 适用于任务量已知，相对耗时的任务

##### 4、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实现特点是，它没有容量，没有线程来取式放不进去的（一手交钱、一手交货）

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();
    
    sleep(1);
    
    new Thread(() -> {
         
    	try {
        
    		log.debug("taking {}", 1);
    		integers.take();
    	} catch (InterruptedException e) {
        
    		e.printStackTrace(); 
    	}
    },"t2").start();
    
    sleep(1);
    
    new Thread(() -> {
         
    	try {
        
    		log.debug("taking {}", 2);
    		integers.take();
    	} catch (InterruptedException e) {
        
    		e.printStackTrace(); 
    	}
    },"t3").start();

> 输出：  
> ![在这里插入图片描述][ee6e13c2d70847ccb7ea55b6f9e0dc83.png]

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

##### 5、newSingleThreadExecutor #####

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

> **使用场景：**  
> 也希望多个任务排队执行。线程数固定为1，任务数多于1时，会放入无界队列排队。任务执行完毕，这唯一的线程也不会被释放。  
> **区别：**
> 
>  *  自己创建一个单线程串行执行任务，如果任务执行失败而终止那么没有任何补救措施，而线程池还会新建一个线程，保证池的正常工作；
>  *  Executors.newSingleThreadExecutor()线程个数始终为1，不能修改  
>     – FinalizableDelegatedExecutorService应用的是装饰器模式，只对外暴露了ExecutorService接口，因此不能调用ThreadPoolExecutor中特有的方法
>  *  Executors.newFixedThreadPool(1)初始时为1，以后还可以修改  
>     – 对外暴露的是ThreadPoolExecutor对象，可以强转后调用setCorePoolSize等方法进行修改

##### 6、提交任务 #####

// 执行任务
    void execute(Runnable command);
    
    // 提交任务 task，用返回值 Future 获得任务执行结果 
    <T> Future<T> submit(Callable<T> task);
    
    // 提交 tasks 中所有任务
    <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
    throws InterruptedException;
    
    // 提交 tasks 中所有任务，带超时时间
    <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks,
    long timeout, TimeUnit unit) throws InterruptedException;
    
    // 提交 tasks 中所有任务，哪个任务先成功执行完毕，返回此任务执行结果，其它任务取消 
    <T> T invokeAny(Collection<? extends Callable<T>> tasks)
    throws InterruptedException, ExecutionException;
    
     // 提交 tasks 中所有任务，哪个任务先成功执行完毕，返回此任务执行结果，其它任务取消，带超时时间 
     <T> T invokeAny(Collection<? extends Callable<T>> tasks,
    long timeout, TimeUnit unit)
    throws InterruptedException, ExecutionException, TimeoutException;

##### 7、关闭线程池 #####

**shutdown**

/*
    线程池状态变为 SHUTDOWN
    - 不会接收新任务
    - 但已提交任务会执行完
    - 此方法不会阻塞调用线程的执行
    */
    void shutdown();

public void shutdown() {
        
    	final ReentrantLock mainLock = this.mainLock; 
    	mainLock.lock();
    	try {
        
    		checkShutdownAccess();
    		// 修改线程池状态
    		advanceRunState(SHUTDOWN);
    		// 仅会打断空闲线程
    		interruptIdleWorkers();
    		onShutdown(); // 扩展点 ScheduledThreadPoolExecutor
    	} finally {
         
    		mainLock.unlock();
    	}
    	// 尝试终结(没有运行的线程可以立刻终结，如果还有运行的线程也不会等)
    	tryTerminate(); 
    }

**shutdownNow**

/*
    线程池状态变为 STOP
    - 不会接收新任务
    - 会将队列中的任务返回
    - 并用 interrupt 的方式中断正在执行的任务
    */
    List<Runnable> shutdownNow();

public List<Runnable> shutdownNow() {
        
    	List<Runnable> tasks;
    	final ReentrantLock mainLock = this.mainLock; mainLock.lock();
    	try {
        
    		checkShutdownAccess(); 
    		// 修改线程池状态 
    		advanceRunState(STOP); 
    		// 打断所有线程 
    		interruptWorkers();
    		// 获取队列中剩余任务
    		tasks = drainQueue(); 
    	} finally {
        
    		mainLock.unlock(); 
    	}
    	// 尝试终结 
    	tryTerminate(); 
    	return tasks;
    }

**其它方法**

// 不在 RUNNING 状态的线程池，此方法就返回 true 
    boolean isShutdown();
    
    // 线程池状态是否是 TERMINATED 
    boolean isTerminated();
    
    // 调用 shutdown 后，由于调用线程并不会等待所有任务运行结束，因此如果它想在线程池 TERMINATED 后做些事 情，可以利用此方法等待
    boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException;

##### 8、模式之Worker Thread（重点） #####

**1、定义**

> 让有限的工作线程(Worker Thread)来轮流异步处理无限多的任务。也可以将其归类为分工模式，它的典型实现 就是线程池，也体现了经典设计模式中的享元模式。

>  *  例如，海底捞的服务员(线程)，轮流处理每位客人的点餐(任务)，如果为每位客人都配一名专属的服务员，那 么成本就太高了(对比另一种多线程设计模式:Thread-Per-Message)
>  *  注意，不同任务类型应该使用不同的线程池，这样能够避免饥饿，并能提升效率
>  *  例如，如果一个餐馆的工人既要招呼客人(任务类型A)，又要到后厨做菜(任务类型B)显然效率不咋地，分成服务员(线程池A)与厨师(线程池B)更为合理，当然你能想到更细致的分工

**2、饥饿**

> 固定大小线程池会有饥饿现象
> 
>  *  两个工人是同一个线程池的两个线程
>  *  他们要做的事情是：为客人点餐和到后厨做菜，这是两个阶段的工作  
>     – 客人点餐：必须先点完餐，等菜做好，上菜，在此期间处理点餐的工人必须等待  
>     – 后厨做菜：没啥说的，做就是了
>  *  比如工人A处理了点餐任务，接下来它要等着工人B把菜做好，然后上菜，他两也配合的蛮好
>  *  但现在同时来了两个客人，这个时候工人A和工人B都去处理点餐了，这时没人做饭了，饥饿

public class TestDeadLock {
        
    	static final List<String> MENU = Arrays.asList("地三鲜", "宫保鸡丁", "辣子鸡丁", "烤鸡翅"); 
    	static Random RANDOM = new Random();
    	static String cooking() {
        
    	return MENU.get(RANDOM.nextInt(MENU.size())); 
    	}
    	public static void main(String[] args) {
        
    		ExecutorService executorService = Executors.newFixedThreadPool(2);
    		executorService.execute(() -> {
         
    			log.debug("处理点餐...");
    			Future<String> f = executorService.submit(() -> {
        
    			log.debug("做菜");
    			return cooking(); 
    		});
    		try {
        
    			log.debug("上菜: {}", f.get());
    		} catch (InterruptedException | ExecutionException e) {
         
    			e.printStackTrace();
    		} 
    	});
    		/*executorService.execute(() -> { 
    			log.debug("处理点餐...");
    			Future<String> f = executorService.submit(() -> {
    				log.debug("做菜");
    				return cooking(); 
    			});
    			try {
    				log.debug("上菜: {}", f.get());
    			} catch (InterruptedException | ExecutionException e) {
    				e.printStackTrace(); 
    			}
    		});*/
    	} 
    }

> 输出：  
> ![在这里插入图片描述][bab09a62e8f24711b386f5aee7ed4450.png]

> 当注释取消后，可能的输出：  
> ![在这里插入图片描述][f316132ccfd14912ad05754cf825eb12.png]

> 解决方法可以增加线程池的大小，不过不是根本解决方案，还是前面提到的，不同的任务类型，采用不同的线程池，例如:

public class TestDeadLock {
        
    	static final List<String> MENU = Arrays.asList("地三鲜", "宫保鸡丁", "辣子鸡丁", "烤鸡翅"); 
    	static Random RANDOM = new Random();
    	static String cooking() {
        
    		return MENU.get(RANDOM.nextInt(MENU.size())); 
    	}
    	public static void main(String[] args) {
        
    		ExecutorService waiterPool = Executors.newFixedThreadPool(1); 
    		ExecutorService cookPool = Executors.newFixedThreadPool(1);
    		waiterPool.execute(() -> {
         
    			log.debug("处理点餐...");
    			Future<String> f = cookPool.submit(() -> {
        
    				log.debug("做菜");
    				return cooking(); 
    			});
    			try {
        
    				log.debug("上菜: {}", f.get());
    			} catch (InterruptedException | ExecutionException e) {
         	
    				 e.printStackTrace();
    			} 
    		});
    		
    		waiterPool.execute(() -> {
         
    			log.debug("处理点餐...");
    			Future<String> f = cookPool.submit(() -> {
        
    				log.debug("做菜");
    				return cooking(); 
    			});
    			try{
        
    				log.debug("上菜: {}", f.get());
    			} catch (InterruptedException | ExecutionException e) {
        
    				e.printStackTrace(); 
    			}
    		});
    	} 
    }

> 输出：  
> ![在这里插入图片描述][323c6e2fda2a4f91a5b3f83b7d7e8d3b.png]

**3、创建多少线程合适**

>  *  过小会导致程序不能充分地利用系统资源，容易导致饥饿
>  *  过大会导致更多的线程上下文切换，占用更多的内存

**3.1 CPU密集型运算**

> 通常采用 `cpu 核数 + 1` 能够实现最优的 CPU 利用率，+1 是保证当线程由于页缺失故障(操作系统)或其它原因 导致暂停时，额外的这个线程就能顶上去，保证 CPU 时钟周期不被浪费

**3.2 I/O密集型运算**

> CPU 不总是处于繁忙状态，例如，当你执行业务计算时，这时候会使用 CPU 资源，但当你执行 I/O 操作时、远程RPC 调用时，包括进行数据库操作时，这时候 CPU 就闲下来了，你可以利用多线程提高它的利用率。

> **经验公式如下**  
> `线程数 = 核数 * 期望 CPU 利用率 * 总时间(CPU计算时间+等待时间) / CPU 计算时间`
> 
>  *  例如 4 核 CPU 计算时间是 50% ，其它等待时间是 50%，期望 cpu 被 100% 利用，套用公式  
>     – `4 * 100% * 100% / 50% = 8`
>  *  例如 4 核 CPU 计算时间是 10% ，其它等待时间是 90%，期望 cpu 被 100% 利用，套用公式  
>     – `4 * 100% * 100% / 10% = 40`

##### 9、任务调度线程池 #####

> 在【任务调度线程池】功能加入之前，可以使用java.util.Timer来实现定时功能，Timer的优点在于简单易用，但由于所有任务都是由同一个线程来调度，因此所有任务都是串行执行的，同一时间只能有一个任务在执行，前一个任务的延迟或异常都将会影响到之后的任务。

public static void main(String[] args) {
         
    	Timer timer = new Timer();
    	TimerTask task1 = new TimerTask() {
        
    		@Override
    		public void run() {
         
    			log.debug("task 1");
    			sleep(2);
    		} 
    	};
    	TimerTask task2 = new TimerTask() {
         
    		@Override
    		public void run() {
         
    			log.debug("task 2");
    		} 
    	};
    	// 使用 timer 添加两个任务，希望它们都在 1s 后执行
    	// 但由于 timer 内只有一个线程来顺序执行队列中的任务，因此『任务1』的延时，影响了『任务2』的执行 
    	timer.schedule(task1, 1000);
    	timer.schedule(task2, 1000);
    }

> 输出：  
> ![在这里插入图片描述][1b6248da69f04c449151283128d18058.png]

**使用ScheduleExecutorService改写：**

ScheduledExecutorService executor = Executors.newScheduledThreadPool(2); // 添加两个任务，希望它们都在 1s 后执行
    executor.schedule(() -> {
        
    	System.out.println("任务1，执行时间:" + new Date());
    	try {
         Thread.sleep(2000); } catch (InterruptedException e) {
         } 
    }, 1000, TimeUnit.MILLISECONDS);
    executor.schedule(() -> {
        
    	System.out.println("任务2，执行时间:" + new Date()); 
    }, 1000, TimeUnit.MILLISECONDS);

> 输出：  
> ![在这里插入图片描述][6505119cf93e483caa414024dce164a1.png]

**1、scheduleAtFixedRate例子：**

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

> 输出：  
> ![在这里插入图片描述][f4e91dace17a4c0e8f14963bb3717c10.png]

**2、scheduleAtFixedRate 例子(任务执行时间超过了间隔时间):**

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

> 输出分析:一开始，延时 1s，接下来，由于任务执行时间 > 间隔时间，间隔被『撑』到了 2s  
> ![在这里插入图片描述][589fd50493084e1283ec42ebaa4d0512.png]

**3、scheduleWithFixedDelay 例子:**

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

> 输出分析:一开始，延时 1s，scheduleWithFixedDelay 的间隔是 上一个任务结束 <-> 延时 <-> 下一个任务开始 所 以间隔都是 3s  
> ![在这里插入图片描述][8f64792923454d6bb6e96c97b696f53b.png]

> **评价：**  
> 整个线程池表现为：线程数固定，任务数多于线程数时，会放入无界队列排队。任务执行完毕，这些线程也不会被释放。用来执行延迟或反复执行的任务。

##### 10、正确处理执行任务异常 #####

**方法1:主动捉异常**

ExecutorService pool = Executors.newFixedThreadPool(1); 
    pool.submit(() -> {
        
    	try {
         
    		log.debug("task1"); 
    		int i = 1 / 0;
    	} catch (Exception e) {
         
    		log.error("error:", e);
    	} 
    });

> 输出：  
> ![在这里插入图片描述][08ad9fe54f104e09b61c7745bd4f3083.png]

**方法2:使用Future**

ExecutorService pool = Executors.newFixedThreadPool(1); 
    Future<Boolean> f = pool.submit(() -> {
        
    	log.debug("task1"); 
    	int i = 1 / 0; 
    	return true;
    });
    log.debug("result:{}", f.get());

> 输出：  
> ![在这里插入图片描述][0315da0b60374d08858d09ba430917fa.png]

##### 11、应用之定时任务（重点） #####

> **定期处理：**  
> 如何让每周四 18:00:00 定时执行任务？

// 获得当前时间
    LocalDateTime now = LocalDateTime.now();
    // 获取本周四 18:00:00.000
    LocalDateTime thursday = now.with(DayOfWeek.THURSDAY).withHour(18).withMinute(0).withSecond(0).withNano(0); 
    // 如果当前时间已经超过 本周四 18:00:00.000， 那么找下周四 18:00:00.000 
    if(now.compareTo(thursday) >= 0) {
        
    	thursday = thursday.plusWeeks(1); 
    }
    // 计算时间差，即延时执行时间
    long initialDelay = Duration.between(now, thursday).toMillis(); 
    // 计算间隔时间，即 1 周的毫秒值
    long oneWeek = 7 * 24 * 3600 * 1000;
    ScheduledExecutorService executor = Executors.newScheduledThreadPool(2);
    System.out.println("开始时间:" + new Date()); executor.scheduleAtFixedRate(() -> {
        
    	System.out.println("执行时间:" + new Date()); 
    }, initialDelay, oneWeek, TimeUnit.MILLISECONDS);

##### 12、Tomcat线程池 #####

> Tomcat在哪里用到了线程池呢？  
> ![在这里插入图片描述][b2dc61d5a72243d1a9d445e860290330.png]
> 
>  *  LimitLatch用来限流，可以控制最大连接个数，类似J.U.C中的Semaphore后面讲
>  *  Acceptor只负责【接收新的socket连接】
>  *  Poller只负责监听socket channel是否有【可读的I/O事件】
>  *  一旦可读，封装一个任务对象（socketProcessor），提交给Executor线程池处理
>  *  Executot线程池中的工作线程最终负责【处理请求】

> Tomcat线程池扩展了ThreadExecutor，行为稍有不同
> 
>  *  如果总线程数达到maximumPoolSize  
>     –这时不会立刻抛RejectedExecutionException异常  
>     – 而是再次尝试将任务放入队列，如果还失败，才抛出RejectedExecutionException异常

> 源码tomcat-7.0.42

public void execute(Runnable command, long timeout, TimeUnit unit) {
        
    	submittedCount.incrementAndGet();
    	try {
        
    		super.execute(command);
    	} catch (RejectedExecutionException rx) {
        
    		if (super.getQueue() instanceof TaskQueue) {
        
    			final TaskQueue queue = (TaskQueue)super.getQueue(); 
    			try {
        
    				if (!queue.force(command, timeout, unit)) {
        
    					submittedCount.decrementAndGet();
    					throw new RejectedExecutionException("Queue capacity is full.");
    				}
    			} catch (InterruptedException x) {
        
    				submittedCount.decrementAndGet();
    				Thread.interrupted();
    				throw new RejectedExecutionException(x);
    			}
    		} else {
         
    			submittedCount.decrementAndGet(); 
    			throw rx;
    		}
    	} 
    }

> TaskQueue.java

public boolean force(Runnable o, long timeout, TimeUnit unit) throws InterruptedException {
         
    	if ( parent.isShutdown() )
    		throw new RejectedExecutionException(
    			"Executor not running, can't force a command into the queue"
    	);
    	return super.offer(o,timeout,unit); //forces the item onto the queue, to be used if the task is rejected
    }

**1、Connector配置**

<table> 
 <thead> 
  <tr> 
   <th>配置项</th> 
   <th>默认值</th> 
   <th>说明</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>acceptorThreadCount</td> 
   <td>1</td> 
   <td>acceptor 线程数量</td> 
  </tr> 
  <tr> 
   <td>pollerThreadCount</td> 
   <td>1</td> 
   <td>poller 线程数量</td> 
  </tr> 
  <tr> 
   <td>minSpareThreads</td> 
   <td>10</td> 
   <td>核心线程数，即 corePoolSize</td> 
  </tr> 
  <tr> 
   <td>maxThreads</td> 
   <td>200</td> 
   <td>最大线程数，即 maximumPoolSize</td> 
  </tr> 
  <tr> 
   <td>executor</td> 
   <td>-</td> 
   <td>Executor 名称，用来引用下面的 Executor</td> 
  </tr> 
 </tbody> 
</table>

**2、Executor线程配置**

<table> 
 <thead> 
  <tr> 
   <th>配置项</th> 
   <th>默认值</th> 
   <th>说明</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>threadPriority</td> 
   <td>5</td> 
   <td>线程优先级</td> 
  </tr> 
  <tr> 
   <td>daemon</td> 
   <td>true</td> 
   <td>是否守护线程</td> 
  </tr> 
  <tr> 
   <td>minSpareThreads</td> 
   <td>25</td> 
   <td>核心线程数，即 corePoolSize</td> 
  </tr> 
  <tr> 
   <td>maxThreads</td> 
   <td>200</td> 
   <td>最大线程数，即 maximumPoolSize</td> 
  </tr> 
  <tr> 
   <td>maxIdleTime</td> 
   <td>60000</td> 
   <td>线程生存时间，单位是毫秒，默认值即 1 分钟</td> 
  </tr> 
  <tr> 
   <td>maxQueueSize</td> 
   <td>Integer.MAX_VALUE</td> 
   <td>队列长度</td> 
  </tr> 
  <tr> 
   <td>prestartminSpareThreads</td> 
   <td>false</td> 
   <td>核心线程是否在服务器启动时启动</td> 
  </tr> 
 </tbody> 
</table>

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

#### 三、Fork/Join ####

**1、概念**

>  *  Fork/Join是JDK1.7加入的新的线程池实现，它体现的是一种分治思想，适用于能够进行任务拆分的cpu密集型运算
>  *  所谓的任务拆分，是将一个大任务拆分为算法上相同的小任务，直至不能拆分可以直接求解。跟递归相关的一些计算，如归并排序、斐波那契数列、都可以用分治思想进行求解
>  *  Fork/Join在分治的基础上加入了多线程，可以把每个任务的分解和合并交给不同的线程来完成，进一步提升了运算效率
>  *  Fork/Join默认会创建与cpu核心数大小相同的线程池

**2、使用**

> 提交给Fork/Join线程池的任务需要继承RecursiveTask（有返回值）或RecursiveAction（没有返回值），例如下面定义一个对 1 ～ n 之间的整数求和的任务

@Slf4j(topic = "c.AddTask")
    class AddTask1 extends RecursiveTask<Integer> {
         
    	int n;
    	public AddTask1(int n) {
         
    		this.n = n;
    	}
    	@Override
    	public String toString() {
         
    		return "{" + n + '}';
    	}
    	@Override
    	protected Integer compute() {
        
    		// 如果 n 已经为 1，可以求得结果了 
    		if (n == 1) {
        
    			log.debug("join() {}", n);
    			return n; 
    		}
    		// 将任务进行拆分(fork)
    		AddTask1 t1 = new AddTask1(n - 1); 
    		t1.fork();
    		log.debug("fork() {} + {}", n, t1);
    		// 合并(join)结果
    		int result = n + t1.join();
    		log.debug("join() {} + {} = {}", n, t1, result); 
    		return result;
    	} 
    }

> 然后提交给ForkJoinPool来执行

public static void main(String[] args) {
         
    	ForkJoinPool pool = new ForkJoinPool(4); 
    	System.out.println(pool.invoke(new AddTask1(5)));
    }

> **1、结果：**  
> ![在这里插入图片描述][fd2aefe88e8d43288678be1ebcb4eec6.png]  
> **2、用图来表示**  
> ![在这里插入图片描述][a1e893b1dc9a47129071189de006d2cd.png]

> 改进

class AddTask3 extends RecursiveTask<Integer> {
         int begin;
    int end;
    public AddTask3(int begin, int end) {
         this.begin = begin;
    this.end = end;
    }
    @Override
    public String toString() {
        
    return "{" + begin + "," + end + '}';
    }
    @Override
    protected Integer compute() {
         
    	// 5, 5
    	if (begin == end) {
         
    		log.debug("join() {}", begin); 
    		return begin;
    	}
    	// 4，5
    	if (end - begin == 1) {
        
    		log.debug("join() {} + {} = {}", begin, end, end + begin); 
    		return end + begin;
    	}
    	// 1 5
    	int mid = (end + begin) / 2; 
    	// 3
    	AddTask3 t1 = new AddTask3(begin, mid); // 1,3 
    	t1.fork();
    	AddTask3 t2 = new AddTask3(mid + 1, end); // 4,5 
    	t2.fork();
    	log.debug("fork() {} + {} = ?", t1, t2);
    	int result = t1.join() + t2.join(); 
    	log.debug("join() {} + {} = {}", t1, t2, result); 
    	return result;
    	} 
    }

> 然后提交给ForkJoinPool来执行

public static void main(String[] args) {
        
    	ForkJoinPool pool = new ForkJoinPool(4); 
    	System.out.println(pool.invoke(new AddTask3(1, 10)));
    }

> 1、结果：  
> ![在这里插入图片描述][7a9e91f4075641bc9e52e98e74287339.png]  
> 2、用图来表示：  
> ![在这里插入图片描述][1040ce0d23e34036b6e4ffaa0daf5817.png]

[75388740ed954a6ea8baf44ed73f6c4f.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/08/e591ecda7dcc4480af15a75634818063.png
[dbf0199c8ce0473eb58345c81b28a95d.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/08/bb2592d233b648558ed23a8759280b97.png
[d7f0615b323e4460898d8466f70badc6.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/08/b702bcc491b44f9db1c8b441d9c0b6d7.png
[a69c472079454628a309b46f1e2f37ba.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/08/7c1ab46f99964065919d181212209287.png
[ee6e13c2d70847ccb7ea55b6f9e0dc83.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/08/e91381f84d4d4a9ab5c38399084474a9.png
[bab09a62e8f24711b386f5aee7ed4450.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/08/e276f7be9ab64c039223777fb9b2e7d3.png
[f316132ccfd14912ad05754cf825eb12.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/08/d3499ecc1f77441b9e6d90f8fd9593ca.png
[323c6e2fda2a4f91a5b3f83b7d7e8d3b.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/08/d8852139bbb044008b4c30698f2be454.png
[1b6248da69f04c449151283128d18058.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/08/59c4e74d34be4e3b8f3b3c16c675066c.png
[6505119cf93e483caa414024dce164a1.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/08/7ec6651997d249c2a663441682fd5a67.png
[f4e91dace17a4c0e8f14963bb3717c10.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/08/35162f52ba914c4e829362de98807ab0.png
[589fd50493084e1283ec42ebaa4d0512.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/08/f7bb7adc19ed404f87535cd01db36c63.png
[8f64792923454d6bb6e96c97b696f53b.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/08/e80697bc9f3042c09d4161d1a037e24b.png
[08ad9fe54f104e09b61c7745bd4f3083.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/08/b4f7cf1b4c0543e996a29eb41b151350.png
[0315da0b60374d08858d09ba430917fa.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/08/8131de054c7e4667b33b72213d53c1d3.png
[b2dc61d5a72243d1a9d445e860290330.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/08/df2530469b204a03a7a5c6b826c995f5.png
[e9213b013cb24d159b8efcd8aed43163.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/08/c52f4ba3d50d451790b2836eb10be380.png
[fd2aefe88e8d43288678be1ebcb4eec6.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/08/e29ff43070804e7ca0c8dd698cbb8e23.png
[a1e893b1dc9a47129071189de006d2cd.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/08/b8d9cdc613c1460681b57eb808f0e720.png
[7a9e91f4075641bc9e52e98e74287339.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/08/938b6cfd67b745ad8d2e7b3af12baa02.png
[1040ce0d23e34036b6e4ffaa0daf5817.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/08/983dec00fa6c48f09d334120bab78da8.png