Java多线程技术研究（三）-线程池-蒲公英云

在多线程开发中，当线程数量较多时，并且每个线程的执行时间较短，因而需要频繁的创建线程和销毁线程，这样会大大较低系统的吞吐能力。这时就可以采用线程技术，实现线程执行完成后不会被销毁，可以被反复使用。假设一个服务器完成一项任务所需要的时间为T1，创建一个线程的时间为T2，销毁一个线程的时间为T3。当((T2+T3))/T1 的值较大时，才用线程池的技术就可以很好的提高服务器性能。

线程池顾名思义指的就是线程的池子，初始时，我们创建一定数量的线程放置在线程池，每当一个新的任务提交时，我们就可以从池子中取出一个空闲的线程去执行该任务，执行完后才能后，线程不会被销毁，被重现放回线程池的空闲队列中。

线程池技术正是关注如何缩短或调整线程的创建时间或销毁线程时间的技术。我们正常情况下，会在启动应用程序时，创建线程池，而退出应用时，销毁线程池，因而在应用处理请求过程中只有执行服务的时间开销。注意：需要根据系统的承受能力，调整线程池中工作线程的数目，防止因为消耗过多的内存，而把服务器累趴下。

1、在Java类库中提供了线程池的机制，常见的以下几种线程池：
1) newScheduledThreadPool
创建一个定长线程池，支持定时及周期性任务执行，类似于Timer

package com.wygu.thread.study;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class MultiThreadPool { 
    public static void main(String []argv){
        TaskRunnable taskRunnableA = new TaskRunnable("ThreadA");
        TaskRunnable taskRunnableB = new TaskRunnable("ThreadB");
        ExecutorService executorService = Executors.newScheduledThreadPool(2);
        executorService.execute(taskRunnableA);
        executorService.execute(taskRunnableB);
    }
}
class TaskRunnable implements Runnable{ 
    private String threadNm;
    public TaskRunnable(String threadNm) {
        this.threadNm = threadNm;
    }
    @Override
    public void run() {
        for(int i=0;i<5;i++){
            System.out.println(this.threadNm+"---->"+i);
            try {
                Thread.sleep(10);
            } catch (InterruptedException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
        }
    }
}

程序运行结果为：
ThreadB—->0
ThreadA—->0
ThreadA—->1
ThreadB—->1
ThreadA—->2
ThreadB—->2
ThreadA—->3
ThreadB—->3
ThreadA—->4
ThreadB—->4
2) newCachedThreadPool
创建一个可缓存的线程池，如果当前线程池的规模超出了处理需求，将回收空闲的线程；当需求增加时，会增加工作线程数量；线程池规模无限制；空闲线程被保留60秒。
3) newFixedThreadPool
创建一个固定长度的线程池，当到达线程最大数量时，线程池的规模将不再变化。
4) newSingleThreadExecutor
线程池中任意时刻只有一个线程，队列中的任务按照顺序执行。

2、工作中的应用
在后台交易系统中，每天都会产生大量的流水日志，如果系统实现实现日志数据清理的机制功能，会导致日志表中的数据日积月累，越来越多，严重影响到系统的性能。笔者刚工作时，接触到的第一个系统就是这种情况，交易日志表中存储接近上千万的数据，严重影响了交易的响应速度。为解决这个问题，往往会创建两张交易日志表，实现按日轮换使用，利用定时任务清理即将使用的表中的记录。

首先介绍ServletContextListener机制：
（1）接口ServletContextListener监听 ServletContext对象的生命周期。
（2）当Servlet容器启动或终止Web应用时，就会触发ServletContextEvent事件，该事件由ServletContextListener来处理。
（3）在 ServletContextListener接口中定义了处理ServletContextEvent事件的两个方法contextInitialized(),contextDestroyed()。
首先在web.xml中配置

<listener>
    <listener-class> com.unionpay.cloudPayment.util.TimedTaskClearUtil</listener-class>
  </listener>

下面创建类TimedTaskClearUtil，并继承ServletContextListener，重写方法contextInitialized()，contextDestroyed()：

