JUC包(五) 线程池

前言

在前面的篇章中.我们将解了线程/锁/多线程容器.本章我们将介绍一个用于管理线程的容器:线程池.

---

正文

线程池基本构造如下所示:

基本使用步骤如下:

• 步骤1: 核心线程池是否已满? 未满直接执行,满了放入等待队列BlockedQueue中.
• 步骤2: 等待队列是否已满? 未满放入其中, 满了执行步骤3.
• 步骤3: 线程池是否已满?未满放入其中, 满了执行步骤4.
• 步骤4: 使用创建时指定的策略进行处理.
  其关键的execute()部分代码如下所示:

    public void execute(Runnable command) {
        if (command == null)
            throw new NullPointerException();
        int c = ctl.get();
        // 如果线程数目小于核心线程数目 创建线程完成任务
        if (workerCountOf(c) < corePoolSize) {
            if (addWorker(command, true))
                return;
            c = ctl.get();
        }
        // 放入线程队列
        if (isRunning(c) && workQueue.offer(command)) {
            int recheck = ctl.get();
            if (! isRunning(recheck) && remove(command))
                reject(command);
            else if (workerCountOf(recheck) == 0)
                addWorker(null, false);
        }
        // 
        else if (!addWorker(command, false))
            reject(command);
    }
# Worker类的run方法如下
public void run() {
     runWorker(this);
}
final void runWorker(Worker w) {
        Thread wt = Thread.currentThread();
        Runnable task = w.firstTask;
        w.firstTask = null;
        w.unlock(); // allow interrupts
        boolean completedAbruptly = true;
        try {
            while (task != null || (task = getTask()) != null) {
                w.lock();
                // If pool is stopping, ensure thread is interrupted;
                // if not, ensure thread is not interrupted.  This
                // requires a recheck in second case to deal with
                // shutdownNow race while clearing interrupt
                if ((runStateAtLeast(ctl.get(), STOP) ||
                     (Thread.interrupted() &&
                      runStateAtLeast(ctl.get(), STOP))) &&
                    !wt.isInterrupted())
                    wt.interrupt();
                try {
                    beforeExecute(wt, task);
                    Throwable thrown = null;
                    try {
                        task.run();
                    } catch (RuntimeException x) {
                        thrown = x; throw x;
                    } catch (Error x) {
                        thrown = x; throw x;
                    } catch (Throwable x) {
                        thrown = x; throw new Error(x);
                    } finally {
                        afterExecute(task, thrown);
                    }
                } finally {
                    task = null;
                    w.completedTasks++;
                    w.unlock();
                }
            }
            completedAbruptly = false;
        } finally {
            processWorkerExit(w, completedAbruptly);
        }
    }

---

基本使用

• 初始化
  new ThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue)
corePoolSize: 核心线程池线程数目
  maximumPoolSize: 最大线程数目
  keepAliveTime: 非核心线程的存活时间
  unit: 非核心线程的存活时间单位
  workQueue: 阻塞队列类型.(ArrayBlockingQueue/LinkedBlockingQueue/SynchronousQueue/PriorityBlockingQueue)
  ThreadFactory: 设置线程的创建工厂.例如,线程起名工厂new ThreadFactoryBuilder().setNameFormat("xx-xxx-%d").build()
RejectedExecutionHandler饱和策略: AbortPolicy(抛出异常)/CallerRunsPolicy(调用者所在线程完成任务) / DiscardOldestPolicy(丢弃最近任务) / DiscardPolicy(不处理,直接丢弃)

• 提交任务
  execute(): 无返回值
  submit(): 返回值Future

• 关闭线程池 threadPool.shutdown();

• 总览

/**
 * 9-1 测试线程池的基本使用功能.
 * 
 * */
public class ThreadPoolExecutorTest {
	public static void main(String[] args) throws InterruptedException {
		ThreadPoolExecutor threadPool = new ThreadPoolExecutor(10, 100, 1000, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>());
		
		threadPool.submit(new Runnable(){

			@Override
			public void run() {
				while(true){
					System.out.println(Thread.currentThread().getName()+":"+System.currentTimeMillis());
				}
			}
			
		});
		Thread.sleep(10000);
		threadPool.shutdown();
	}
}

---

Reference

[]