文档库

最新最全的文档下载
当前位置:文档库 > Java并发--任务执行

Java并发--任务执行

首先来看一下,任务的定义:

所谓的任务,就是抽象,离散的工作单位。你可以简单理解为代码级别的 (Runnable接口)大多数并发应用程序都是围绕着任务进行管理的.

我们来看一小段代码:

Java代码

package com.ivan.concurrent.charpter6;

import http://www.wendangku.net/doc/b041706d25c52cc58bd6be44.html.ServerSocket;

import http://www.wendangku.net/doc/b041706d25c52cc58bd6be44.html.Socket;

/**

* 顺序化的Web Server.

* @author root

* OS:Ubuntu 9.04

* Date:2010-6-19

*/

public class SingleThreadWebServer {

public static void main(String[] args) throws Exception {

ServerSocket server=new ServerSocket(8080);

while(true){

Socket socket=server.accept();

handleRequest(socket);

}

}

private static void handleRequest(Socket socket) {

/**

* 做相关的处理……,比如请求运算与I/O

* 这将会导致出现阻塞,会延迟当前请求的处理,

* 而且会产生非常严重的后果,比如:假死。

* 那样会极度考验用户的耐心,知道他忍无可忍的关闭浏览器。

* 同时,单线程在等待IO操作时,CPU处于闲置状态,这样也降低了资源的利用率

*

* 这样的服务器,缺乏良好的吞吐量和快速的响应性。

*/

}

}

package com.ivan.concurrent.charpter6;

import http://www.wendangku.net/doc/b041706d25c52cc58bd6be44.html.ServerSocket;

import http://www.wendangku.net/doc/b041706d25c52cc58bd6be44.html.Socket;

/**

* 顺序化的Web Server.

* @author root

* OS:Ubuntu 9.04

* Date:2010-6-19

*/

public class SingleThreadWebServer {

public static void main(String[] args) throws Exception {

ServerSocket server=new ServerSocket(8080);

while(true){

Socket socket=server.accept();

handleRequest(socket);

}

}

private static void handleRequest(Socket socket) {

/**

* 做相关的处理……,比如请求运算与I/O

* 这将会导致出现阻塞,会延迟当前请求的处理,

* 而且会产生非常严重的后果,比如:假死。

* 那样会极度考验用户的耐心,知道他忍无可忍的关闭浏览器。

* 同时,单线程在等待IO操作时,CPU处于闲置状态,这样也降低了资源的利用率

*

* 这样的服务器,缺乏良好的吞吐量和快速的响应性。

*/

}

}

上面的代码是顺序地执行任务,主线程在不断接受连接与处理请求之间交替运行。

一个Web请求会做相关的处理……,比如请求运算与I/O

这将会导致出现阻塞,会延迟当前请求的处理,

而且会产生非常严重的后果,比如:假死。

那样会极度考验用户的耐心,知道他忍无可忍的关闭浏览器。

同时,单线程在等待IO操作时,CPU处于闲置状态,这样也降低了资源的利用率

这样的服务器,缺乏良好的吞吐量和快速的响应性。

所以,基于上面代码的基础上,我们需要给他作些小许的改进:

Java代码

package com.ivan.concurrent.charpter6;

import http://www.wendangku.net/doc/b041706d25c52cc58bd6be44.html.ServerSocket;

import http://www.wendangku.net/doc/b041706d25c52cc58bd6be44.html.Socket;

public class ThreadPerTaskWebServer {

public static void main(String[] args) throws Exception {

ServerSocket server=new ServerSocket(80);

while(true){

final Socket socket=server.accept();

new Thread(new Runnable(){

public void run() {

handleRequest(socket);

}

}).start();

}

}

protected static void handleRequest(Socket socket) {

/**

*相比较而言,这样的处理方式有良好的改进:

* 1.执行人物的负载已经脱离主线程,让主循环能更加迅速的重新开始等待下一个连接。提高了响应性

* 2.并发处理任务,多个请求可以同时得到处理,提高了吞吐性

* 3.任务处理代码必须要是线程安全的。防止出现并发性数据共享问题。

*

* 这个程序可能在开发阶段运行良好,一旦部署,就可能出现致命的错误,

* 我们接着来分析:

*/

}

}