import java.sql.Connection;
import java.sql.SQLException;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Calendar;
import java.util.Date;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import javax.servlet.ServletContextEvent;
import javax.servlet.ServletContextListener;
public class TimedTaskClearUtil implements ServletContextListener{  
    private static  ScheduledThreadPoolExecutor service;
    public void contextInitialized(ServletContextEvent sce) {
        service = new ScheduledThreadPoolExecutor(2); //初始化线程池中线程数为2
        service.setExecuteExistingDelayedTasksAfterShutdownPolicy(false);//当容器停止时，任务也将立刻停止
        service.setContinueExistingPeriodicTasksAfterShutdownPolicy(false);
    }
    public void contextDestroyed(ServletContextEvent sce) {
        try {
            if(!service.isShutdown()){
                service.shutdownNow();
                System.out.println("*****************定时任务：日志表清理销毁成功******************");
            }
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
    }
    public static void execuTimedTask(){
        final SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMddHHmmss");// 设置日期格式
        String curDateStr = sdf.format(new Date());
        String nextTime = String.format("%s%s", curDateStr.substring(0, 8),"234000");
        String cycleUnit = "H";  //H表示以小时为单位，M表示以分钟为单位，S表示以秒为单位
        int cycleSpan = 24;      //表示24 cycleUnit
        long secondDiff = 0;
        try {
            Date curDate = sdf.parse(curDateStr);
            Date nextDate=  sdf.parse(nextTime);
            secondDiff=(nextDate.getTime()-curDate.getTime())/1000; //距离第一次任务的执行还需要多长时间
        } catch (ParseException e) {
            e.printStackTrace();
        }
        Runnable runnable = new Runnable() {  
            public void run() { 
                    /*执行日志表的清理 **清理时建议使用truncate操作，减少磁盘碎片 */
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }; 
        service.scheduleAtFixedRate(runnable, secondDiff, revToSecond(cycleUnit，cycleSpan), TimeUnit.SECONDS);//定时任务按照固定的频率执行
    } 
    private static long revToSecond(String cycleUnit,int cycleSpan){
        long timeSpan = 0;
        if("H".equals(cycleUnit)){
            timeSpan = cycleSpan*3600;
        }else if("M".equals(cycleUnit)){
            timeSpan = cycleSpan*60;
        }else if("S".equals(cycleUnit)){
            timeSpan = cycleSpan;
        }
        return timeSpan;
    }
}

在启动servlet中，可以直接加载定时任务，比如在InitServlet中，调用：

TimedTaskClearUtil.execuTimedTask();

线程池除了可以直接使用Java Api提供的类创建外，也可以自己创建。
2、自定义线程池
一个线程池包括以下四个基本组成部分：
（1）线程池管理器（ThreadPool）：用于创建并管理线程池，包括创建线程池，销毁线程池，添加新任务；
（2）工作线程（PoolWorker）：线程池中线程，在没有任务时处于等待状态，可以循环式执行任务；
（3）任务接口（Task）：每个任务必须实现的接口，以供工作线程调度任务的执行，它主要规定了任务的入口，任务执行完后的收尾工作，任务的执行状态等；
（4）任务队列（taskQueue）：用于存放没有处理的任务，提供一种缓冲机制。

自定义线程池代码：参考博客：http://www.cnblogs.com/kinghitomi/archive/2012/01/19/2327418.html
创建线程管理器

package com.wygu.thread.study;
import java.util.concurrent.LinkedBlockingQueue;
/** * * @author guweiyu *@category 一个完整的线程至少需要包含以下四个部分：线程池管理器（ThreadPoll），工作线程（WorkThread），任务接口（Task），任务队列（TaskQueue） */
//线程池管理器
public final class ThreadPoolExecutOwn { 
    // 设置线程池中默认线程数为5
    private static int worker_thread_num = 5;
    private static int default_thread_num = 5;
    // 工作线程
    private WorkThread[] workThread;
    // 未处理的任务
    public static volatile int finished_task = 0;
    // 任务队列，作为一个缓冲
    public static LinkedBlockingQueue<Runnable> taskQueue = new LinkedBlockingQueue<Runnable>();
    private static ThreadPoolExecutOwn threadPool;
    // 创建具有默认线程个数的线程池
    private ThreadPoolExecutOwn() {
        this(default_thread_num);
    }
    // 创建线程池中的线程,worker_thread_num为线程池中工作线程的个数
    private ThreadPoolExecutOwn(int worker_thread_num) {
        ThreadPoolExecutOwn.worker_thread_num = worker_thread_num;
        workThread = new WorkThread[worker_thread_num];
        for (int i = 0; i < worker_thread_num; i++) {
            workThread[i] = new WorkThread();
            workThread[i].start();
        }
    }
    // 获得默认值default_thread_num的线程池
    public static ThreadPoolExecutOwn getThreadPool() {
        return getThreadPool(ThreadPoolExecutOwn.default_thread_num);
    }
    // 获得一个指定线程个数的线程池,worker_num(>0)为线程池中工作线程的个数
    public static ThreadPoolExecutOwn getThreadPool(int worker_num) {
        if (worker_num <= 0)
            worker_num = ThreadPoolExecutOwn.default_thread_num;
        if (threadPool == null)
            threadPool = new ThreadPoolExecutOwn(worker_num);
        return threadPool;
    }
    // 执行任务,将任务加入到任务队列中，等待线程池管理器调用
    public void execute(Runnable task) {
        synchronized (ThreadPoolExecutOwn.taskQueue) {
            ThreadPoolExecutOwn.taskQueue.add(task);
            ThreadPoolExecutOwn.taskQueue.notify();
        }
    }
    // 销毁线程池,方法保证在所有任务都完成的情况下才会销毁所有线程，否则等待任务完成才销毁
    public void destroy() {
        while (!ThreadPoolExecutOwn.taskQueue.isEmpty()) {
            try {
                Thread.sleep(10);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        // 工作线程停止工作，且置为null
        for (int i = 0; i < worker_thread_num; i++) {
            workThread[i].stopWorker();
            workThread[i] = null;
        }
        threadPool=null;
        ThreadPoolExecutOwn.taskQueue.clear();// 清空任务队列
    }
    // 返回工作线程的个数
    public int getWorkThreadNumber() {
        return worker_thread_num;
    }
    // 返回已完成任务的个数,这里的已完成是只出了任务队列的任务个数，可能该任务并没有实际执行完成
    public int getFinishedTasknumber() {
        return finished_task;
    }
    // 返回任务队列的长度，即还没处理的任务个数
    public int getWaitTasknumber() {
        return ThreadPoolExecutOwn.taskQueue.size();
    }
}

创建工作线程

package com.wygu.thread.study;
/* * @author guweiyu */
public class WorkThread extends Thread { 
    private boolean isRunning = true;  //用于判断工作线程是否有效，能否继续执行
    //如果任务队列为空，则工作线程等待，否则从队列中取出任务继续执行
    @Override
    public void run() {
        Runnable r = null;
        while (isRunning) {
            synchronized (ThreadPoolExecutOwn.taskQueue) {
                while (isRunning && ThreadPoolExecutOwn.taskQueue.isEmpty()) {
   // 队列为空
                    try {
                        ThreadPoolExecutOwn.taskQueue.wait(20);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                if (!ThreadPoolExecutOwn.taskQueue.isEmpty())
                    r = ThreadPoolExecutOwn.taskQueue.remove();// 取出任务
            }
            if (r != null) {
                r.run();// 执行任务
            }
            ThreadPoolExecutOwn.finished_task++;
            r = null;
        }
    }
    // 停止工作，让该线程自然执行完run方法，自然结束
    public void stopWorker() {
        isRunning = false;
    }
}

创建工作任务

package com.wygu.thread.study;
public class WorkTask implements Runnable{ 
    private String taskName = null;
    public WorkTask(String taskName) {
        this.taskName = taskName;
    }
    @Override
    public void run() {
        System.out.println("任务开始执行："+this.taskName);
        try {
            Thread.sleep(100);
        } catch (InterruptedException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
        System.out.println("任务完成："+this.taskName);
    }
}

Main方法调用

package com.wygu.thread.study;
public class Main {
    public static void main(String[] args) {
        ThreadPoolExecutOwn threadPoolExecutOwn = ThreadPoolExecutOwn.getThreadPool(3);//创建线程数为4的线程池
        WorkTask []workTask = new WorkTask[100];
        for(int i=0;i<10;i++){
            workTask[i] = new WorkTask("TASK"+i);
            threadPoolExecutOwn.execute(workTask[i]);
        }
    }
}

程序运行结果为
TASK0:开始执行**
TASK2:开始执行*
TASK1:开始执行*
TASK0:执行结束###
TASK2:执行结束###
TASK1:执行结束###
TASK4:开始执行*
TASK3:开始执行*
TASK5:开始执行*
TASK4:执行结束###
TASK5:执行结束###
TASK3:执行结束###
TASK6:开始执行*
TASK8:开始执行*
TASK7:开始执行*
TASK6:执行结束###
TASK8:执行结束###
TASK9:开始执行**
TASK7:执行结束###
TASK9:执行结束###