5個(gè)并發(fā)處理技巧代碼示例

更新時(shí)間：2017年10月30日 11:35:39 作者：Igor Sorokin

這篇文章主要介紹了5個(gè)并發(fā)處理技巧代碼示例，具有一定參考價(jià)值，需要的朋友可以了解下。

【譯者注】在本文中，作者總結(jié)出了5個(gè)關(guān)于處理并發(fā)性程序的技巧，并給出代碼示例，讓讀者更好地理解和使用這5種方法。以下為譯文：

1.捕獲InterruptedException錯(cuò)誤

請檢查下面的代碼片段：

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;
		}
	}
}

代碼的問題是，在等待隊(duì)列中的新元素時(shí)，是不可能終止線程的，因?yàn)橹袛嗟臉?biāo)志永遠(yuǎn)不會(huì)被恢復(fù)：

1.運(yùn)行代碼的線程被中斷。
2.BlockingQueue # poll()方法拋出InterruptedException異常，并清除了中斷的標(biāo)志。
3.while中的循環(huán)條件 (!Thread.currentThread().isInterrupted())的判斷是true，因?yàn)闃?biāo)記已被清除。

為了防止這種行為，當(dāng)一個(gè)方法被顯式拋出（通過聲明拋出InterruptedException）或隱式拋出（通過聲明/拋出一個(gè)原始異常）時(shí)，總是捕獲InterruptedException異常，并恢復(fù)中斷的標(biāo)志。

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．使用特定的執(zhí)行程序來阻止操作

因?yàn)橐粋€(gè)緩慢的操作而使整個(gè)服務(wù)器變得無響應(yīng)，這通常不是開發(fā)人員想要的。不幸的是，對于RPC，響應(yīng)時(shí)間通常是不可預(yù)測的。

假設(shè)服務(wù)器有100個(gè)工作線程，有一個(gè)端點(diǎn)，稱為100 RPS。在內(nèi)部，它發(fā)出一個(gè)RPC調(diào)用，通常需要10毫秒。在某個(gè)時(shí)間點(diǎn)，此RPC的響應(yīng)時(shí)間變?yōu)?秒，在峰值期間服務(wù)器能夠做的惟一的一件事就是等待這些調(diào)用，而其他端點(diǎn)則無法訪問。

@GET
@Path("/genre/{name}")
@Produces(MediaType.APPLICATION_JSON)
public Response getGenre(@PathParam("name") String genreName) {
	Genre genre = potentiallyVerySlowSynchronousCall(genreName);
	return Response.ok(genre).build();
}

解決這個(gè)問題最簡單的方法是提交代碼，它將阻塞調(diào)用變成一個(gè)線程池：

@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）通常用于存儲(chǔ)單個(gè)任務(wù)的特定值。例如，在web應(yīng)用程序中，它可能為每個(gè)請求存儲(chǔ)一個(gè)請求id和一個(gè)用戶id，因此MDC查找與單個(gè)請求或整個(gè)用戶活動(dòng)相關(guān)的日志記錄變得更加容易。

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

可是如果代碼的某些部分是在專用線程池中執(zhí)行的，則線程（提交任務(wù)的線程）中MDC就不會(huì)被繼續(xù)傳值。在下面的示例中，第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．更改線程名稱

為了簡化日志讀取和線程轉(zhuǎn)儲(chǔ)，可以自定義線程的名稱。這可以通過創(chuàng)建ExecutorService時(shí)用一個(gè)ThreadFactory來完成。在流行的實(shí)用程序庫中有許多ThreadFactory接口的實(shí)現(xiàn):

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);

將線程轉(zhuǎn)儲(chǔ)與默認(rèn)命名進(jìn)行比較：

"pool-1-thread-3" #14 prio=5 os_prio=31 tid=0x00007fc06b19f000 nid=0x5703 runnable [0x0000700001ff9000]
 java.lang.Thread.State: RUNNABLE
at com.github.sorokinigor.article.tipsaboutconcurrency.setthreadsname.TaskHandler.compute(TaskHandler.java:16)
...
"pool-2-thread-3" #15 prio=5 os_prio=31 tid=0x00007fc06aa10800 nid=0x5903 runnable [0x00007000020fc000]
 java.lang.Thread.State: RUNNABLE
at com.github.sorokinigor.article.tipsaboutconcurrency.setthreadsname.HealthCheckCallback.recordFailure(HealthChecker.java:21)
at com.github.sorokinigor.article.tipsaboutconcurrency.setthreadsname.HealthChecker.check(HealthChecker.java:9)
...
"pool-1-thread-2" #12 prio=5 os_prio=31 tid=0x00007fc06aa10000 nid=0x5303 runnable [0x0000700001df3000]
 java.lang.Thread.State: RUNNABLE
at com.github.sorokinigor.article.tipsaboutconcurrency.setthreadsname.TaskHandler.compute(TaskHandler.java:16)
 ...

與自定義命名進(jìn)行比較：

"task-handler-thread-1" #14 prio=5 os_prio=31 tid=0x00007fb49c9df000 nid=0x5703 runnable [0x000070000334a000]
 java.lang.Thread.State: RUNNABLE
at com.github.sorokinigor.article.tipsaboutconcurrency.setthreadsname.TaskHandler.compute(TaskHandler.java:16)
...
"authentication-service-ping-thread-0" #15 prio=5 os_prio=31 tid=0x00007fb49c9de000 nid=0x5903 runnable [0x0000700003247000]
 java.lang.Thread.State: RUNNABLE
at com.github.sorokinigor.article.tipsaboutconcurrency.setthreadsname.HealthCheckCallback.recordFailure(HealthChecker.java:21)
at com.github.sorokinigor.article.tipsaboutconcurrency.setthreadsname.HealthChecker.check(HealthChecker.java:9)
...
"task-handler-thread-0" #12 prio=5 os_prio=31 tid=0x00007fb49b9b5000 nid=0x5303 runnable [0x0000700003144000]
 java.lang.Thread.State: RUNNABLE
at com.github.sorokinigor.article.tipsaboutconcurrency.setthreadsname.TaskHandler.compute(TaskHandler.java:16)
 ...

想象一下，可能會(huì)不止3個(gè)線程。

5．使用LongAdder計(jì)數(shù)器

在高競爭的情況下，會(huì)采用java.util.concurrent.atomic.LongAdder進(jìn)行計(jì)數(shù)，而不會(huì)采用AtomicLong/AtomicInteger。

LongAdder可以跨越多個(gè)單元間仍保持值不變，但是如果需要的話，也可以增加它們的值，但與父類AtomicXX比較，這會(huì)導(dǎo)致更高的吞吐量，也會(huì)增加內(nèi)存消耗。

LongAdder counter = new LongAdder();
counter.increment();
...
long currentValue = counter.sum();

總結(jié)

以上就是本文關(guān)于5個(gè)并發(fā)處理技巧代碼示例的全部內(nèi)容，希望對大家有所幫助。感興趣的朋友可以繼續(xù)參閱本站：Java并發(fā)編程Semaphore計(jì)數(shù)信號量詳解、優(yōu)化Tomcat配置（內(nèi)存、并發(fā)、緩存等方面）方法詳解等，有什么問題可以隨時(shí)留言，小編會(huì)及時(shí)回復(fù)大家的。感謝朋友們對本站的支持！

您可能感興趣的文章: