【java线程系列】java线程系列之java线程池详解

标签：

一线程池的概念及为何需要线程池：

我们知道当我们自己创建一个线程时如果该线程执行完任务后就进入死亡状态，这样如果我们需要在次使用一个线程时得重新创建一个线程，但是线程的创建是要付出一定的代价的，如果在我们的程序中需要频繁使用线程，且每个线程执行的时间很短，短到几乎小于线程创建及销毁的时间那么代价将会更大，如：服务器应用程序中经常出现的情况是：单个任务处理的时间很短而请求的数目却是巨大的。显然如果频繁的创建销毁线程效率将非常低。

那么我们能否让一个线程可以复用，即当一个线程执行完后不销毁该线程，而是让其等待执行其它的任务.答案就是使用线程池。

何谓线程池：池线程池的基本思想还是一种对象池的思想，开辟一块内存空间，里面存放了众多（未死亡）的线程，池中线程执行调度由池管理器来处理。当有线程任务时，从池中取一个，执行完成后线程对象归池，这样可以避免反复创建线程对象所带来的性能开销，节省了系统的资源。

合理的使用线程池相对于单独使用线程的好处如下：

1 降低资源消耗，因为线程池减少了创建和销毁线程的次数，每个工作线程都可以被重复利用，可执行多个任务。  

2 提高响应速度。因为线程池中的线程的创建是线程池管理器来创建的，当任务到达时，任务可以不需要等到线程创建就能立即执行。

3提高线程的可管理性，线程池为线程生命周期开销问题和资源不足问题提供了解决方案，使用线程池可以对线程进行统一的分配，调优和监控。

二关于java线程池的几个核心类

说到线程池首先我们得了解三个类：Executors ，ExecutorService与ThreadPoolExecutor。其中Executors 相当于一个创建线程池的工具类，而ExecutorService才是真正的线程池接口，而ThreadPoolExecutor是ExecutorService的具体实现类。我们一个一个来介绍：

1Executors：创建线程池的工具类，在该类中提供了许多静态方法来创建一个线程池，该类中的重要方法如下：

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

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

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

public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
        return new ScheduledThreadPoolExecutor(corePoolSize);
    }

1 public static ExecutorService newFixedThreadPool(int nThreads)

通过传入的int类型整数创建一个固定大小的线程池（Creates a thread pool that reuses a fixed number of threads），每次提交一个任务就创建一个线程，当线程达到线程池的最大大小后提交的线程会在队列中等待。

2 public static ExecutorService newSingleThreadExecutor()

 创建一个单线程的线程池（Creates an Executor that uses a single worker thread）。

3 public static ExecutorService newCachedThreadPool()

创建一个可以缓存的线程池，具体思想是当无线程空闲时创建一个新线程，否则重用先前创建的线程当先前创建的线程空闲时（Creates a thread pool that creates new threads as needed, but will reuse previously constructed threads when they are available.）

4 public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize)

创建一个可以定时执行或周期性执行的线程池（Creates a thread pool that can schedule commands to run after a given delay, or to execute periodically.）

2ExecutorService：可以看到在上述介绍的Executors的几个重要方法的返回值均为ExecutorService，我们先来看一下其类的定义：

public interface ExecutorService extends Executor {

  void shutdown();
  List<Runnable> shutdownNow();
  boolean isShutdown();
  boolean isTerminated();

  boolean awaitTermination(long timeout, TimeUnit unit)
        throws InterruptedException;

  <T> Future<T> submit(Callable<T> task);
  <T> Future<T> submit(Runnable task, T result);

  Future<?> submit(Runnable task);

  <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
        throws InterruptedException;

  <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks,
                                  long timeout, TimeUnit unit)
        throws InterruptedException;

  <T> T invokeAny(Collection<? extends Callable<T>> tasks)
        throws InterruptedException, ExecutionException;

  <T> T invokeAny(Collection<? extends Callable<T>> tasks,
                    long timeout, TimeUnit unit)
        throws InterruptedException, ExecutionException, TimeoutException;
}

可以看到ExecutorService是一个接口它继承自Executor（注意此处不是上面介绍的Executors），那我们来看一下Executor接口的定义：

