Java 线程池的原理及实现

1、线程池简介：

多线程技术主要解决处理器单元内多个线程执行的问题，它可以显著减少处理器单元的闲置时间，增加处理器单元的吞吐能力。

假设一个服务器完成一项任务所需时间为：T1 创建线程时间，T2 在线程中执行任务的时间，T3 销毁线程时间。

如果：T1 + T3 远大于 T2，则可以采用线程池，以提高服务器性能。

一个线程池包括以下四个基本组成部分：

　　1、线程池管理器（ThreadPool）：用于创建并管理线程池，包括 创建线程池，销毁线程池，添加新任务；

　　2、工作线程（PoolWorker）：线程池中线程，在没有任务时处于等待状态，可以循环的执行任务；

　　3、任务接口（Task）：每个任务必须实现的接口，以供工作线程调度任务的执行，它主要规定了任务的入口，任务执行完后的收尾工作，任务的执行状态等；

　　4、任务队列（taskQueue）：用于存放没有处理的任务。提供一种缓冲机制。
                
线程池技术正是关注如何缩短或调整T1，T3时间的技术，从而提高服务器程序性能的。它把T1，T3分别安排在服务器程序的启动和结束的时间段或者一些空闲的时间段，这样在服务器程序处理客户请求时，不会有T1，T3的开销了。线程池不仅调整T1,T3产生的时间段，而且它还显著减少了创建线程的数目，看一个例子：

假设一个服务器一天要处理50000个请求，并且每个请求需要一个单独的线程完成。在线程池中，线程数一般是固定的，所以产生线程总数不会超过线程池中线程的数目，而如果服务器不利用线程池来处理这些请求则线程总数为50000。一般线程池大小是远小于50000。所以利用线程池的服务器程序不会为了创建50000而在处理请求时浪费时间，从而提高效率。

线程池实现代码：

package com.wb.thread;

import java.util.LinkedList;
import java.util.List;

/**
 * 线程池类
 * @author wangbo
 *
 */
public class ThreadPool {
    
    private static int worker_num = 5;//线程池中线程的个数，默认为5
    
    private WorkThread[] workthreads;//工作线程
    
    private static volatile int finished_task = 0;//未处理的任务
    
    private List<Runnable> taskQueue = new LinkedList<Runnable>();//任务队列
    
    private static ThreadPool threadPool;
    
    /**
     * 无参构造器，创建线程池
     */
    private ThreadPool(){
        this(5);
    }
    
    /**
     * 含参构造器，创建线程池
     * @param num
     */
    private ThreadPool(int num){
        worker_num = num;
        workthreads = new WorkThread[num];
        for (int i = 0; i < workthreads.length; i++) {
            workthreads[i] = new WorkThread();
            workthreads[i].start();//开启线程
        }
    }
    
    /**
     * 获得一个默认线程个数的线程池
     * @return
     */
    public static ThreadPool getThreadPool(){
        return getThreadPool(ThreadPool.worker_num);
    }
    
    /**
     * 获得一个指定线程个数的线程池
     * @param num
     * @return
     */
    public static ThreadPool getThreadPool(int num) {
        if (num <= 0) {
            num = ThreadPool.worker_num;
        }
        if (threadPool == null) {
            threadPool = new ThreadPool(num);
        }
        return threadPool;
    }

    /**
     * 将任务单个添加到队列
     * @param task
     */
    public void execute(Runnable task){
        synchronized (taskQueue) {
            taskQueue.add(task);
            taskQueue.notify();
        }
    }
    
    /**
     * 将任务批量添加到队列
     * @param tasks
     */
    public void execute(Runnable[] tasks){
        synchronized (taskQueue) {
            for (Runnable runnable : tasks) {
                taskQueue.add(runnable);
            }
            taskQueue.notify();
        }
    }
    
    /**
     * 将任务批量添加到队列
     * @param tasks
     */
    public void execute(List<Runnable> tasks){
        synchronized (taskQueue) {
            for (Runnable runnable : tasks) {
                taskQueue.add(runnable);
            }
            taskQueue.notify();
        }
    }
    
