【转】java 并发工具不完全浅谈_JAVA

您所在的位置：程序员俱乐部 > 编程开发 > JAVA > 【转】java 并发工具不完全浅谈

【转】java 并发工具不完全浅谈

2015/3/11 21:16:14 RoomFourteen224 程序员俱乐部我要评论(0)

摘要：转】java并发工具不完全浅谈这篇文章就主要讨论讨论Java并发中的任务执行，来作为我整理的地一篇文章吧。文中难免有错，如果发现问题，可以即时站内或者回帖交流。OK，言归正传首先来看一下，任务的定义：所谓的任务，就是抽象，离散的工作单位。你可以简单理解为代码级别的(Runnable接口）大多数并发应用程序都是围绕着任务进行管理的.我们来看一小段代码：Java代码packagecom.ivan.concurrent.charpter6;importjava.net.ServerSocket
标签：工具 Java 浅谈

转】java 并发工具不完全浅谈这篇文章就主要讨论讨论Java并发中的任务执行，来作为我整理的地一篇文章吧。文中难免有错，如果发现问题，可以即时站内或者回帖交流。

? OK，言归正传
??
首先来看一下，任务的定义：

?所谓的任务，就是抽象，离散的工作单位。你可以简单理解为代码级别的 (Runnable 接口）

?大多数并发应用程序都是围绕着任务进行管理的.

?我们来看一小段代码：

Java代码??

package?com.ivan.concurrent.charpter6;??
??
import?java.net.ServerSocket;??
import?java.net.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?java.net.ServerSocket;??
import?java.net.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?java.net.ServerSocket;??
import?java.net.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?java.net.ServerSocket;??
import?java.net.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<ImageInfo>?imageinfos=scanForImageInfo(source);??
????????Callable<List<ImageData>>?task=??
????????????????new?Callable<List<ImageData>>(){??
????????????????????public?List<ImageData>?call()?throws?Exception?{??
????????????????????????List<ImageData>?result=new?ArrayList<ImageData>();??
????????????????????????for(ImageInfo?imageinfo:imageinfos){??
????????????????????????????result.add(imageinfo.downloadImage());??
????????????????????????}??
????????????????????????return?result;??
????????????????????}??
??????????????
????????};??
??????????
??????????
????????Future<List<ImageData>>?future=exec.submit(task);??
????????//保证渲染文本与下载图像数据并发执行。??
????????renderText(source);??
????????try?{??
????????????/**?
?????????????*?到达需要所有图像的时间点时，主任务会等待future.get调用的结果，?
?????????????*??幸运的话，我们请求的同时，下载已经完成，即使没有，下载也已经预先开始了。?
?????????????*???
?????????????*??这里还有一定的局限性，?用户可能不希望等待所有图片下载完成后才可以看见，?
?????????????*???他希望下载完成一张图片后，就可以立即看到。?……?这里还待优化。?
?????????????*/??
????????????List<ImageData>?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<ImageInfo>?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?java.util.concurrent.CompletionService;??
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<ImageInfo>?imageinfos=scanForImageInfo(source);??
??????????
????????CompletionService<ImageData>?completionService=new?ExecutorCompletionService<ImageData>(exec);??
??????????
????????for(final?ImageInfo?imageinfo:imageinfos){??
????????????completionService.submit(new?Callable<ImageData>(){??
????????????????public?ImageData?call()?throws?Exception?{??
????????????????????//提高性能点一：?将顺序的下载，变成并发的下载，缩短下载时间??
????????????????????return?imageinfo.downloadImage();??
????????????????}??
????????????});??
????????}??
????????renderText(source);??
????????try?{??
????????????for(int?i=0;i<imageinfos.size();i++){??
????????????????Future<ImageData>?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)?{??
????????//?TODO?Auto-generated?method?stub??
??????????
????} ?
??
????private?List<ImageInfo>?scanForImageInfo(CharSequence?source)?{??
????????//?TODO?Auto-generated?method?stub??
????????return?null;??
????}??
}??