`
mshijie
  • 浏览: 96124 次
  • 性别: Icon_minigender_1
  • 来自: 杭州
社区版块
存档分类
最新评论

java并发编程-Executor框架

阅读更多

java并发编程-构建块

Executor框架是指java 5中引入的一系列并发库中与executor相关的一些功能类,其中包括线程池,Executor,Executors,ExecutorService,CompletionService,Future,Callable等。他们的关系为:


 

并发编程的一种编程方式是把任务拆分为一些列的小任务,即Runnable,然后在提交给一个Executor执行,Executor.execute(Runnalbe) 。Executor在执行时使用内部的线程池完成操作。

一、创建线程池

Executors类,提供了一系列工厂方法用于创先线程池,返回的线程池都实现了ExecutorService接口。

public static ExecutorService newFixedThreadPool(int nThreads)

创建固定数目线程的线程池。

public static ExecutorService newCachedThreadPool()

创建一个可缓存的线程池,调用execute 将重用以前构造的线程(如果线程可用)。如果现有线程没有可用的,则创建一个新线程并添加到池中。终止并从缓存中移除那些已有 60 秒钟未被使用的线程。

public static ExecutorService newSingleThreadExecutor()

创建一个单线程化的Executor。

public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize)

创建一个支持定时及周期性的任务执行的线程池,多数情况下可用来替代Timer类。

Executor executor = Executors.newFixedThreadPool(10);
Runnable task = new Runnable() {
	@Override
	public void run() {
		System.out.println("task over");
	}
};
executor.execute(task);

executor = Executors.newScheduledThreadPool(10);
ScheduledExecutorService scheduler = (ScheduledExecutorService) executor;
scheduler.scheduleAtFixedRate(task, 10, 10, TimeUnit.SECONDS);

 二、ExecutorService与生命周期

ExecutorService扩展了Executor并添加了一些生命周期管理的方法。一个Executor的生命周期有三种状态,运行关闭终止 。Executor创建时处于运行状态。当调用ExecutorService.shutdown()后,处于关闭状态,isShutdown()方法返回true。这时,不应该再想Executor中添加任务,所有已添加的任务执行完毕后,Executor处于终止状态,isTerminated()返回true。

如果Executor处于关闭状态,往Executor提交任务会抛出unchecked exception RejectedExecutionException。

ExecutorService executorService = (ExecutorService) executor;
while (!executorService.isShutdown()) {
	try {
		executorService.execute(task);
	} catch (RejectedExecutionException ignored) {
		
	}
}
executorService.shutdown();

 三、使用Callable,Future返回结果

Future<V>代表一个异步执行的操作,通过get()方法可以获得操作的结果,如果异步操作还没有完成,则,get()会使当前线程阻塞。FutureTask<V>实现了Future<V>和Runable<V>。Callable代表一个有返回值得操作。

		Callable<Integer> func = new Callable<Integer>(){
			public Integer call() throws Exception {
				System.out.println("inside callable");
				Thread.sleep(1000);
				return new Integer(8);
			}		
		};		
		FutureTask<Integer> futureTask  = new FutureTask<Integer>(func);
		Thread newThread = new Thread(futureTask);
		newThread.start();
		
		try {
			System.out.println("blocking here");
			Integer result = futureTask.get();
			System.out.println(result);
		} catch (InterruptedException ignored) {
		} catch (ExecutionException ignored) {
		}

 ExecutoreService提供了submit()方法,传递一个Callable,或Runnable,返回Future。如果Executor后台线程池还没有完成Callable的计算,这调用返回Future对象的get()方法,会阻塞直到计算完成。

例子:并行计算数组的和。

package executorservice;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.FutureTask;

public class ConcurrentCalculator {

	private ExecutorService exec;
	private int cpuCoreNumber;
	private List<Future<Long>> tasks = new ArrayList<Future<Long>>();

	// 内部类
	class SumCalculator implements Callable<Long> {
		private int[] numbers;
		private int start;
		private int end;

		public SumCalculator(final int[] numbers, int start, int end) {
			this.numbers = numbers;
			this.start = start;
			this.end = end;
		}

		public Long call() throws Exception {
			Long sum = 0l;
			for (int i = start; i < end; i++) {
				sum += numbers[i];
			}
			return sum;
		}
	}

	public ConcurrentCalculator() {
		cpuCoreNumber = Runtime.getRuntime().availableProcessors();
		exec = Executors.newFixedThreadPool(cpuCoreNumber);
	}

	public Long sum(final int[] numbers) {
		// 根据CPU核心个数拆分任务,创建FutureTask并提交到Executor
		for (int i = 0; i < cpuCoreNumber; i++) {
			int increment = numbers.length / cpuCoreNumber + 1;
			int start = increment * i;
			int end = increment * i + increment;
			if (end > numbers.length)
				end = numbers.length;
			SumCalculator subCalc = new SumCalculator(numbers, start, end);
			FutureTask<Long> task = new FutureTask<Long>(subCalc);
			tasks.add(task);
			if (!exec.isShutdown()) {
				exec.submit(task);
			}
		}
		return getResult();
	}

	/**
	 * 迭代每个只任务,获得部分和,相加返回
	 * 
	 * @return
	 */
	public Long getResult() {
		Long result = 0l;
		for (Future<Long> task : tasks) {
			try {
				// 如果计算未完成则阻塞
				Long subSum = task.get();
				result += subSum;
			} catch (InterruptedException e) {
				e.printStackTrace();
			} catch (ExecutionException e) {
				e.printStackTrace();
			}
		}
		return result;
	}

	public void close() {
		exec.shutdown();
	}
}

 Main

int[] numbers = new int[] { 1, 2, 3, 4, 5, 6, 7, 8, 10, 11 };
ConcurrentCalculator calc = new ConcurrentCalculator();
Long sum = calc.sum(numbers);
System.out.println(sum);
calc.close();

 四、CompletionService

在刚在的例子中,getResult()方法的实现过程中,迭代了FutureTask的数组,如果任务还没有完成则当前线程会阻塞,如果我们希望任意字任务完成后就把其结果加到result中,而不用依次等待每个任务完成,可以使CompletionService。生产者submit()执行的任务。使用者take()已完成的任务,并按照完成这些任务的顺序处理它们的结果 。也就是调用CompletionService的take方法是,会返回按完成顺序放回任务的结果,CompletionService内部维护了一个阻塞队列BlockingQueue,如果没有任务完成,take()方法也会阻塞。修改刚才的例子使用CompletionService:

public class ConcurrentCalculator2 {

	private ExecutorService exec;
	private CompletionService<Long> completionService;


	private int cpuCoreNumber;

	// 内部类
	class SumCalculator implements Callable<Long> {
		......
	}

	public ConcurrentCalculator2() {
		cpuCoreNumber = Runtime.getRuntime().availableProcessors();
		exec = Executors.newFixedThreadPool(cpuCoreNumber);
		completionService = new ExecutorCompletionService<Long>(exec);


	}

	public Long sum(final int[] numbers) {
		// 根据CPU核心个数拆分任务,创建FutureTask并提交到Executor
		for (int i = 0; i < cpuCoreNumber; i++) {
			int increment = numbers.length / cpuCoreNumber + 1;
			int start = increment * i;
			int end = increment * i + increment;
			if (end > numbers.length)
				end = numbers.length;
			SumCalculator subCalc = new SumCalculator(numbers, start, end);	
			if (!exec.isShutdown()) {
				completionService.submit(subCalc);


			}
			
		}
		return getResult();
	}

	/**
	 * 迭代每个只任务,获得部分和,相加返回
	 * 
	 * @return
	 */
	public Long getResult() {
		Long result = 0l;
		for (int i = 0; i < cpuCoreNumber; i++) {			
			try {
				Long subSum = completionService.take().get();
				result += subSum;			
			} catch (InterruptedException e) {
				e.printStackTrace();
			} catch (ExecutionException e) {
				e.printStackTrace();
			}
		}
		return result;
	}

	public void close() {
		exec.shutdown();
	}
}

 五、例子HtmlRender

该例子模拟浏览器的Html呈现过程,先呈现文本,再异步下载图片,下载完毕每个图片即显示,见附件eclipse项目htmlreander包。

 

所有代码见附件,Eclipse项目。本文参考《Java并发编程实践 》。

 

  • 大小: 102.6 KB
分享到:
评论
9 楼 wslovenide 2012-06-15  
8 楼 nakata_yf 2010-03-16  
路过。。。
7 楼 lianj_lee 2009-09-06  
最近在做一个关于多线程的项目,而且项目中的线程需要能定时给点任务。
6 楼 vae_in_zdsoft 2009-07-27  
我们的应用中也有通过线程池管理并发任务,主要是有20个线程进行数据同步,但是现在存在的问题时有时候会一直卡住,需要重启下服务才行。我想问其中某些线程阻塞掉了能否显示的让其不再执行任务,可以让线程空出来供其他同步使用
5 楼 mercyblitz 2009-05-08  
jd2bs 写道

问个问题:楼主的例子里都是用的以下的形式:ExecutorService exec=Executors.newFixedThreadPool(nSize);exec.execute(tasks);exec.shutdown()我的疑问:实际使用中(比如我15分钟内可能有数万人要调用一个发mail的task)肯定不能这样每次都shutdown的吧?要是每次都shutdown每次都要new 那这开销也太大了我想是否应该写一个servlet,在init中new这个pool,然后在destroy中shutdown;在app启动时此servlet也启动了 这样这个ExecutorService可用(我们将它put到request的某个attribute中,甚至是application对象中);然后我们在service层拿到这个ExecutorService,执行任务。具体项目应用是这样的吗?还是我理解有误 直接将上面三步放一起就可以?



如果用户过多的话,尽量使用ThreadPoolExecutor设置一个阻塞队列,并且只需要
ThreadPoolExecutor#submit(Runnable r) ,
如果需要结果的返回的话,使用ThreadPoolExecutor#submit(Callable c) 然后使用Future来获取一个结果集合。

4 楼 mshijie 2009-04-28  
jd2bs 写道
问个问题:楼主的例子里都是用的以下的形式:
ExecutorService exec=Executors.newFixedThreadPool(nSize);
exec.execute(tasks);
exec.shutdown()

我的疑问:实际使用中(比如我15分钟内可能有数万人要调用一个发mail的task)肯定不能这样每次都shutdown的吧?要是每次都shutdown每次都要new 那这开销也太大了

我想是否应该写一个servlet,在init中new这个pool,然后在destroy中shutdown;
在app启动时此servlet也启动了 这样这个ExecutorService可用(我们将它put到request的某个attribute中,甚至是application对象中);然后我们在service层拿到这个ExecutorService,执行任务。

具体项目应用是这样的吗?

还是我理解有误 直接将上面三步放一起就可以?


我这只是一个Demo,演示如果关闭线程池,如果创建就关闭,那还用线程池干什么。实际中只要把exec保存到可以并发访问到的地方就可以。你的那个方法可以。
3 楼 jd2bs 2009-04-28  
问个问题:楼主的例子里都是用的以下的形式:
ExecutorService exec=Executors.newFixedThreadPool(nSize);
exec.execute(tasks);
exec.shutdown()

我的疑问:实际使用中(比如我15分钟内可能有数万人要调用一个发mail的task)肯定不能这样每次都shutdown的吧?要是每次都shutdown每次都要new 那这开销也太大了

我想是否应该写一个servlet,在init中new这个pool,然后在destroy中shutdown;
在app启动时此servlet也启动了 这样这个ExecutorService可用(我们将它put到request的某个attribute中,甚至是application对象中);然后我们在service层拿到这个ExecutorService,执行任务。

具体项目应用是这样的吗?

还是我理解有误 直接将上面三步放一起就可以?

2 楼 niveko 2009-04-17  
Executors只是对线程池一些特定情况的简洁使用,直接用ThreadPoolExecutor构造线程池会获得更为强大的功能。ThreadPoolExecutor的构造函数
public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue,
                              ThreadFactory threadFactory,
                              RejectedExecutionHandler handler)
1 楼 erlengleng 2009-04-17  
Thinking in Java 第四版第21章并发对Executor有介绍,但是没提供类图,没楼主介绍的简单明了!
1.5Concurrent的新特性是另人振奋的,相信看了此贴,大家对Executor有了初步的认识。
刚被禁言3天啊。注册1年多,积分是0,不敢贸然发表评论了。

相关推荐

    java并发编程-从入门到精通

    Java并发编程是Java开发者必须掌握的关键技能之一,尤其是在开发高性能、多线程的应用时。本教程“java并发编程-从入门到精通”旨在帮助你深入理解这个领域,并逐步提升你的编程能力。 首先,我们要理解Java并发的...

    java并发编程-构建块

    "java并发编程-构建块"这个主题涵盖了使程序能够同时处理多个任务的关键概念和技术。在这个主题下,我们将深入探讨Java中用于构建高效并发应用的核心工具和概念。 1. **线程**:Java中的线程是并发编程的基础,每个...

    java并发编程实战源码,java并发编程实战pdf,Java

    《Java并发编程实战》是Java并发编程领域的一本经典著作,它深入浅出地介绍了如何在Java平台上进行高效的多线程编程。这本书的源码提供了丰富的示例,可以帮助读者更好地理解书中的理论知识并将其应用到实际项目中。...

    Java 并发编程实战.pdf

    《Java并发编程实战》这本书是关于Java语言中并发编程技术的经典著作。它详细介绍了如何在Java环境中有效地实现多线程程序和并发控制机制。在Java平台上,由于其本身提供了强大的并发编程支持,因此,掌握并发编程...

    Java并发编程利器:Executor框架深度解析与应用实践

    在现代Java应用开发中,多线程并发编程已成为提升程序性能的关键技术之一。Java通过引入Executor框架,为并发任务的执行提供了一种高效、灵活的管理机制。本文将深入探讨Executor框架的设计哲学、核心组件,并结合...

    JAVA并发编程实践-中文-高清-带书签-完整版

    《JAVA并发编程实践》是Java开发人员深入理解并发编程的一本经典著作,由Doug Lea撰写,本书中文版高清完整,包含丰富的书签,便于读者查阅和学习。这本书旨在帮助开发者掌握在Java平台上进行高效、安全并发编程的...

    Java-Executor并发框架.docx

    Java并发框架中的Executor服务是Java 1.5引入的核心组件,位于`java.util.concurrent`包下,极大地简化了多线程编程。Executor接口虽然历史悠久,但其重要性不言而喻,很多开发者对其背后的原理并不十分了解。本文将...

    java 并发编程的艺术pdf清晰完整版 源码

    《Java并发编程的艺术》这本书是Java开发者深入理解并发编程的重要参考书籍。这本书全面地介绍了Java平台上的并发和多线程编程技术,旨在帮助开发者解决在实际工作中遇到的并发问题,提高程序的性能和可伸缩性。 ...

    JAVA并发编程艺术pdf版

    《JAVA并发编程艺术》是Java开发者深入理解和掌握并发编程的一本重要著作,它涵盖了Java并发领域的核心概念和技术。这本书详细阐述了如何在多线程环境下有效地编写高效、可靠的代码,对于提升Java程序员的技能水平...

    Java并发编程-并发编程知识点总结.docx

    ### Java并发编程知识点总结 #### 1. 什么是线程? 线程是操作系统能够进行运算调度的最小单位,它被包含在进程之中,是进程中的实际运作单位。程序员可以通过线程进行多处理器编程,利用多线程对运算密集型任务...

    java并发编程内部分享PPT

    Java并发编程是Java开发中的重要领域,特别是在多核处理器和分布式系统中,高效地利用并发可以极大地提升程序的性能和响应速度。这份“java并发编程内部分享PPT”显然是一个深入探讨这一主题的资料,旨在帮助开发者...

    13-Java并发编程学习宝典.zip

    Java并发编程是软件开发中的重要领域,特别是在大型系统和高并发场景中不可或缺。"13-Java并发编程学习宝典.zip" 包含了一系列关于Java并发编程的学习资源,旨在帮助开发者掌握多线程编程的核心技术和最佳实践。以下...

    Java 并发编程实战-随书源码

    《Java并发编程实战》这本书是Java开发者深入理解并发编程的重要参考书籍。本书旨在帮助程序员解决在多线程环境中遇到的实际问题,提升系统性能并保证其稳定性。随书源码提供了丰富的示例,让读者能够动手实践,加深...

    Java并发编程从入门到精通(pdf)(附源码)

    《Java并发编程从入门到精通》是一本专为Java开发者设计的深度学习并发编程的书籍。作者韩剑锋,凭借其12年的IT行业经验,曾担任多家IT公司的研发总监和技术总监,以其丰富的实战经验和深厚的理论知识,为读者提供了...

    《Java并发编程的艺术》

    《Java并发编程的艺术》内容涵盖Java并发编程机制的底层实现原理、Java内存模型、Java并发编程基础、Java中的锁、并发容器和框架、原子类、并发工具类、线程池、Executor框架等主题,每个主题都做了深入的讲解,同时...

    java并发编程书籍

    Java并发编程是软件开发中的一个关键领域,尤其是在大型企业级应用和分布式系统中。通过学习相关的书籍,开发者可以深入理解如何有效地设计和实现高效的多线程应用程序,避免并发问题,如竞态条件、死锁、活锁等。...

    Java并发编程实践.pdf

    ### Java并发编程实践 #### 一、并发编程基础 ##### 1.1 并发与并行的区别 在Java并发编程中,首先需要理解“并发”(Concurrency)和“并行”(Parallelism)的区别。“并发”指的是多个任务在同一时间段内交替...

    Java并发编程实践-电子书1-9章pdf

    《Java并发编程实践》是Java开发者深入理解并发编程的重要参考资料,尤其对于想要提升多线程应用设计和性能优化技能的程序员来说,这本书提供了丰富的实践经验和深入的理论知识。以下是根据提供的章节内容概述的一些...

    [Java并发编程实践].(Java.Concurrency.in.Practice).Brian.Goetz.英文原版.pdf

    ### Java并发编程实践 #### 书籍概述 《Java并发编程实践》是一本由Brian Goetz等人编写的关于Java并发编程的经典著作。本书深入浅出地介绍了Java 5.0及之后版本中新增加的并发特性,并对并发编程进行了全面而详尽...

Global site tag (gtag.js) - Google Analytics