【译】Java 并发处理的5种常见方法转载
1.永远不要忽略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;
}
}
}
代码的问题是不可能终止线程,因为它正在等待队列中的新元素,所以中断标志永远不会恢复:
-
正在运行代码的线程被中断。
-
BlockingQueue#poll() 抛出 InterruptedException 并清除中断标志。
-
当标志被清除时, while 循环条件 (
!Thread.currentThread().isInterrupted()
) 为ture。
为防止这种行为,请始终读取 InterruptedException 并在方法显式(通过声明 throwing InterruptedException)或隐式(通过声明/抛出原始 Exception)抛出时恢复中断标志:
<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. 注意使用专用的执行器进行阻塞操作
开发人员通常不希望因为一个“慢动作”而使整个服务器无响应。不幸的是,对于 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(映射诊断上下文)通常用于存储单个任务的特定值。例如,在 Web 应用程序中,它可能为每个请求存储一个请求 ID 和一个用户 ID,因此 MDC 使查找与单个请求或整个用户活动相关的日志条目变得更加容易。
不幸的是,如果代码的某些部分在专用线程池中执行,则来自提交任务的线程的 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<String, String> mdcCopy = MDC.getCopyOfContextMap();
executorService.submit(() -> {
MDC.setContextMap(mdcCopy);
try {
function.run();
} finally {
MDC.clear();
}
});
}
4.关于线程的重命名
自定义线程名称以简化读取日志和线程转储。这可以通过 在创建ExecutorService期间传递ThreadFactory来完成。流行的实用程序库中有很多 ThreadFactory接口的实现:
- com.google.common.util.concurrent.ThreadFactoryBuilder在番石榴中。
- 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);
只需将线程转储与默认名称进行比较:
对于名称有特指的线程:
并想象可能有超过 3 个线程。
5. 使用 LongAdder 作为计数器
考虑使用 java.util.concurrent.atomic.LongAdder而不是 AtomicLong / AtomicInteger 用于高争用的计数器。LongAdder维护多个单元的值并在需要时增加它们的数量,这会导致更高的吞吐量,但与 AtomicXX 系列类相比,也会导致更高的内存消耗。
LongAdder counter = new LongAdder();
counter.increment();
...
long currentValue = counter.sum();