线程池

wdt1988520

浏览: 15513 次

最近访客更多访客>>

woodding2008

zhufengxiang

博主相关

博客

微博

相册

留言

关于我

文章分类

社区版块

存档分类

博客分类：

Java基础知识

线程池在java中非常重要，在java se1.4以前的版本线程池非常简单，但在1.5以后新增了java.util.concurrent多线包，使线程池变的非常强大！再和java se1.4新增的NIO可以实现非常高性能的功能。

为什么使用线程池：

1、减少线程的创建销毁所到来的资源消耗，线程池可以使线程得到复用，并且易于维护管理。

2、可更具系统资源消耗（如、cpu、rom、IO）动态配置线程的使用，防止系统消耗过大导致系统奔溃。

线程池的分类：

一、同步线程池

工作队列选用的SynchronousQueue,SynchronousQueue(一进一出)它将任务交给它的线程并不保存任务，如果有新的任务加入线程池就会创建新的线程并立即执行。

Executors.newCachedThreadPool() 的实现就是SynchronousQueue完成的。

二、无界线程池

工作队列选用的LinkedBlockingQueue.

1、当线程池的线程数未达到corePoolSize时则创建新的线程执行任务，直到到达饱和线程值 corePoolSize

2、当达到corePoolSize时，新来的任务请求直接加入到无界队列中，如果有无限的任务，直到系统被撑爆才停止。

3、由于要队列中不能添加新任务，线程池才会新建线程直到达到最大线程值maximumPoolSize ，由于这里是无界队列，所以这个最大线程值maximumPoolSize永远达不到！

Executors.newFixedThreadPool(10);Executors.newSingleThreadExecutor();就采用的这种实现方式.

三、有界线程池

工作队列选用的ArrayBlockingQueue.

线程池的饱和值在maximumPoolSize运行，在最大线程值运行时，新加入的任务采用4种策略模式处理。

有界队列要依靠系统的配置参数作为依据设置有界队列的大小。

1、当队列太小，池线程数太大，这样会增加CPU资源的开销，这样线程调度的开销（上下文切换）也会增加，从而降低吞吐量。

2、当队列太大，池线程数太小，这样虽可以减小CPU资源的消耗和线程上下文（系统到用户空间）切换带来的开销，但吞吐量也不会高。

有界队列线程池java se中没有实现，我们可以自己更具系统资源来实现自己的线程池。

执行器的五种状态：

1、RUNNING 在ThreadPoolExecutor被实例化时就是这个状态

2、SHUTDOWN 这一方法是不在接受新的任务，等待池中和队列中的任务完成后关闭，

ThreadPoolExecutor的awaitTermination()阻塞等待shutdown请求后所有线程终止，会有时间参数，超时和中断也会令方法调用结束。

3、STOP状态执行的是ShutDownNow方法，不在执行池中和队列中的任务，并试图终止正在执行的任务

4、TIDYING线程池为空，就会处于这个状态，执行terminated方法

5、TERMINATED terminated()执行完毕，就会到这个状态，此时ThreadPoolExecutor终结

例：有界队列的四种策略方式

package threadpool;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 *create by datao.wang 2014-2-12-下午3:08:47	
 *有界队列线程池
 *ThreadPoolExecutor 是ExecutorService线程接口的实现
 *ThreadPoolExecutor poolExecutor=new ThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory , rejectedExecutionHandler);
 *	corePoolSize 是线程池的核心线程数，通常线程池会维持这个线程数
 *	maximumPoolSize 是线程池所能维持的最大线程数
 *	keepAliveTime 和 unit 则分别是超额线程的空闲存活时间数和时间单位
 *	workQueue 是提交任务到线程池的入队列
 *	threadFactory 是线程池创建新线程的线程构造器
 *	rejectedExecutionHandler 是当线程池不能接受提交任务的时候的处理策略
 *
 */
