详细讲解线程池的使用

      前言：说起threadpoolexector应该大家多少都接触过,现在我详细的讲解下其的用法

一：解析参数

为了更好地理解threadpoolexecutor,我先讲一个例子,话说一个工作多年的高T,一天突然决定自己要单干组织一个团队,经过仔细的考虑他做出了如下的决定

1、团队的核心人员为10个

2、如果一旦出现项目过多人员不足的时候,则会聘请5个外包人员

3、接的项目单子最多堆积100

4、如果项目做完了团队比较空闲,则裁掉外包人员,空闲的时间为一个月

5、如果接的单子超过100个,则后续考虑一些兜底策略(比如拒绝多余的单子,或者把多出100个以外的单子直接交付第三方公司做)

6、同时他还考虑了如果效益一直不好,那么就裁掉所有人,宣布公司直接倒闭

上面的例子恰恰和我们的线程池非常的像,我们来看下threadpoolexecutor的定义。

ThreadPoolExecutor poolExecutor = new ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, RejectedExecutionHandler handler);

corePoolSize：核心线程数(就是我们上面说的一个公司核心人员)

maximumPoolSize：最大线程数(就是我们说的一旦公司接收的单子过多则聘请外包,此时也是公司最大的人员了,因为人多了办公地方不够了)

keepAliveTime：超多核心线程数之外线程的存活时间(就是如果公司一旦活不多要多久进行裁掉外包人员)

unit：上面时间的单元(可以年月日时分秒等)

workQueue：任务队列(就是如果公司最大能存的单子)

handler：拒绝策略(就是一旦任务满了应该如果处理多余的单子)

allowCoreThreadTimeOut：设置是否清理核心线程(如果设置true,如果任务少于实际执行的线程则会清理核心线程,默认为false)

二：实际演练

先验证核心线程数

RejectedExecutionHandler handler = new RejectedExecutionHandlerImpl();
ThreadPoolExecutor poolExecutor = new ThreadPoolExecutor(2, 5, 20L, TimeUnit.MILLISECONDS, new ArrayBlockingQueue<Runnable>(5), handler);
for (int i = 0; i < 12; i++) {
AppTask appTask = new AppTask(i);
poolExecutor.execute(appTask);
System.out.println("线程池中线程的数目:" + poolExecutor.getPoolSize() + ",线程池中等待的队列数目：" + poolExecutor.getQueue().size() + ";线程池中已执行完毕的任务数据:" + poolExecutor.getCompletedTaskCount());
}
MainThread.exec();
poolExecutor.allowCoreThreadTimeOut(true);
if (!poolExecutor.isShutdown()) {
poolExecutor.shutdown();
}
static LinkedBlockingQueue<Runnable> blockingQueue = new LinkedBlockingQueue(10000);
static class RejectedExecutionHandlerImpl implements RejectedExecutionHandler {
@Override
public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
try {
blockingQueue.put(r);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}

static int count=0;
static class MainThread {
public static void exec() throws InterruptedException {
while (true) {
if (blockingQueue.size() >= 8000) {
System.out.println("报警");
}
Runnable r = blockingQueue.poll();
if (r == null) {
Thread.sleep(1000);
continue;
}
count++;
r.run();
System.out.println("非堵塞线程池执行的个数:"+count);
}
}
}

执行结果

 有人问和上面有什么区别呢,其实区别就是这个不会出现堵塞,这里我虽然采用堵塞队列来存储,为了更好的展示,其实这里你可以打印日志,或者存入redis中然后进行处理。

优点：

1、也不会造成内存溢出

2、消费不会出现堵塞

缺点：

这样设计明显会复杂的很多,而且获取消息值不易,就我目前来看我更倾向于采用堵塞线程池

五：线程池的设计

线程池的设计最重要的一点就是计算出生产者最大的qps量和单个线程消费的能力,比如我们生产者qps是1万,但是我们单个线程处理每个任务的时间是2毫秒,如果我们cpu是4核,那么我们核心线程是4*2=8个,所以我们每秒处理的任务数是1000/2*8=4000,很显然我们的消费能力远远不足,这个时候我们应该考虑采用多台机器处理,有人不是可以堵塞队列么,其实那是一种兜底策略,避免消息丢失,但这并不是我们设计的核心。如果我们能计算出单条消息的大小(如1k)我们分配这个消息服务的内存是300M,那么我们可以做个折中150M来存储多余的消息,那么可以存储的量是1百万,如果我们的流量高峰是30分钟,每秒处理剩余的消息是200,那么这半小时之内总共剩余的消息总量是30*60*200=360000,这样一来完全可以满足我们的业务需求。

六：线程池的有点

1、减少频繁的创建和销毁线程(由于线程创建和销毁都会耗用一定的内存)

2、线程池也是多线程,充分利用CPU,提高系统的效率

七：Executors下的4种创建线程池的方式

1、newSingleThreadExecutor();

这个线程池的特点是核心线程数只有一个,最大线程数也只有一个,采用的队列是无界队列,可能会导致内存溢出

2、newFixedThreadPool(5)

这个线程池特点是核心线程数由自己设置并且一旦设置就是固定的不在改变,采用的队列是无界队列,可能会导致内存溢出

3、newCachedThreadPool()

这个线程池特点是核心线程数为0,最大线程数无界,这样也会造成内存溢出

4、newScheduledThreadPool(5)

这个线程池特点是可以周期性执行任务,核心线程数需要自己进行初始化,最大线程数无界,这样也会造成内存溢出

我们在实际操作中,尽量避免使用Executors来创建多线程,因为如果消息量过大会导致内存溢出,消息丢失