    /**
     * 销毁线程池
     */
    public void destroy(){
        //还有任务没有执行完
        while(!taskQueue.isEmpty()){
            try {
                Thread.sleep(10);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        //停止工作线程，且置为null
        for (int i = 0; i < workthreads.length; i++) {
            workthreads[i].stopWorker();
            workthreads[i] = null;
        }
        threadPool = null;
        taskQueue.clear();//清空队列
    }
    
    /**
     * 获取工作线程的个数
     * @return
     */
    public int getWorkThreadNumber(){
        return worker_num;
    }
    
    /**
     * 获取已完成任务数量
     * @return
     */
    public int getFinishedTaskNumber(){
        return finished_task;
    }
    
    /**
     * 获取未完成任务数量
     * @return
     */
    public int getWaitTaskNumber(){
        return taskQueue.size();
    }
    
    /**
     * 获取线程池信息
     */
    @Override
    public String toString() {
        return "工作线程数量：" + getWorkThreadNumber() 
                + ",已完成任务数量" + getFinishedTaskNumber()
                + ",未完成任务数量" + getWaitTaskNumber();
                
    }

    /**
     * 内部类，工作线程
     * @author wangbo
     *
     */
    private class WorkThread extends Thread{
        
        private boolean isRunning = true;//线程有效标志

        @Override
        public void run() {
            Runnable runnable = null;
            while (isRunning) {
                synchronized (taskQueue) {
                    //队列为空
                    while (isRunning && taskQueue.isEmpty()) {
                        try {
                            taskQueue.wait(20);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                    //队列不为空
                    if (!taskQueue.isEmpty()) {
                        runnable = taskQueue.remove(0);//去除任务
                    }
                }
                if (runnable != null) {
                    runnable.run();//执行任务
                }
                finished_task++;
                runnable = null;
            }
            
        }
        
        /**
         * 停止线程
         */
        public void stopWorker() {
            isRunning = false;
        }
        
    }

}

测试代码：

package com.wb.thread;

public class ThreadPoolTest {

     public static void main(String[] args) {  
        // 创建3个线程的线程池  
        ThreadPool t = ThreadPool.getThreadPool(3);  
        t.execute(new Runnable[] { new Task(), new Task(), new Task() });  
        t.execute(new Runnable[] { new Task(), new Task(), new Task() });  
        System.out.println(t);  
        t.destroy();//所有线程都执行完成才destory  
        System.out.println(t);  
    }
      
    // 任务类  
    static class Task implements Runnable {
        
        private static volatile int i = 1;
        
        @Override
        public void run() {// 执行任务
            System.out.println("任务 " + (i++) + " 完成");
        }  
    }  

}

2、java类库中提供的线程池简介：

java.util.concurrent包提供了现成的线程池的实现。

示例代码：

package com.wb.thread;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
/**
 * newCachedThreadPool()
 * 线程池为无限大，当执行第二个任务时第一个任务已经完成，会复用执行第一个任务的线程，而不用每次新建线程。
 * 有任务才会创建线程，空闲线程会被保留60s
 * @author wangbo
 *
 */
public class ThreadPoolExecutorTest1 {
    
    public static void main(String[] args) {
        ExecutorService cachedThreadPool = Executors.newCachedThreadPool();
        for (int i = 0; i < 10; i++) {
            final int index = i;
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            cachedThreadPool.execute(new Runnable() {
                @Override
                public void run() {
                    System.out.println(index);
                    System.out.println(Thread.currentThread().getName());
                }
            });
        }
    }

}

package com.wb.thread;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
/**
 * newFixedThreadPool(int nThreads)
 * 创建一个定长线程池，可控制线程最大并发数，超出的线程会在队列中等待。
 * 线程池中包含固定数目的线程，空闲线程会一直保留，参数nThreads表示设定线程池中线程的数目
 * @author wangbo
 *
 */
public class ThreadPoolExecutorTest2 {
    
    public static void main(String[] args) {
        ExecutorService fixedThreadPool = Executors.newFixedThreadPool(2);
        for (int i = 0; i < 10; i++) {
            final int index = i;
            fixedThreadPool.execute(new Runnable() {
                @Override
                public void run() {
                    try {
                        System.out.println(index);
                        System.out.println(Thread.currentThread().getName());
                        Thread.sleep(2000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            });
        }
    }

}

package com.wb.thread;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
/**
 * newScheduledThreadPool(int corePoolSize)
 * 线程池能按时间计划来执行任务，允许用户设定计划执行任务的时间。
 * 参数corePoolSize设定线程池中线程最小数目，当任务较多时，线程池可能会创建更多的工作线程来执行任务。
 * @author wangbo
 *
 */
public class ThreadPoolExecutorTest3 {
    
    public static void main(String[] args) {
        
        method1();
        method2();
        
    }
    
    /**
     * 延迟3s执行
     */
    private static void method1(){
        System.out.println(new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date()));
        ScheduledExecutorService scheduledThreadPool = Executors.newScheduledThreadPool(3);
        scheduledThreadPool.schedule(new Runnable() {
            public void run() {
                System.out.println(new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date()));
                System.out.println("延迟2s执行");
            }
        }, 2, TimeUnit.SECONDS);
    }
    
    /**
     * 延迟2s执行后每3s执行一次
     */
    private static void method2() {
        System.out.println(new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date()));
        ScheduledExecutorService scheduledThreadPool = Executors.newScheduledThreadPool(3);
        scheduledThreadPool.scheduleAtFixedRate(new Runnable() {
            public void run() {
                System.out.println(new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date()));
                System.out.println("延迟2s执行后每3s执行一次");
                System.out.println(Thread.currentThread().getName());
            }
        }, 2, 3, TimeUnit.SECONDS);
    }

}

package com.wb.thread;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
/**
 * newSingleThreadExecutor(int nThreads)
 * 线程池中只有一个线程，它依次执行每个任务。
 * 创建一个单线程化的线程池，它只会用唯一的工作线程来执行任务，保证所有任务按照指定顺序(FIFO, LIFO, 优先级)执行。
 * @author wangbo
 *
 */
public class ThreadPoolExecutorTest4 {
    
    public static void main(String[] args) {
        ExecutorService singleThreadPool = Executors.newSingleThreadExecutor();
        for (int i = 0; i < 10; i++) {
            final int index = i;
            singleThreadPool.execute(new Runnable() {
                @Override
                public void run() {
                    try {
                        System.out.println(index);
                        System.out.println(Thread.currentThread().getName());
                        Thread.sleep(2000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            });
        }
    }

}

package com.wb.thread;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
/**
 * newSingleThreadScheduledExecutor()
 * 线程池中只有一个线程，它能按照时间计划执行每个任务。
 * @author wangbo
 *
 */
public class ThreadPoolExecutorTest5 {
    
    public static void main(String[] args) {
        
        method1();
        method2();
        
    }
    
    /**
     * 延迟3s执行
     */
    private static void method1(){
        System.out.println(new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date()));
        ScheduledExecutorService scheduledThreadPool = Executors.newSingleThreadScheduledExecutor();
        scheduledThreadPool.schedule(new Runnable() {
            public void run() {
                System.out.println(new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date()));
                System.out.println("延迟2s执行");
            }
        }, 2, TimeUnit.SECONDS);
    }
    
    /**
     * 延迟2s执行后每3s执行一次
     */
    private static void method2() {
        System.out.println(new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date()));
        ScheduledExecutorService scheduledThreadPool = Executors.newSingleThreadScheduledExecutor();
        scheduledThreadPool.scheduleAtFixedRate(new Runnable() {
            public void run() {
                System.out.println(new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date()));
                System.out.println("延迟2s执行后每3s执行一次");
                System.out.println(Thread.currentThread().getName());
            }
        }, 2, 3, TimeUnit.SECONDS);
    }

}