簡單介紹Java 并發(fā)處理的5種技巧

本文來自于HeapDump性能社區(qū)! !有性能問題,上HeapDump性能社區(qū)!

1.永遠(yuǎn)不要忽略InterruptedException

讓我們檢查以下代碼片段:

public class Task implements Runnable {
  private final BlockingQueue<String> queue = ...;

  @Override
  public void run() {
    while (!Thread.currentThread().isInterrupted()) {
      String result = getOrDefault(() -> queue.poll(1L, TimeUnit.MINUTES), "default");
      //do smth with the result
    }
  }

  <T> T getOrDefault(Callable<T> supplier, T defaultValue) {
    try {
      return supplier.call();
    } catch (Exception e) {
      logger.error("Got exception while retrieving value.", e);
      return defaultValue;
    }
  }
}

代碼的問題是不可能終止線程,因?yàn)樗诘却?duì)列中的新元素,所以中斷標(biāo)志永遠(yuǎn)不會(huì)恢復(fù):

正在運(yùn)行代碼的線程被中斷。
BlockingQueue#poll() 拋出 InterruptedException 并清除中斷標(biāo)志。
當(dāng)標(biāo)志被清除時(shí), while 循環(huán)條件 ( !Thread.currentThread().isInterrupted()) 為ture。
為防止這種行為,請(qǐng)始終讀取 InterruptedException 并在方法顯式(通過聲明 throwing InterruptedException)或隱式(通過聲明/拋出原始 Exception)拋出時(shí)恢復(fù)中斷標(biāo)志:

<T> T getOrDefault(Callable<T> 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í)行器進(jìn)行阻塞操作

開發(fā)人員通常不希望因?yàn)橐粋€(gè)“慢動(dòng)作”而使整個(gè)服務(wù)器無響應(yīng)。不幸的是,對(duì)于 RPC,響應(yīng)時(shí)間通常是不可預(yù)測(cè)的。
假設(shè)一臺(tái)服務(wù)器有 100 個(gè)工作線程,并且有一個(gè)端點(diǎn),它以 100 RPS 調(diào)用。它在內(nèi)部進(jìn)行 RPC 調(diào)用,通常需要 10 毫秒。在某個(gè)時(shí)間點(diǎn),這個(gè) RPC 的響應(yīng)時(shí)間變成了 2 秒,而服務(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();
}

解決問題的最簡單方法是將進(jìn)行阻塞調(diào)用的代碼提交到線程池:

@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(映射診斷上下文)通常用于存儲(chǔ)單個(gè)任務(wù)的特定值。例如,在 Web 應(yīng)用程序中,它可能為每個(gè)請(qǐng)求存儲(chǔ)一個(gè)請(qǐng)求 ID 和一個(gè)用戶 ID,因此 MDC 使查找與單個(gè)請(qǐng)求或整個(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ì)傳播。在以下示例中,第 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<String, String> mdcCopy = MDC.getCopyOfContextMap();
  executorService.submit(() -> {
    MDC.setContextMap(mdcCopy);
    try {
      function.run();
    } finally {
      MDC.clear();
    }
  });
}

4.關(guān)于線程的重命名

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

com.google.common.util.concurrent.ThreadFactoryBuilder in Guava。
Spring 中的org.springframework.scheduling.concurrent.CustomizableThreadFactory。
Apache Commons Lang 3 中的org.apache.commons.lang3.concurrent.BasicThreadFactory。
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)
    ...

對(duì)于名稱有特指的線程:

"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)
    ...

并想象可能有超過 3 個(gè)線程。

5. 使用 LongAdder 作為計(jì)數(shù)器

考慮使用 java.util.concurrent.atomic.LongAdder而不是 AtomicLong / AtomicInteger 用于高爭用的計(jì)數(shù)器。LongAdder維護(hù)多個(gè)單元的值并在需要時(shí)增加它們的數(shù)量,這會(huì)導(dǎo)致更高的吞吐量,但與 AtomicXX 系列類相比,也會(huì)導(dǎo)致更高的內(nèi)存消耗。

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

更多高并發(fā)案例:

高并發(fā)服務(wù)優(yōu)化篇:JVM--工程師進(jìn)階的必經(jīng)之路

淺談JDK并發(fā)包下面的分治思想及分治思想在高并發(fā)場(chǎng)景的運(yùn)用

這玩意兒也能算叫高并發(fā)?

面試官問我:什么是高并發(fā)下的請(qǐng)求合并?

架構(gòu)與思維:高并發(fā)下冪等性解決方案

高并發(fā)服務(wù)優(yōu)化篇:詳解一次由讀寫鎖引起的內(nèi)存泄漏

一次 HashSet 所引起的并發(fā)問題

?著作權(quán)歸作者所有,轉(zhuǎn)載或內(nèi)容合作請(qǐng)聯(lián)系作者
【社區(qū)內(nèi)容提示】社區(qū)部分內(nèi)容疑似由AI輔助生成,瀏覽時(shí)請(qǐng)結(jié)合常識(shí)與多方信息審慎甄別。
平臺(tái)聲明:文章內(nèi)容(如有圖片或視頻亦包括在內(nèi))由作者上傳并發(fā)布,文章內(nèi)容僅代表作者本人觀點(diǎn),簡書系信息發(fā)布平臺(tái),僅提供信息存儲(chǔ)服務(wù)。

相關(guān)閱讀更多精彩內(nèi)容

友情鏈接更多精彩內(nèi)容