public class ArrayBlockingThreadPoolExecutor {
	//有界队列 队列大小为10
	ArrayBlockingQueue<Runnable> jobs = new ArrayBlockingQueue<Runnable>(10);
	/**
	 * create by datao.wang  2014-2-12
	 * 这是ThreadPoolExecutor默认的一个策略
	 * 作用：当线程池到达最高线程数6，并且有界队列已满，线程池抛出rejectedExecutorException异常
	 */
	public void policy_Abort(){
		RejectedExecutionHandler rejectedExecutionHandler = new ThreadPoolExecutor.AbortPolicy();
		ThreadPoolExecutor poolExecutor = new ThreadPoolExecutor(2, 6, 30, TimeUnit.SECONDS,jobs,rejectedExecutionHandler );
		runWorker(poolExecutor);
	}
	
	/**
	 * create by datao.wang  2014-2-12
	 * 丢弃策略
	 * 当线程池到达最高线程数6，并且有界队列已满，新加入任务将被丢弃
	 */
	public void policy_Discard(){
		RejectedExecutionHandler rejectedExecutionHandler = new ThreadPoolExecutor.DiscardPolicy();
		ThreadPoolExecutor poolExecutor = new ThreadPoolExecutor(2, 6, 30, TimeUnit.SECONDS,jobs,rejectedExecutionHandler );
		runWorker(poolExecutor);
	}

	/**
	 * create by datao.wang  2014-2-12
	 * 丢弃最旧的任务策略
	 * 当线程池到达最高线程数6，并且有界队列已满，新来的任务将队列最前端的任务踢掉。
	 */
	public void policy_DiscardOldest(){
		RejectedExecutionHandler rejectedExecutionHandler = new ThreadPoolExecutor.DiscardOldestPolicy();
		ThreadPoolExecutor poolExecutor = new ThreadPoolExecutor(2, 6, 30, TimeUnit.SECONDS,jobs,rejectedExecutionHandler );
		runWorker(poolExecutor);
	}
	
	/**
	 * create by datao.wang  2014-2-12
	 * Caller线程运行策略，任务发起者现在来执行任务策略
	 * 当线程池到达最高线程数6，并且有界队列已满，此时将由任务发起者来执行新来的任务
	 */
	public void policy_CallerRuns(){
		RejectedExecutionHandler rejectedExecutionHandler = new ThreadPoolExecutor.CallerRunsPolicy();
		ThreadPoolExecutor poolExecutor = new ThreadPoolExecutor(2, 6, 30, TimeUnit.SECONDS,jobs,rejectedExecutionHandler );
		runWorker(poolExecutor);
	}
	
	public void runWorker(ThreadPoolExecutor poolExecutor){
		int numJobs = 20;
		System.out.println("Starting application to add "+ numJobs + " jobs");
		for (int i=1; i<= numJobs; i++){
			try {
				WorkerThread job=new WorkerThread(i);
				poolExecutor.submit(job);
			//	 System.out.println("Added job #" + (i));
			}catch (RejectedExecutionException e) {
//				e.printStackTrace();
		      System.err.println("RejectedExecutionException job #" + (i));
		    }
		}
	}
	
	public static void main(String[] args) {
		ArrayBlockingThreadPoolExecutor executor=new ArrayBlockingThreadPoolExecutor();
//		executor.policy_Abort();//运行结果当运行到6个任务时将抛出RejectExecutorException异常，后面的任务将继续执行，导致结构有一部分任务未被执行。
//		executor.policy_Discard();//运行结果当运行到6个任务时将放弃后面的几个任务，然后后面的任务将继续执行。
//		executor.policy_DiscardOldest();//运行结果当运行到6个任务是放弃队列最前端的任务，然后后面任务继续执行。
		executor.policy_CallerRuns();//运行结果当运行到6个任务时队列已满，后面新加的任务将由任务提交者线程来执行，这样将减缓任务的提交，结果所以线程都得到执行。
	}
}


/***
 * 工作线程
 * @author wdt
 *
 */
class WorkerThread implements Runnable {
	int jobid;
	
	public WorkerThread (int jobid){
		this.jobid = jobid;
	}
	
	public void run (){
		try{
			System.out.println("exec job" + (jobid)+"---"+(jobid*500+1000));
			//线程睡眠时间
			Thread.sleep(jobid*500 + 1000);
		}catch (Exception excep){
			
		}
	}
}