package com.ivan.concurrent.charpter6;

import http://www.wendangku.net/doc/b041706d25c52cc58bd6be44.html.ServerSocket;

import http://www.wendangku.net/doc/b041706d25c52cc58bd6be44.html.Socket;

public class ThreadPerTaskWebServer {

public static void main(String[] args) throws Exception {

ServerSocket server=new ServerSocket(80);

while(true){

final Socket socket=server.accept();

new Thread(new Runnable(){

public void run() {

handleRequest(socket);

}

}).start();

}

}

protected static void handleRequest(Socket socket) {

/**

*相比较而言,这样的处理方式有良好的改进:

* 1.执行人物的负载已经脱离主线程,让主循环能更加迅速的重新开始等待下一个连接。提高了响应性

* 2.并发处理任务,多个请求可以同时得到处理,提高了吞吐性

* 3.任务处理代码必须要是线程安全的。防止出现并发性数据共享问题。

*

* 这个程序可能在开发阶段运行良好,一旦部署,就可能出现致命的错误,

* 我们接着来分析:

*/

}

}

相比较而言,这样的处理方式有良好的改进:

1.执行人物的负载已经脱离主线程,让主循环能更加迅速的重新开始等待下一个连接。提高了响应性

2.并发处理任务,多个请求可以同时得到处理,提高了吞吐性

3.任务处理代码必须要是线程安全的。防止出现并发性数据共享问题。

这个程序可能在开发阶段运行良好,一旦部署,就可能出现致命的错误,

我们接着来分析:

我们看到,上面的代码中,是为每个请求的到来,创建一个新的线程来处理,那么这样就会有以下的问题出现:

无限创建线程的缺点:

1.线程生命周期的开销

1.1.线程的创建与关闭并非是免费的,实际的开销根据不同的OS有不同的处理.但是线程的

创建的确需要时间,带来处理请求的延迟.一般的Web Server的请求是很频繁的,为每个请求创建一个线程,无非要耗费大量的资源.

2.资源消耗量

2.1. 活动的线程会消耗资源,尤其是内存.如果可运行的线程数多于可用的处理器数,线程将会空闲。大量的空闲线程占用更多的内存,给垃圾回收器带来压力,而且,线程在竞争CPU 的同时,也会带来许多其他的性能开销。所以,建议在有足够多的线程让CPU忙碌时,不要再创建多余的线程.

3.应用的稳定性

3.1. 应该限制创建线程的数量,限制的数目根据不同的平台而定,同时也受到JVM的启动参数,Thread的构造函数中栈大小等因素的影响. 如果打破了这个限制,你很可能会得到一个OutOfMemoryError. 在一定范围内增加线程可以提高系统的吞吐量,但是一旦超过这个范围,再创建线程只会拖垮你的系统。甚至可能会导致应用程序的崩溃.

我们的解决办法:

使用线程池,当然,你完全没有必要自己写一个线程池的实现(好吧,或许你跟我一样,也希望能从重复创造轮子中,找到自己想要了解的东西),你可以利用 Executor框架来帮你处理,java.util.concurrent提供了一个灵活的线程池实现。在新的java类库当中,任务执行的首要抽象不是Thread,而是Executor.

Executor仅仅是一个简单的接口,但是它很强大,包括用于异步任务的执行,支持不同类型的任务执行策略,为任务提交和任务执行之间的解藕,提供了标准的方式等等,我们后续再重点讨论。

Executor基于生产者-消费者模式。提交任务的是生产者,执行任务的是消费者。也就是说,采用Executor框架实现生产者-消费者模式,十分简单。

Java代码

