多线程并发处理方式

消耗积分:1 | 格式:rar | 大小:0.3 MB | 2017-09-28

分享资料个

　　1. 捕获InterruptedException错误

　　请检查下面的代码片段：

　　public class Task implements Runnable {

　　private final BlockingQueue queue = 。。.;

　　@Override

　　public void run（） {

　　while （！Thread.currentThread（）.isInterrupted（）） {

　　String result = getOrDefault（（） -》 queue.poll（1L， TimeUnit.MINUTES）， “default”）;

　　//do smth with the result

　　}

　　T getOrDefault（Callable supplier， T defaultValue） {

　　try {

　　return supplier.call（）;

　　} catch （Exception e） {

　　logger.error（“Got exception while retrieving value.”， e）;

　　return defaultValue;

　　}

　　代码的问题是，在等待队列中的新元素时，是不可能终止线程的，因为中断的标志永远不会被恢复：

　　运行代码的线程被中断。

　　BlockingQueue # poll（）方法抛出InterruptedException异常，并清除了中断的标志。

　　while中的循环条件（！Thread.currentThread（）.isInterrupted（））的判断是true，因为标记已被清除。

　　为了防止这种行为，当一个方法被显式抛出（通过声明抛出InterruptedException）或隐式抛出（通过声明/抛出一个原始异常）时，总是捕获InterruptedException异常，并恢复中断的标志。

　　T getOrDefault（Callable supplier， T defaultValue） {

　　try {

　　return supplier.call（）;

　　} catch （InterruptedException e） {

　　logger.error（“Got interrupted while retrieving value.”， e）;

　　Thread.currentThread（）.interrupt（）;

　　return defaultValue;

　　} catch （Exception e） {

　　logger.error（“Got exception while retrieving value.”， e）;

　　return defaultValue;

　　}

　　2．使用特定的执行程序来阻止操作

　　因为一个缓慢的操作而使整个服务器变得无响应，这通常不是开发人员想要的。不幸的是，对于RPC，响应时间通常是不可预测的。

　　假设服务器有100个工作线程，有一个端点，称为100 RPS。在内部，它发出一个RPC调用，通常需要10毫秒。在某个时间点，此RPC的响应时间变为2秒，在峰值期间服务器能够做的惟一的一件事就是等待这些调用，而其他端点则无法访问。

　　@GET

　　@Path（“/genre/{name}”）

　　@Produces（MediaType.APPLICATION_JSON）

　　public Response getGenre（@PathParam（“name”） String genreName） {

　　Genre genre = potentiallyVerySlowSynchronousCall（genreName）;

　　return Response.ok（genre）.build（）;

　　}

　　解决这个问题最简单的方法是提交代码，它将阻塞调用变成一个线程池：

　　@GET

　　@Path（“/genre/{name}”）

　　@Produces（MediaType.APPLICATION_JSON）

　　public void getGenre（@PathParam（“name”） String genreName， @Suspended AsyncResponse response） {

　　response.setTimeout（1L， TimeUnit.SECONDS）;

　　executorService.submit（（） -》 {

　　Genre genre = potentiallyVerySlowSynchronousCall（genreName）;

　　return response.resume（Response.ok（genre）.build（））;

　　}）;

　　}

　　3. 传MDC的值

　　MDC（Mapped Diagnostic Context）通常用于存储单个任务的特定值。例如，在web应用程序中，它可能为每个请求存储一个请求id和一个用户id，因此MDC查找与单个请求或整个用户活动相关的日志记录变得更加容易。

　　2017-08-27 14:38:30，893 INFO ［server-thread-0］［requestId=060d8c7f， userId=2928ea66］ c.g.s.web.Controller - Message.

　　可是如果代码的某些部分是在专用线程池中执行的，则线程（提交任务的线程）中MDC就不会被继续传值。在下面的示例中，第7行的日志中包含“requestId”，而第9行的日志则没有：

　　@GET

　　@Path（“/genre/{name}”）

　　@Produces（MediaType.APPLICATION_JSON）

　　public void getGenre（@PathParam（“name”） String genreName， @Suspended AsyncResponse response） {

　　try （MDC.MDCCloseable ignored = MDC.putCloseable（“requestId”， UUID.randomUUID（）.toString（））） {

　　String genreId = getGenreIdbyName（genreName）; //Sync call

　　logger.trace（“Submitting task to find genre with id ‘{}’。”， genreId）; //‘requestId’ is logged

　　executorService.submit（（） -》 {

　　logger.trace（“Starting task to find genre with id ‘{}’。”， genreId）; //‘requestId’ is not logged

　　Response result = getGenre（genreId） //Async call

　　.map（artist -》 Response.ok（artist）.build（））

　　.orElseGet（（） -》 Response.status（Response.Status.NOT_FOUND）.build（））;

　　response.resume（result）;

　　}）;

　　}

　　这可以通过MDC#getCopyOfContextMap（）方法来解决：

　　。。.

　　public void getGenre（@PathParam（“name”） String genreName， @Suspended AsyncResponse response） {

　　try （MDC.MDCCloseable ignored = MDC.putCloseable（“requestId”， UUID.randomUUID（）.toString（））） {

　　。。.

　　logger.trace（“Submitting task to find genre with id ‘{}’。”， genreId）; //‘requestId’ is logged

　　withCopyingMdc（executorService，（） -》 {

　　logger.trace（“Starting task to find genre with id ‘{}’。”， genreId）; //‘requestId’ is logged

　　。。.

　　}）;

　　}

　　private void withCopyingMdc（ExecutorService executorService， Runnable function） {

　　Map

　　4．更改线程名称

　　为了简化日志读取和线程转储，可以自定义线程的名称。这可以通过创建ExecutorService时用一个ThreadFactory来完成。在流行的实用程序库中有许多ThreadFactory接口的实现：

　　com.google.common.util.concurrent.ThreadFactoryBuilde+r in Guava.

　　org.springframework.scheduling.concurrent.CustomizableThreadFactory in Spring.

　　org.apache.commons.lang3.concurrent.BasicThreadFactory in Apache Commons Lang 3.

　　ThreadFactory threadFactory = new BasicThreadFactory.Builder（）

　　.namingPattern（“computation-thread-%d”）

　　.build（）;

　　ExecutorService executorService = Executors.newFixedThreadPool（numberOfThreads， threadFactory）;

　　尽管ForkJoinPool不使用ThreadFactory接口，但也支持对线程的重命名：

　　ForkJoinPool.ForkJoinWorkerThreadFactory forkJoinThreadFactory = pool -》 {

　　ForkJoinWorkerThread thread = ForkJoinPool.defaultForkJoinWorkerThreadFactory.newThread（pool）;

　　thread.setName（“computation-thread-” + thread.getPoolIndex（））;

　　return thread;

　　};

　　ForkJoinPool forkJoinPool = new ForkJoinPool（numberOfThreads， forkJoinThreadFactory， null， false）;

　　将线程转储与默认命名进行比较：

　　“pool-1-thread-3” #14 prio=5 os_prio=31 tid=0x00007fc06b19f000 nid=0x5703 runnable ［0x0000700001ff9000］