public interface Executor {

   void execute(Runnable command);
}

3ThreadPoolExecutor类：这个是创建一个线程的核心类，也是我们讲解的重点，首先我们来看一下其类的定义：

public class ThreadPoolExecutor extends AbstractExecutorService 

public abstract class AbstractExecutorService implements ExecutorService

接下来看一下ThreadPoolExecutor类中的构造器。

 public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue) {
        this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
             Executors.defaultThreadFactory(), defaultHandler);
    }

 public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue,
                              ThreadFactory threadFactory) {
        this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
             threadFactory, defaultHandler);
    }

public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue,
                              RejectedExecutionHandler handler) {
        this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
             Executors.defaultThreadFactory(), handler);
    }


 public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue,
                              ThreadFactory threadFactory,
                              RejectedExecutionHandler handler) {
        if (corePoolSize < 0 ||
            maximumPoolSize <= 0 ||
            maximumPoolSize < corePoolSize ||
            keepAliveTime < 0)
            throw new IllegalArgumentException();
        if (workQueue == null || threadFactory == null || handler == null)
            throw new NullPointerException();
        this.corePoolSize = corePoolSize;
        this.maximumPoolSize = maximumPoolSize;
        this.workQueue = workQueue;
        this.keepAliveTime = unit.toNanos(keepAliveTime);
        this.threadFactory = threadFactory;
        this.handler = handler;
    }

可以看到ThreadPoolExecutor类为我们提供了四个构造器，其中第四个是最基本的构造器，其余三个构造器均在其方法内调用了第四个构造器。所以我们重点讲解第四个构造器的各参数的意义：

1 int corePoolSize：内核池的大小，是一个int型的参数，

2 int maximumPoolSize：线程池的最大线程数，是一个int型的参数，它表示在线程池中最多能创建多少个线程（the maximum number of threads to allow in the  pool）

3 long keepAliveTime：表示无任务执行时线程最多维持多久后终止，是一个long类型的参数（this is the maximum time that excess idle threads will wait for new tasks before terminating.），只有当线程池中的线程数大于corePoolSize时，keepAliveTime才会起作用（when the number of threads is greater than  the core），直到线程池中的线程数不大于corePoolSize，即当线程池中的线程数大于corePoolSize时，如果一个线程空闲的时间达到keepAliveTime，则会终止，直到线程池中的线程数不超过corePoolSize。

4 TimeUnit unit：参数keepAliveTime的时间单位（the time unit for the {@code keepAliveTime} argument）

5  BlockingQueue<Runnable> workQueue：阻塞队列，用来存储等待执行的任务（the queue to use for holding tasks before they are executed），这个队列仅仅容纳通过execute方法提交的Runnable接口的任务（the queue to use for holding tasks before they are executed.  This queue will hold only the {@code Runnable}  tasks submitted by the {@code execute} method.）

6  ThreadFactory threadFactory：线程工厂，主要用来创建线程（the factory to use when the executor  creates a new thread）

7  RejectedExecutionHandler handler：它表示当一个线程的执行因为到达现场边界且队列容量达到极值而阻塞时（to use when execution is blocked because the thread           bounds and queue capacities are reached）而拒绝执行任务（ RejectedExecution）时应该采取的策略。它的取值是一个 RejectedExecutionHandler 接口，取值供四种情    况：

ThreadPoolExecutor.AbortPolicy:丢弃任务且抛出RejectedExecutionException异常。 
ThreadPoolExecutor.DiscardPolicy：也是丢弃任务，但是不抛出异常。 
ThreadPoolExecutor.DiscardOldestPolicy：丢弃队列最前面的任务，然后重新尝试执行任务（重复此过程）
ThreadPoolExecutor.CallerRunsPolicy：由调用线程处理该任务

其中上述四个类都是ThreadPoolExecutor的静态内部类，它们均实现了 RejectedExecutionHandler 接口，其类的定义如下：

 public static class AbortPolicy implements RejectedExecutionHandler 

public static class DiscardPolicy implements RejectedExecutionHandler

public static class DiscardOldestPolicy implements RejectedExecutionHandler