package com.ivan.concurrent.charpter6;

import http://www.wendangku.net/doc/b041706d25c52cc58bd6be44.html.ServerSocket;

import http://www.wendangku.net/doc/b041706d25c52cc58bd6be44.html.Socket;

import java.util.concurrent.Executor;

import java.util.concurrent.Executors;

public class TaskExecutionWebServer{

private static final int NTHREADS=100;

//使用线程池来避免为每个请求创建一个线程。

private static final Executor threadPool=Executors.newFixedThreadPool(NTHREADS);

public static void main(String[] args) throws Exception {

ServerSocket server=new ServerSocket(8011);

while(true){

final Socket socket=server.accept();

threadPool.execute(new Runnable(){

public void run() {

handleRequest(socket);

}

});

}

}

protected static void handleRequest(Socket socket) {

/**

*

*/

System.out.println(Thread.currentThread().getId());

try {

Thread.sleep(5000);

} catch (InterruptedException e) {

e.printStackTrace();

}

}

}

package com.ivan.concurrent.charpter6;

import http://www.wendangku.net/doc/b041706d25c52cc58bd6be44.html.ServerSocket;

import http://www.wendangku.net/doc/b041706d25c52cc58bd6be44.html.Socket;

import java.util.concurrent.Executor;

import java.util.concurrent.Executors;

public class TaskExecutionWebServer{

private static final int NTHREADS=100;

//使用线程池来避免为每个请求创建一个线程。

private static final Executor threadPool=Executors.newFixedThreadPool(NTHREADS);

public static void main(String[] args) throws Exception {

ServerSocket server=new ServerSocket(8011);

while(true){

final Socket socket=server.accept();

threadPool.execute(new Runnable(){

public void run() {

handleRequest(socket);

}

});

}

}

protected static void handleRequest(Socket socket) {

/**

*

*/

System.out.println(Thread.currentThread().getId());

try {

Thread.sleep(5000);

} catch (InterruptedException e) {

e.printStackTrace();

}

}

}

线程池:

线程池管理着一个工作者线程的同构池,线程池是与工作队列紧密绑定的。工作队列的作用就是持有所有等待执行的任务,工作者队列只需要从工作队列中获取到下一个任务,执行,然后回来等待下一个线程。

Java类库中提供了以下几种线程池:

1.newFixedThreadPool :创建定长的线程池,每当提交一个任务就创建一个线程,直到达到池的最大长度。

2.newCachedThreadPool:创建一个可缓存的线程池,如果当前线程池的长度超过了处理的需要,它可以灵活的收回空闲线程,当需求增加时,它可以灵活添加新的线程,而并不对池的长度做任何限制

3.newSingleThreadExecutor:创建单线程化的executor,它只创建唯一的工作者线程来执行任务,如果这个线程异常结束,会有另外一个线程来取代它.它会保证任务按照任务队列规定的顺序来执行。

4.NewScheduledThreadPool:创建一个定长的线程池,而且支持定时的,以及周期性的任务执行,类似Timer.

Executor的生命周期:

它的创建已经说了,我们来看看它如何关闭, Executor 是为了执行任务而创建线程,而JVM通常会在所有非后台线程退出后才退出,如果它无法正确的关闭,则会影响到JVM的结束。

这里需要提一下,在我们了解如何关闭Executor的一些疑惑,由于Executor是异步执行任务,那么这些任务的状态不是立即可见的,换句话说,在任务时间里,这些执行的任务中,有的可能已经完成,有的还可能在运行,其他的还可能在队列里面等待。为了解决这些问题, Java引入了另外一个接口,它扩展了Executor,并增加一些生命周期的管理方法:ExecutorService.

ExecutorService表示生命周期有三种状态:运行,关闭,终止。

关闭和终止?怎么看上去是一个意思,这里我们先搁置着,留着后续来讨论。

ExecutorService最初创建后的初始状态就是运行状态;