例：无界队列

package threadpool;

import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 *create by datao.wang 2014-2-13-上午12:02:17	
 *无界队列线程池
 *ThreadPoolExecutor 是ExecutorService线程接口的实现
 *ThreadPoolExecutor poolExecutor=new ThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory , rejectedExecutionHandler);
 *无界队列的特点：
 * 1、当线程池的线程数未达到corePoolSize时则创建新的线程执行任务，直到到达饱和线程值corePoolSize
 * 2、当达到corePoolSize时，新来的任务请求直接加入到无界队列中，如果有无限的任务，直到系统被撑爆才停止。
 * 3、由于要队列中不能添加新任务，线程池才会新建线程直到达到最大线程值maximumPoolSize，由于这里是无界队列，所以这个最大线程值maximumPoolSize永远达不到！
 */
public class LinkedBlockingThreadPoolExecutor {

	public static void main(String[] args) {
		LinkedBlockingQueue<Runnable> jobs = new LinkedBlockingQueue<Runnable>();
		ThreadPoolExecutor poolExecutor = new ThreadPoolExecutor(2, 6, 30, TimeUnit.SECONDS,jobs);
	}
}

例：同步队列

package threadpool;

import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 *create by datao.wang 2014-2-13-上午12:21:49	
 *同步队列线程池  （一进一出）
 *同步队列自身的特性：
 *  1、size永远为0
 *  2、当向队列中放入元素（执行put方法）时，队列就发生阻塞，直到有线程调用take方法取走元素。
 *  
 */
public class SynchronousQueueThreadPoolExecutor {
	
	public static void main(String[] args) {
		SynchronousQueue<Runnable> jobs = new SynchronousQueue<Runnable>();
		//同步队列也执行策略模式
		RejectedExecutionHandler rejectedExecutionHandler = new ThreadPoolExecutor.CallerRunsPolicy();
		ThreadPoolExecutor poolExecutor = new ThreadPoolExecutor(2, 6, 30, TimeUnit.SECONDS,jobs,rejectedExecutionHandler );
	}
}

例：java se中已实现线程池。

package threadpool;

import java.util.concurrent.Executors;

/**
 *create by datao.wang 2014-2-13-上午11:36:48	
 */
public class ExecutorTool {
	
   public static void main(String[] args) {
	   /**
	    * 无限大小的线程池(受限jvm内存)，当线程执行完任务会自动重用
	    * 空闲线程也会被回收（默认60秒）
	    * 实现队列为：SynchronousQueue
	    */
	   Executors.newCachedThreadPool(); 
	   /**
	    * 固定线程数的线程池
	    * 实现队列为：LinkedBlockingQueue
	    */
	   Executors.newFixedThreadPool(10);
	   /**
	    * 单线程执行器，线程池只有一个线程，所以线程按顺序执行，当一个线程异常结束时
	    * 会创建新的线程替代。     
	    * 实现队列为：LinkedBlockingQueue
	    */
	   Executors.newSingleThreadExecutor(); 
	   /**
	    * 创建一个无限大小的线程池，周期执行任务
	    * 实现队列为：DelayedWorkQueue
	    */
	   Executors.newScheduledThreadPool(10);//创建核心线程数10的时间相关线程池
	   
	   /***********************************************************************/
	   //以上线程池的创建都是通过ThreadPoolExecutor的构造方法创建的。
   }
}

分享到：

JAVA NIO | 责任链模式(Chain of Responsibility Metho ...

2014-02-13 15:16
浏览 993
评论(0)
分类:编程语言
查看更多

发表评论

您还没有登录,请您登录后再发表评论

最近访客更多访客>>

博主相关

文章分类

社区版块

存档分类

最新评论

线程池

评论

发表评论

相关推荐

最近访客 更多访客>>

博主相关

文章分类

社区版块

存档分类

最新评论

线程池

评论

发表评论

相关推荐

JAVA NIO

Java基础知识

Java基础知识

最近访客更多访客>>