public static class CallerRunsPolicy implements RejectedExecutionHandler 

接下来看一下ThreadPoolExecutor类的重要方法：

public void execute(Runnable command) 
public void shutdown()
public List<Runnable> shutdownNow()
submit()

注意execute()方法实际上是Executor中声明的方法，在ThreadPoolExecutor进行了具体的实现，这个方法是ThreadPoolExecutor的核心方法，通过这个方法可以向线程池提交一个任务，交给线程池去执行。

三java线程池的使用：

使用线程池时我们通常不是使用ThreadPoolExecutor这个核心类，而是使用Executors这个工具类中的几个静态方法，

Executors.newCachedThreadPool();            //创建一个缓冲池，缓冲池容量大小为Integer.MAX_VALUE
Executors.newSingleThreadExecutor();       //创建容量为1的缓冲池
Executors.newFixedThreadPool(int);          //创建固定大小的线程池
Executors.newScheduledThreadPool(int)      //创建一个定时执行的线程池

public class Server {
	private ExecutorService executorService;// 线程池
	private ServerSocket serverSocket = null;
	private Socket socket = null;
	private boolean isStarted = true;

	public Server() {
		try {
			// 创建线程池，池中具有(cpu个数*50)条线程
			executorService = Executors.newFixedThreadPool(Runtime.getRuntime()
					.availableProcessors() * 50);
			serverSocket = new ServerSocket(Constants.SERVER_PORT);
		} catch (IOException e) {
			e.printStackTrace();
			quit();
		}
	} 

	public void start() {
		System.out.println(MyDate.getDateCN() + " 服务器已启动...");
		try {
			while (isStarted) {//将服务器端的accept操作放在一个while循环中，用来不断监测客户端的连接请求
				socket = serverSocket.accept();
				String ip = socket.getInetAddress().toString();
				System.out.println(MyDate.getDateCN() + " 用户：" + ip + " 已建立连接");
				// 为支持多用户并发访问，采用线程池管理每一个用户的连接请求
				//即将每个用户的请求单独放到一个线程中执行
				if (socket.isConnected())
					executorService.execute(new SocketTask(socket));// 添加到线程池
			}
			if (socket != null)
				socket.close();
			if (serverSocket != null)
				serverSocket.close();
		} catch (IOException e) {
			e.printStackTrace();
			// isStarted = false;
		}
	}

	private final class SocketTask implements Runnable {
		private Socket socket = null;
		private InputThread in;
		private OutputThread out;
		private OutputThreadMap map;

		public SocketTask(Socket socket) {
			this.socket = socket;
			map = OutputThreadMap.getInstance();
		}

		@Override
		public void run() {
			out = new OutputThread(socket, map);//
			// 先实例化写消息线程,（把对应用户的写线程存入map缓存器中）
			in = new InputThread(socket, out, map);// 再实例化读消息线程
			out.setStart(true);
			in.setStart(true);
			in.start();
			out.start();
			
			
		}
	}

	/**
	 * 退出
	 */
	public void quit() {
		try {
			this.isStarted = false;
			serverSocket.close();
		} catch (IOException e) {
			e.printStackTrace();
		}
	}

	public static void main(String[] args) {
		new Server().start();
	}
}

1使用Executors这个工具类中的几个静态方法创建一个ThreadPoolExecutor对象，如

executorService = Executors.newFixedThreadPool(Runtime.getRuntime()
					.availableProcessors() * 50);// 创建线程池，池中具有(cpu个数*50)条线程

2创建一个实现了Runnable接口的线程，如：

private final class SocketTask implements Runnable

重写其run方法，在run方法中完成自己的业务逻辑。

executorService.execute(new SocketTask(socket));// 添加到线程池

注意execute()方法实际上是Executor中声明的方法，在ThreadPoolExecutor进行了具体的实现。

好了以上就是本人理解的关于java线程池的内容，看官如果觉得不错，请记得点击下方的”顶“或赞给我一点鼓励哦！

【java线程系列】java线程系列之java线程池详解

标签：

原文地址：http://blog.csdn.net/htq__/article/details/51250763