shutdown与shutdownNow方法,都是ExecutorService的关闭方法,区别在于:

shutdown:

会启动一个平稳的关闭过程,停止接受新任务,同时等待已经提交的任务完成(包括尚未开始执行的任务)

shutdownNow:

会启动一个强制关闭的过程:尝试取消所有运行中的任务和排在队列中尚未开始的任务。

一旦所有任务全完成后,ExecutorService会转到终止状态, awaitTermination可以用来等待ExecutorService到达终止状态,也可以轮询isTerminated判断ExecutorService 是否已经终止。

Java代码

package com.ivan.concurrent.charpter6;

import java.io.IOException;

import http://www.wendangku.net/doc/b041706d25c52cc58bd6be44.html.ServerSocket;

import http://www.wendangku.net/doc/b041706d25c52cc58bd6be44.html.Socket;

import java.util.concurrent.ExecutorService;

import java.util.concurrent.Executors;

import java.util.concurrent.RejectedExecutionException;

/**

* 线程池的生命周期是如何管理的?

* @author root

* OS:Ubuntu 9.04

* Date:2010-6-19

*/

public class LifeCycleWebServer {

private static final int NTHREADS=100;

private static final ExecutorService exec=Executors.newFixedThreadPool(NTHREADS);

public void start() throws IOException{

ServerSocket server=new ServerSocket(8011);

while(exec.isShutdown()){

try {

final Socket socket=server.accept();

exec.execute(new Runnable(){

public void run() {

handleRequest(socket);

}

});

} catch (RejectedExecutionException e) { if(!exec.isShutdown()){

//log.error(...)

}

}

}

}

protected void handleRequest(Socket socket) {

Request req=readRequest(socket);

if(isShutDown(req)){

stop();

}else{

dispatchRequest(req);

}

}

public void stop(){

exec.shutdown();

}

//~ Mock Object And Function..

private static class Request{

}

private Request readRequest(Socket socket) {

// TODO Auto-generated method stub

return null;

}

private boolean isShutDown(Request req) {

// TODO Auto-generated method stub

return false;

}

private void dispatchRequest(Request req) {

// TODO Auto-generated method stub

}

}

package com.ivan.concurrent.charpter6;

import java.io.IOException;

import http://www.wendangku.net/doc/b041706d25c52cc58bd6be44.html.ServerSocket;

import http://www.wendangku.net/doc/b041706d25c52cc58bd6be44.html.Socket;

import java.util.concurrent.ExecutorService;

import java.util.concurrent.Executors;

import java.util.concurrent.RejectedExecutionException;

/**

* 线程池的生命周期是如何管理的?

* @author root

* OS:Ubuntu 9.04

* Date:2010-6-19

*/

public class LifeCycleWebServer {

private static final int NTHREADS=100;

private static final ExecutorService exec=Executors.newFixedThreadPool(NTHREADS);

public void start() throws IOException{

ServerSocket server=new ServerSocket(8011);

while(exec.isShutdown()){

try {

final Socket socket=server.accept();

exec.execute(new Runnable(){

public void run() {

handleRequest(socket);

}

});

} catch (RejectedExecutionException e) {

if(!exec.isShutdown()){

//log.error(...)

}

}

}

}

protected void handleRequest(Socket socket) { Request req=readRequest(socket);

if(isShutDown(req)){

stop();

}else{

dispatchRequest(req);

}

}

public void stop(){

exec.shutdown();

}

//~ Mock Object And Function..

private static class Request{

}

private Request readRequest(Socket socket) { // TODO Auto-generated method stub

return null;

}

private boolean isShutDown(Request req) {

// TODO Auto-generated method stub

return false;

}

private void dispatchRequest(Request req) { // TODO Auto-generated method stub

}

}

OK,了解了线程池的使用,这里有必要介绍介绍执行策略,

执行策略:

简单来说,就是任务执行的”What,When,Where,How”,包括:

1.任务在什么线程中执行(what)

