线程池

Java线程池实现原理及其在美团业务中的实践

随着计算机行业的飞速发展，摩尔定律逐渐失效，多核CPU成为主流。使用多线程并行计算逐渐成为开发人员提升服务器性能的基本武器。J.U.C提供的线程池：ThreadPoolExecutor类，帮助开发人员管理线程并方便地执行并行任务。了解并合理使用线程池，是一个开发人员必修的基本功。 ...

https://tech.meituan.com/2020/04/02/java-pooling-pratice-in-meituan.html

使用线程池的优点

1. 降低资源消耗：通过池化技术重复利用已创建的线程，降低线程创建和销毁造成的损耗。

2. 提高响应速度：任务到达时，无需等待线程创建即可立即执行。

3. 提高线程的可管理性：线程是稀缺资源，如果无限制创建，不仅会消耗系统资源，还会因为线程的不合理分布导致资源调度失衡，降低系统的稳定性。使用线程池可以进行统一的分配、调优和监控。

4. 提供更多更强大的功能：线程池具备可拓展性，允许开发人员向其中增加更多的功能。比如延时定时线程池ScheduledThreadPoolExecutor，就允许任务延期执行或定期执行。

ThreadPoolExecutor

Executors工具类

newFixedThreadPool

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

newCachedThreadPool

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

1. 核心线程数是 0， 最大线程数是 Integer.MAX_VALUE，救急线程的空闲生存时间是 60s，意味着 全部都是救急线程（60s 后可以回收）,救急线程可以无限创建

2. 队列采用了 SynchronousQueue 实现特点是，它没有容量，没有线程来取是放不进去的（一手交钱、一手交 货）

newSingleThreadExecutor

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

1. Executors.newSingleThreadExecutor() 线程个数始终为1，不能修改.FinalizableDelegatedExecutorService 应用的是装饰器模式，只对外暴露了 ExecutorService 接口，因 此不能调用 ThreadPoolExecutor 中特有的方法

2. Executors.newFixedThreadPool(1) 初始时为1，以后还可以修改 对外暴露的是 ThreadPoolExecutor 对象，可以强转后调用 setCorePoolSize 等方法进行修改

newScheduledThreadPool

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);
}

20:46:09.444 c.TestTimer [main] - start... 
20:46:10.447 c.TestTimer [Timer-0] - task 1 
20:46:12.448 c.TestTimer [Timer-0] - task 2

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);

任务1，执行时间：Thu Jan 03 12:45:17 CST 2019 
任务2，执行时间：Thu Jan 03 12:45:17 CST 2019

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

21:45:43.167 c.TestTimer [main] - start... 
21:45:44.215 c.TestTimer [pool-1-thread-1] - running... 
21:45:45.215 c.TestTimer [pool-1-thread-1] - running... 
21:45:46.215 c.TestTimer [pool-1-thread-1] - running... 
21:45:47.215 c.TestTimer [pool-1-thread-1] - running...

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

21:44:30.311 c.TestTimer [main] - start... 
21:44:31.360 c.TestTimer [pool-1-thread-1] - running... 
21:44:33.361 c.TestTimer [pool-1-thread-1] - running... 
21:44:35.362 c.TestTimer [pool-1-thread-1] - running... 
21:44:37.362 c.TestTimer [pool-1-thread-1] - running...

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

21:40:55.078 c.TestTimer [main] - start... 
21:40:56.140 c.TestTimer [pool-1-thread-1] - running... 
21:40:59.143 c.TestTimer [pool-1-thread-1] - running... 
21:41:02.145 c.TestTimer [pool-1-thread-1] - running... 
21:41:05.147 c.TestTimer [pool-1-thread-1] - running...

延迟执行和周期性执行的原理

private class ScheduledFutureTask<V> extends FutureTask<V> implements RunnableScheduledFuture<V> {
		private final long sequenceNumber;
		private volatile long time;
		private final long period;
		ScheduledFutureTask(Runnable r, V result, long triggerTime,long period, long sequenceNumber) {
				super(r, result);
				this.time = triggerTime; // 延迟时间
				this.period = period; // 周期
				this.sequenceNumber = sequenceNumber;
		}
		// 实现Delayed接口
		public long getDelay(TimeUnit unit) {
				return unit.convert(time - System.nanoTime(), NANOSECONDS);
		}
		// 实现Comparable接口
		public int compareTo(Delayed other) {
				if (other == this) // compare zero if same object
						return 0;
				if (other instanceof ScheduledFutureTask) {
						ScheduledFutureTask<?> x = (ScheduledFutureTask<?>)other;
				long diff = time - x.time;
				if (diff < 0)
						return -1;
				else if (diff > 0)
						return 1;
					// 延迟时间相等，进一步比较序列号
				else if (sequenceNumber < x.sequenceNumber)
						return -1;
				else
					return 1;
			}
			long diff = getDelay(NANOSECONDS) - other.getDelay(NANOSECONDS);
			return (diff < 0) ? -1 : (diff > 0) ? 1 : 0;
}

// 实现Runnable接口
public void run() {
			if (!canRunInCurrentRunState(this))
				cancel(false);
				// 如果不是周期执行，则执行一次
				else if (!isPeriodic())
						super.run();
				// 如果是周期执行，则重新设置下一次运行的时间，重新入队列
				else if (super.runAndReset()) {
					setNextRunTime();
					reExecutePeriodic(outerTask);
				}
			}
// 下一次执行时间
private void setNextRunTime() {
			long p = period;
			if (p > 0)
				time += p;
			else
				time = triggerTime(-p);
			}
}
// 下一次触发时间
long triggerTime(long delay) {
			return System.nanoTime() + ((delay < (Long.MAX_VALUE >> 1)) ? delay : overflowFree(delay));
}
// 放到队列中，等待下一次执行
void reExecutePeriodic(RunnableScheduledFuture<?> task) {
		if (canRunInCurrentRunState(task)) {
					super.getQueue().add(task);
					if (canRunInCurrentRunState(task) || !remove(task)) {
							ensurePrestart();
							return;
					}
		}
		task.cancel(false);
}

最佳实践

合理地创建线程池

多线程如何提升性能

创建多少线程合适？

CPU 密集型运算

I/O 密集型运算

异常处理

1. 主动捕捉异常

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

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())

Fork/Join

开源框架实现的线程池

tomcat

• LimitLatch用来限流,可以控制最大连接个数,类似JUC中的Semaphore

• Acceptor 只负责接收新的socket连接

• Poller 只负责监听 socket channel 是否有 可读的I/O事件

• 一旦可读,封装一个任务对象socketProcessor,提交给Executor线程池处理

• Executor 线程池中的工作线程最终负责处理请求

• 如果总线程数达到maximumPoolSize ,这时不会立刻抛RejectedExecutionException 异常

• 而是再次尝试将任务放入队列,如果还失败,才抛出RejectedExecutionException 异常

自定义线程池实现