[京东技术]声明：本文转载自微信公众号“开涛的博客”，转载务必声明。
在做电商系统时，流量入口如首页、活动页、商品详情页等系统承载了网站的大部分流量，而这些系统的主要职责包括聚合数据拼装模板、热点统计、缓存、下游功能降级开关、托底数据等等。其中聚合数据需要调用其它多个系统服务获取数据、拼装数据/模板然后返回给前端，聚合数据来源主要有依赖系统/服务、缓存、数据库等；而系统之间的调用可以通过如http接口调用（如HttpClient）、SOA服务调用（如dubbo、thrift）等等。
在Java中，如使用Tomcat，一个请求会分配一个线程进行请求处理，该线程负责获取数据、拼装数据或模板然后返回给前端；在同步调用获取数据接口的情况下（等待依赖系统返回数据），整个线程是一直被占用并阻塞的。如果有大量的这种请求，每个请求占用一个线程，但线程一直处于阻塞，降低了系统的吞吐量，这将导致应用的吞吐量下降；我们希望在调用依赖的服务响应比较慢，此时应该让出线程和CPU来处理下一个请求，当依赖的服务返回了再分配相应的线程来继续处理。而这应该有更好的解决方案：异步/协程。而Java是不支持协程的（虽然有些Java框架说支持，但还是高层API的封装），因此在Java中我们还可以使用异步来提升吞吐量。目前java一些开源框架（HttpClient\HttpAsyncClient、dubbo、thrift等等）大部分都支持。
几种调用方式
同步阻塞调用
即串行调用，响应时间为所有服务的响应时间总和；
半异步(异步Future)
线程池，异步Future，使用场景：并发请求多服务，总耗时为最长响应时间；提升总响应时间，但是阻塞主请求线程，高并发时依然会造成线程数过多，CPU上下文切换；
全异步(Callback)
Callback方式调用，使用场景：不考虑回调时间且只能对结果做简单处理，如果依赖服务是两个或两个以上服务，则不能合并两个服务的处理结果；不阻塞主请求线程，但使用场景有限。
异步回调链式编排
异步回调链式编排（JDK8 CompletableFuture），使用场景：其实不是异步调用方式，只是对依赖多服务的Callback调用结果处理做结果编排，来弥补Callback的不足，从而实现全异步链式调用。
接下来看看如何设计利用全异步Callback调用和异步回调链式编排处理结果来实现全异步系统设计。
同步阻塞调用
public class Test {
public static void main(String[] args) throws Exception {
RpcService rpcService = new RpcService();
HttpService httpService = new HttpService();
//耗时10ms
Map<String, String> result1 = rpcService.getRpcResult();
//耗时20ms
Integer result2 = httpService.getHttpResult();
//总耗时30ms
}
static class RpcService {
Map<String, String> getRpcResult() throws Exception {
//调用远程方法（远程方法耗时约10ms，可以使用Thread.sleep模拟）
}
}
static class HttpService {
Integer getHttpResult() throws Exception {
//调用远程方法（远程方法耗时约20ms，可以使用Thread.sleep模拟）
Thread.sleep(20);
return 0;
}
}
}
半异步(异步Future)
public class Test {
final static ExecutorService executor = Executors.newFixedThreadPool(2);
public static void main(String[] args) {
RpcService rpcService = new RpcService();
HttpService httpService = new HttpService();
Future<Map<String, String>> future1 = null;
Future<Integer> future2 = null;
try {
future1 = executor.submit(() -> rpcService.getRpcResult());
future2 = executor.submit(() -> httpService.getHttpResult());
//耗时10ms
Map<String, String> result1 = future1.get(300, TimeUnit.MILLISECONDS);
//耗时20ms
Integer result2 = future2.get(300, TimeUnit.MILLISECONDS);
//总耗时20ms
} catch (Exception e) {
if (future1 != null) {
future1.cancel(true);
}
if (future2 != null) {
future2.cancel(true);
}
throw new RuntimeException(e);
}
}
static class RpcService {
Map<String, String> getRpcResult() throws Exception {
//调用远程方法（远程方法耗时约10ms，可以使用Thread.sleep模拟）
}
}
static class HttpService {
Integer getHttpResult() throws Exception {
//调用远程方法（远程方法耗时约20ms，可以使用Thread.sleep模拟）
}
}
}
全异步(Callback)
public class AsyncTest {
public staticHttpAsyncClient httpAsyncClient;
public static CompletableFuture<String> getHttpData(String url) {
CompletableFuture asyncFuture = new CompletableFuture();
HttpPost post = new HttpPost(url);
HttpAsyncRequestProducer producer = HttpAsyncMethods.create(post);
AsyncCharConsumer<HttpResponse> consumer = newAsyncCharConsumer<HttpResponse>() {
HttpResponse response;
protected HttpResponse buildResult(final HttpContext context) {
return response;
}
…...
};
FutureCallback callback = new FutureCallback<HttpResponse>() {
public void completed(HttpResponse response) {
asyncFuture.complete(EntityUtils.toString(response.getEntity()));
}
…...
};
httpAsyncClient.execute(producer, consumer, callback);
return asyncFuture;
}
public static void main(String[] args) throws Exception {
AsyncTest.getHttpData("网页链接);
Thread.sleep(1000000);
}
}
本示例使用HttpAsyncClient演示。
异步回调链式编排
CompletableFuture提供了50多个API，可以满足所需的各种场景的异步处理的编排，在此列举三个场景：
场景1：三个服务并发异步调用，返回CompletableFuture，不阻塞主线程；
方法test1：
public static void test1() throws Exception {
HelloClientDemoTest service = new HelloClientDemoTest();
/**
* 场景1 两个以上服务并发异步调用，返回CompletableFuture,不阻塞主线程
* 并且两个服务也是异步非阻塞调用
*/
CompletableFuture future1 = service.getHttpData("网页链接);
CompletableFuture future2 = service.getHttpData("网页链接);
CompletableFuture future3 =service.getHttpData("网页链接);
List<CompletableFuture> futureList = Lists.newArrayList(future1,future2, future3);
CompletableFuture<Void> allDoneFuture =CompletableFuture.allOf(futureList.toArray(newCompletableFuture[futureList.size()]));
CompletableFuture<String> future4 =allDoneFuture.thenApply(v -> {
List<Object> result =futureList.stream().map(CompletableFuture::join)
.collect(Collectors.toList());
//注意顺序
String result1 = (String)result.get(0);
String result2 = (String)result.get(1);
String result3 = (String)result.get(2);
//处理业务....
return result1 + result2 + result3;
}).exceptionally(e -> {
//e.printStackTrace();
return "";
});
//返回
}
场景2、两个服务并发异步调用，返回CompletableFuture，不阻塞主线程；
方法test2：
public void test2() throws Exception {
HelloClientDemoTest service = new HelloClientDemoTest();
/**
* 场景2 两个接口并发异步调用，返回CompletableFuture,不阻塞主线程
* 并且两个服务也是异步非阻塞调用
*/
CompletableFuture future1 = service.getHttpData("网页链接);
CompletableFuture future2 =service.getHttpData("网页链接);
CompletableFuture future3 =future1.thenCombine(future2, (f1, f2) -> {
//处理业务....
return f1 + "," + f2;
}).exceptionally(e -> {
return "";
});
//返回
}
场景3、两个服务，并发异步调用两个服务，并且一个服务的结果返回后再次调用另一服务，然后将三个结果后并处理，返回CompletableFuture，整个处理过程中不阻塞任何线程；
方法test3：
publicvoid test3() throws Exception {
HelloClientDemoTest service = new HelloClientDemoTest();
/**
* 场景3 两请求依赖调用，然后与另一服务结果组合处理，返回CompletableFuture,不阻塞主线程
* 并且两个服务也是异步非阻塞调用
*/
CompletableFuture future1 = service.getHttpData("网页链接);
CompletableFuture future2 = service.getHttpData("网页链接);
CompletableFuture<String> future3= future1.thenApply((param) -> {
CompletableFuture future4 =service.getHttpData("网页链接);
return future4;
});
CompletableFuture future5 =future2.thenCombine(future3, (f2, f3) -> {
//....处理业务
return f2 + "," + f3;
}).exceptionally(e -> {
return "";
});
//返回future5
}
全异步Web系统设计
主要技术：servlet3，JDK8 CompletableFuture，支持异步Callback调用的RPC框架。
先看一下处理流程图：
servlet3：Servlet 接收到请求之后，可能首先需要对请求携带的数据进行一些预处理；接着，Servlet 线程将请求转交给一个异步线程来执行业务处理，线程本身返回至容器。针对业务处理较耗时的情况，这将大大减少服务器资源的占用，并且提高并发处理速度。servlet3可参考商品详情页系统的Servlet3异步化实践，结合其中讲解的servlet3整合：
public void submitFuture(finalHttpServletRequest req, final Callable<CompletableFuture> task) throwsException{
final String uri = req.getRequestURI();
final Map<String, String[]> params = req.getParameterMap();
final AsyncContext asyncContext = req.startAsync();
asyncContext.getRequest().setAttribute("uri", uri);
asyncContext.getRequest().setAttribute("params", params);
asyncContext.setTimeout(asyncTimeoutInSeconds * 1000);
if(asyncListener != null) {
asyncContext.addListener(asyncListener);
}
CompletableFuture future = task.call();
future.thenAccept(result -> {
HttpServletResponse resp = (HttpServletResponse)asyncContext.getResponse();
try {
if(result instanceof String) {
byte[] bytes = new byte[0];
if (StringUtils.isBlank(result)){
resp.setContentType("text/html;charset=gbk");
resp.setContentLength(0);
} else {
bytes =result.getBytes("GBK");
}
//resp.setBufferSize(bytes.length);
resp.setContentType("text/html;charset=gbk");
if(StringUtils.isNotBlank(localIp)) {
resp.setHeader("t.ser", localIp);
}
resp.setContentLength(bytes.length);
resp.getOutputStream().write(bytes);
} else {
write(resp,JSONUtils.toJSON(result));
}
} catch (Throwable e) {
resp.setStatus(HttpServletResponse.SC_INTERNAL_SERVER_ERROR); //程序内部错误
try {
LOG.error("get infoerror, uri : {}, params : {}", uri,JSONUtils.toJSON(params), e);
} catch (Exception ex) {
}
} finally {
asyncContext.complete();
}
}).exceptionally(e -> {
asyncContext.complete();
return null;
});
}
另外还有Java中协程库Quasar，可参考《Java的纤程库 - Quasar》，目前没有在应用中使用并在测试FiberHttpServlet的时候遇到很多坑，日后把Quasar自如运用后形成日记，希望能结实更多的朋友一起研究，踩坑。
作者介绍：孙伟，目前负责京东商品详情页统一服务系统，写过java，写过ngx_lua，还写过storm等，喜欢学习研究新事物。
相关推荐
在Java编程中，同步、异步、阻塞和非阻塞是四...同步和阻塞适合处理简单的任务，而异步和非阻塞更适合需要高并发和低延迟的场景。理解这些概念，并结合Java提供的并发工具，能够帮助开发者编写出更加高效和可靠的程序。
这些客户端库同样采用了异步非阻塞I/O模型，以适应Redis的单线程模型。它们会将请求打包成队列，然后在一个后台线程中发送到Redis，避免阻塞主线程。 PPT.pptx可能包含更深入的Redis线程模型讲解，例如如何利用事件...
Netty 是一个高性能、异步事件驱动的网络应用框架，常用于开发高并发、低延迟的网络服务。在 Netty 中，Reactor 模型是其处理 I/O 事件的关键设计，它允许服务端高效地处理大量并发连接。在本文中，我们将探讨 ...
Java NIO（非阻塞I/O）是Java标准库中的一部分，从Java 1.4版本开始引入，为处理I/O操作提供了全新的机制。相比于传统的Java IO，NIO提供了更高效、灵活的I/O操作方式，特别适用于高并发、大数据量的场景。`java.nio...
Netty的非阻塞I/O模型有助于在高并发场景下保持低延迟和高吞吐量。 **项目结构与组件** 1. **用户模块**：实现用户注册、登录功能，通常会涉及到数据库操作，如MySQL，存储用户信息。 2. **聊天模块**：通过...
当用户访问`/async`端点时，`asyncTask()`方法将在后台线程中执行，而控制器立即返回"异步任务已启动"，这样就实现了非阻塞的异步处理。 然而，`@Async`还有一些高级用法，比如支持返回值、异常处理和异步方法的回...
当需要执行非I/O任务时，Reactor会将任务提交到队列，由专门的工作线程进行处理，确保了I/O线程不会被阻塞，从而维持高并发性能。 总的来说，Netty的Reactor模型通过高效地处理I/O事件和调度任务执行，实现了网络...
Netty的NIO（非阻塞I/O）模型使得服务器能够同时处理大量连接，显著提升了系统的并行处理能力。在即时通讯服务器中，Netty用于建立客户端与服务器之间的连接，处理各种网络事件，如连接建立、数据传输和断开连接等。...
为了避免这些问题，我们需要合理设计同步策略，如使用公平锁或非阻塞算法。 在Java 5之后引入的并发集合，如`ConcurrentHashMap`, `CopyOnWriteArrayList`等，为并发编程提供了线程安全的数据结构，减少了对显式...
- **EventLoop剖析**：深入EventLoop的工作原理，它是Netty高并发性能的关键。 - **accept和read流程**：解析接受新连接和处理读取数据的具体步骤，揭示了Netty如何高效处理网络I/O。 通过以上内容，我们可以全面...
考虑到并发性，服务器端可能需要多线程或者EventLoopGroup来处理多个连接，确保系统的高并发能力。 **安全性**也不能忽视。虽然这是一个学习项目，但实际应用中必须考虑用户认证、数据加密以及防止恶意攻击。例如，...
在Java领域，Spring WebFlux提供了一种处理HTTP请求的新方式，与传统的基于Servlet的Spring MVC不同，它支持异步、事件驱动的编程模型，旨在优化高并发场景下的性能。 在Spring WebFlux中，核心概念包括： 1. **...
Node.js利用JavaScript运行时环境，提供了异步、非阻塞I/O模型，使得处理高并发请求变得高效。在这个项目中，开发者可能利用了Express.js，一个流行的Node.js框架，来构建RESTful API，以便前端与后端进行通信。 ...