2.任务以什么顺序执行(fifo,lifo,优先级)?

3.可以有多少个任务并发执行?(how many)

4.可以有多少个任务进入等待执行队列

5.系统过载时,需要放弃一个任务,该挑选哪一个?如何通知应用程序知道?

另外,java类库中还提供有一种特别的任务,----可携带结果的任务:

Callable 和 Future

Runnable 作为任务的基本表达形式只是个相当有限的抽象;它的局限在于,不能返回一个值或者抛出受检查的异常。

通常,很多任务都会引起严重的计算延迟,比如执行数据库查询,从网络下载资源,进行复杂的计算。对于这样的任务,Callable是更佳的抽象:它在主进入点,等待返回值,并为可能抛出的异常预先作准备。

Runnable与Callable描述的都是首相的计算型任务,这些任务通常都是有限的。,任务的所生命周期分为4个阶段:创建、提交、开始和完成。

Future描述了任务的生命周期,并提供了相关的方法来获取任务的结果、取消任务以及检验任务是否已经完成或者被取消。

Future的get方法取决于任务的状态,如果任务已经完成,get会立即返回或者抛出异常,如果任务没有完成,get会阻塞直到它的完成。

创建Future的方法有很多, ExecutorService的submit会返回一个Future,你可以将一个Callable或者Runnable提交给executor,然后得到一个Future,用它来重新获得任务执行的结果,或者取消任务。

你也可以显示的为给定的Callable和Runnable实例化一个FutureTask.

OK, 前面介绍了很多关于并发的理论知识,下面我们来看看,如果寻找可强化的并发性。

首先,我们从一个例子开始,开始之前,简单介绍一下这个例子所要表达的事情:它的来源是浏览器程序中渲染页面的那部分功能,首先获取HTML,并将它渲染到图像缓存里。为了简单起见,我们假设HTML只有文本标签。 OK,开始吧。

首先,如果按照一般的处理方式,我们会这样做:

1.遇到文本标签,将它渲染到图像缓存中

2.当遇到的是一个图片标签,我们通过网络获取它,再将它放到缓存里面。

很明显,这是最简单的方式,它很容易实现,但是,问题在于,你这样做,是在考验用户的耐心,结果就是他会对着屏幕丢一句 ****.然后毫不犹豫的关掉浏览器.

另外一种方法:

它先渲染文本,并为图像预留出占位符;在完成第一趟文本处理后,程序返回开始,并下载图像,将它们绘制到占位符上去。但是这样的问题也很明显,需要最少2次的文档处理,其性能与效率稍有提升,但是还不足解决用户希望快速浏览页面的需求。

为了使我们的渲染器具有更高的并发性,我们需要做的第一步就是,将渲染过程分为两部分:一个用来渲染文本,一个用来下载所有图像。(一个受限于CPU,另外一个受限于IO,即使在单CPU系统上,效率的提升也很明显。)

Callable与Future可以用来表达所有协同工作的任务之间的交互。我们来看代码:

Java代码

package com.ivan.concurrent.charpter6;

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;

public class FutureRenderer {

private static final int NTHREADS=100;

private static final ExecutorService exec=Executors.newFixedThreadPool(NTHREADS);

void renderPage(CharSequence source){

final List imageinfos=scanForImageInfo(source);

Callable> task=

new Callable>(){

public List call() throws Exception {

List result=new ArrayList();

for(ImageInfo imageinfo:imageinfos){

result.add(imageinfo.downloadImage());

}

return result;

}

};

Future> future=exec.submit(task);

//保证渲染文本与下载图像数据并发执行。

renderText(source);

try {

/**

* 到达需要所有图像的时间点时,主任务会等待future.get调用的结果, * 幸运的话,我们请求的同时,下载已经完成,即使没有,下载也已经预先开始了。

*

* 这里还有一定的局限性,用户可能不希望等待所有图片下载完成后才可以看见,

* 他希望下载完成一张图片后,就可以立即看到。……这里还待优化。 */

List imageData=future.get();

for(ImageData data:imageData){

reanderImage(data);

}

} catch (InterruptedException e) {

Thread.currentThread().interrupt();

future.cancel(true);//取消任务

}catch(ExecutionException e){

e.printStackTrace();

}

}

private void renderText(CharSequence source) {

// TODO Auto-generated method stub

}

private void reanderImage(ImageData data) {

// TODO Auto-generated method stub

}

private List scanForImageInfo(CharSequence source) {

// TODO Auto-generated method stub

return null;

}

}

package com.ivan.concurrent.charpter6;

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;

public class FutureRenderer {

private static final int NTHREADS=100;

private static final ExecutorService exec=Executors.newFixedThreadPool(NTHREADS);

void renderPage(CharSequence source){

final List imageinfos=scanForImageInfo(source);

Callable> task=

new Callable>(){

public List call() throws Exception {

List result=new ArrayList();

for(ImageInfo imageinfo:imageinfos){

result.add(imageinfo.downloadImage());

}

return result;

}

};

Future> future=exec.submit(task);

//保证渲染文本与下载图像数据并发执行。

renderText(source);

try {

/**

* 到达需要所有图像的时间点时,主任务会等待future.get调用的结果,

* 幸运的话,我们请求的同时,下载已经完成,即使没有,下载也已经预先开始了。

*

* 这里还有一定的局限性,用户可能不希望等待所有图片下载完成后才可以看见,

* 他希望下载完成一张图片后,就可以立即看到。……这里还待优化。

*/

List imageData=future.get();

for(ImageData data:imageData){

reanderImage(data);

}

} catch (InterruptedException e) {

Thread.currentThread().interrupt();

future.cancel(true);//取消任务

}catch(ExecutionException e){

e.printStackTrace();

}

}

private void renderText(CharSequence source) {

// TODO Auto-generated method stub

}

private void reanderImage(ImageData data) {

// TODO Auto-generated method stub

}

private List scanForImageInfo(CharSequence source) {

// TODO Auto-generated method stub

return null;

}

}

CompletionService: 当executorService遇到BlockingQueue

CompletionService整合了Executor和BlockingQueue的功能,你可以将Callable任务提交给它去执行,然后使用类似于队列中的take和poll方法,在结果完成可用时,获得这个结果,像一个打包的Future.

我们利用它来为我们的渲染器需要优化的地方做些处理,代码如下:

Java代码

package com.ivan.concurrent.charpter6;

import java.util.List;

import java.util.concurrent.Callable;

import http://www.wendangku.net/doc/b041706d25c52cc58bd6be44.htmlpletionService;

import java.util.concurrent.ExecutionException;

import java.util.concurrent.ExecutorCompletionService;

import java.util.concurrent.ExecutorService;

import java.util.concurrent.Executors;

import java.util.concurrent.Future;

public class FutureRenderer2 {

private static final int NTHREADS=100;

private static final ExecutorService exec=Executors.newFixedThreadPool(NTHREADS);

void renderPage(CharSequence source){

final List imageinfos=scanForImageInfo(source);

CompletionService completionService=new ExecutorCompletionService(exec);

for(final ImageInfo imageinfo:imageinfos){

completionService.submit(new Callable(){

public ImageData call() throws Exception {

//提高性能点一:将顺序的下载,变成并发的下载,缩短下载时间 return imageinfo.downloadImage();

}

});

}

renderText(source);

try {

for(int i=0;i

Future f=completionService.take();

//提高性能点二:下载完成一张图片后,立刻渲染到页面。

ImageData imagedata=f.get();

reanderImage(imagedata);

}

} catch (InterruptedException e) {

Thread.currentThread().interrupt();

}catch(ExecutionException e){

e.printStackTrace();

}

}

private void renderText(CharSequence source) {

// TODO Auto-generated method stub

}

private void reanderImage(ImageData data) {