博客
关于我
强烈建议你试试无所不能的chatGPT，快点击我
商品详情页系统的Servlet3异步化实践
阅读量：5942 次
发布时间：2019-06-19

本文共 16584 字，大约阅读时间需要 55 分钟。

在京东工作的这一年多时间里，我在整个商品详情页系统（后端数据源）及商品详情页统一服务系统（页面中异步加载的很多服务，如库存服务、图书相关服务、延保服务等）中使用了Servlet3请求异步化模型，总结了Servlet3请求异步化的一些经验和想法跟大家分享和交流。

 

我将从如下几点阐述Servlet3异步化之后的好处：

1、为什么实现请求异步化需要使用Servlet3

2、请求异步化后得到的好处是什么

3、如何使用Servlet3异步化

4、一些Servlet3异步化压测数据

 

首先说下一般http请求处理的流程：

1、容器负责接收并解析请求为HttpServletRequest；

2、然后交给Servlet进行业务处理；

3、最后通过HttpServletResponse写出响应。

 

在Servlet2.x规范中，所有这些处理都是同步进行的，也就是说必须在一个线程中完成从接收请求、业务处理到响应。

 

1、为什么实现请求异步化需要使用Servlet3

此处已Tomcat6举例子，Tomcat6没有实现Servlet3规范，它在处理请求时是通过如下方式实现的：

org.apache.catalina.connector.CoyoteAdapter#service

Java代码  
  1. // Recycle the wrapper request and response  
  2.   if (!comet) {  
  3.       request.recycle();    
  4.       response.recycle();  
  5.   } else {  
  6.       // Clear converters so that the minimum amount of memory  
  7.       // is used by this processor  
  8.       request.clearEncoders();  
  9.       response.clearEncoders();  
  10.   }  
// Recycle the wrapper request and response            if (!comet) {                request.recycle();                  response.recycle();            } else {                // Clear converters so that the minimum amount of memory                // is used by this processor                request.clearEncoders();                response.clearEncoders();            }

在请求结束时会同步进行请求的回收，也就是说请求解析、业务处理和响应必须在一个线程内完成，不能跨越线程界限。

 

这也就说明了必须使用实现了Servlet规范的容器进行处理，如Tomcat 7.x。

 

2、请求异步化后得到的好处是什么

2.1、更高的并发能力；

2.2、请求解析和业务处理线程池分离；

2.3、根据业务重要性对业务分级，并分级线程池；

2.4、对业务线程池进行监控、运维、降级等处理。

 

2.1、更高的并发能力

得益于技术的升级，在JDK7配合Tomcat7压测中获得了不错的性能表现。

 

2.2、请求解析和业务处理线程池分离

在引入Servlet3之前我们的线程模型是如下样子的：

 
整个请求解析、业务处理、生成响应都是由Tomcat线程池进行处理，而且都是在一个线程中处理；不能分离线程处理；比如接收到请求后交给其他线程处理，这样不能灵活定义业务处理模型。

 

引入Servlet3之后，我们的线程模型可以改造为如下样子：

 

此处可以看到请求解析使用Tomcat单线程；而解析完成后会扔到业务队列中，由业务线程池进行处理；这种处理方式可以得到如下好处：

1、根据业务重要性对业务进行分级，然后根据分级定义线程池；

2、可以拿到业务线程池，可以进行很多的操作，比如监控、降级等。

 

2.3、根据业务重要性对业务分级，并分级线程池

在一个系统的发展期间，我们一般把很多服务放到一个系统中进行处理，比如库存服务、图书相关服务、延保服务等等；这些服务中我们可以根据其重要性对业务分级别和做一些限制：

1、可以把业务分为核心业务级别和非核心业务级别；

2、为不同级别的业务定义不同的线程池，线程池之间是隔离的；

3、根据业务量定义各级别线程池大小。

 

此时假设非核心业务因为数据库连接池或者网络问题抖动造成响应时间慢，不会对我们核心业务产生影响。

 

2.4、对业务线程池进行监控、运维、降级等处理

因为业务线程池从Tomcat中分离出来，可以进行线程池的监控，比如查看当前处理的请求有多少，是否到了负载瓶颈，到了瓶颈后可以进行业务报警等处理。

上图是我们一个简陋的监控图，可实时查看到当前处理情况：正在处理的任务有多少，队列中等待的任务有多少；可以根据这些数据进行监控和预警。

 

另外我们还可以进行一些简单的运维：

对业务线程池进行扩容，或者业务出问题时立即清空线程池防止容器崩溃等问题；而不需要等待容器重启（容器重启需要耗费数十秒甚至数几十毫秒、而且启动后会有预热问题，而造成业务产生抖动）。

 

如果发现请求处理不过来，负载比较高，最简单的办法就是直接清空线程池，将老请求拒绝掉，而没有雪崩效应。

 

 

因为业务队列和业务线程池都是自己的，可以对这些基础组件做很多处理，比如定制业务队列，按照用户级别对用户请求排序，高级别用户得到更高优先级的业务处理。

 

3、如何使用Servlet3异步化

对于Servlet3的使用，可以参考我之前的博客：

   

  

 

而在我项目里使用就比较简单：

 

1、接收请求 

Java代码  
  1. @RequestMapping("/book")  
  2. public void getBook(HttpServletRequest request, @RequestParam(value="skuId") final Long skuId,  
  3.                     @RequestParam(value="cat1") final Integer cat1, @RequestParam(value="cat2") final Integer cat2) throws Exception {  
  4.   
  5.     oneLevelAsyncContext.submitFuture(request, () -> bookService.getBook(skuId, cat1, cat2));  
  6. }  
@RequestMapping("/book")    public void getBook(HttpServletRequest request, @RequestParam(value="skuId") final Long skuId,                        @RequestParam(value="cat1") final Integer cat1, @RequestParam(value="cat2") final Integer cat2) throws Exception {        oneLevelAsyncContext.submitFuture(request, () -> bookService.getBook(skuId, cat1, cat2));    }

通过一级业务线程池接收请求，并提交业务处理到该线程池；  

 

 

2、业务线程池封装

Java代码  
  1. public void submitFuture(final HttpServletRequest req, final Callable<Object> task) {  
  2.     final String uri = req.getRequestURI();  
  3.     final Map<String, String[]> params = req.getParameterMap();  
  4.     final AsyncContext asyncContext = req.startAsync();  //开启异步上下文  
  5.     asyncContext.getRequest().setAttribute("uri", uri);  
  6.     asyncContext.getRequest().setAttribute("params", params);  
  7.     asyncContext.setTimeout(asyncTimeoutInSeconds * 1000);  
  8.     if(asyncListener != null) {  
  9.         asyncContext.addListener(asyncListener);  
  10.     }  
  11.     executor.submit(new CanceledCallable(asyncContext) { //提交任务给业务线程池  
  12.         @Override  
  13.         public Object call() throws Exception {  
  14.             Object o = task.call();  //业务处理调用  
  15.             if(o == null) {  
  16.                 callBack(asyncContext, o, uri, params);  //业务完成后，响应处理  
  17.             }  
  18.             if(o instanceof CompletableFuture) {  
  19.                 CompletableFuture<Object> future = (CompletableFuture<Object>)o;  
  20.                 future.thenAccept(resultObject -> callBack(asyncContext, resultObject, uri, params))  
  21.                 .exceptionally(e -> {  
  22.                     callBack(asyncContext, "", uri, params);  
  23.                     return null;  
  24.                 });  
  25.             } else if(o instanceof String) {  
  26.                 callBack(asyncContext, o, uri, params);  
  27.             }  
  28.             return null;  
  29.         }  
  30.     });  
  31. }  
public void submitFuture(final HttpServletRequest req, final Callable task) {        final String uri = req.getRequestURI();        final Map
       
         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);        }        executor.submit(new CanceledCallable(asyncContext) { //提交任务给业务线程池            @Override            public Object call() throws Exception {                Object o = task.call();  //业务处理调用                if(o == null) {                    callBack(asyncContext, o, uri, params);  //业务完成后，响应处理                }                if(o instanceof CompletableFuture) {                    CompletableFuture
         future = (CompletableFuture)o;                    future.thenAccept(resultObject -> callBack(asyncContext, resultObject, uri, params))                    .exceptionally(e -> {                        callBack(asyncContext, "", uri, params);                        return null;                    });                } else if(o instanceof String) {                    callBack(asyncContext, o, uri, params);                }                return null;            }        });    }
Java代码  
  1. private void callBack(AsyncContext asyncContext, Object result, String uri, Map<String, String[]> params) {  
  2.     HttpServletResponse resp = (HttpServletResponse) asyncContext.getResponse();  
  3.     try {  
  4.         if(result instanceof String) {  
  5.             write(resp, (String)result);  
  6.         } else {  
  7.             write(resp, JSONUtils.toJSON(result));  
  8.         }  
  9.     } catch (Throwable e) {  
  10.         resp.setStatus(HttpServletResponse.SC_INTERNAL_SERVER_ERROR); //程序内部错误  
  11.         try {  
  12.             LOG.error("get info error, uri : {},  params : {}", uri, JSONUtils.toJSON(params), e);  
  13.         } catch (Exception ex) {  
  14.         }  
  15.     } finally {  
  16.         asyncContext.complete();  
  17.     }  
  18. }  
private void callBack(AsyncContext asyncContext, Object result, String uri, Map
      
        params) {        HttpServletResponse resp = (HttpServletResponse) asyncContext.getResponse();        try {            if(result instanceof String) {                write(resp, (String)result);            } else {                write(resp, JSONUtils.toJSON(result));            }        } catch (Throwable e) {            resp.setStatus(HttpServletResponse.SC_INTERNAL_SERVER_ERROR); //程序内部错误            try {                LOG.error("get info error, uri : {},  params : {}", uri, JSONUtils.toJSON(params), e);            } catch (Exception ex) {            }        } finally {            asyncContext.complete();        }    }

 

  

线程池的初始化

Java代码  
  1. @Override  
  2. public void afterPropertiesSet() throws Exception {  
  3.     String[] poolSizes = poolSize.split("-");  
  4.     //初始线程池大小  
  5.     int corePoolSize = Integer.valueOf(poolSizes[0]);  
  6.     //最大线程池大小  
  7.     int maximumPoolSize = Integer.valueOf(poolSizes[1]);  
  8.     queue = new LinkedBlockingDeque<Runnable>(queueCapacity);  
  9.     executor = new ThreadPoolExecutor(  
  10.             corePoolSize, maximumPoolSize,  
  11.             keepAliveTimeInSeconds, TimeUnit.SECONDS,  
  12.             queue);  
  13.   
  14.     executor.allowCoreThreadTimeOut(true);  
  15.     executor.setRejectedExecutionHandler(new RejectedExecutionHandler() {  
  16.         @Override  
  17.         public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {  
  18.             if(r instanceof CanceledCallable) {  
  19.                 CanceledCallable cc = ((CanceledCallable) r);  
  20.                 AsyncContext asyncContext = cc.asyncContext;  
  21.                 if(asyncContext != null) {  
  22.                     try {  
  23.                         String uri = (String) asyncContext.getRequest().getAttribute("uri");  
  24.                         Map params = (Map) asyncContext.getRequest().getAttribute("params");  
  25.                         LOG.error("async request rejected, uri : {}, params : {}", uri, JSONUtils.toJSON(params));  
  26.                     } catch (Exception e) {}  
  27.                     try {  
  28.                         HttpServletResponse resp = (HttpServletResponse) asyncContext.getResponse();  
  29.                         resp.setStatus(HttpServletResponse.SC_INTERNAL_SERVER_ERROR);  
  30.                     } finally {  
  31.                         asyncContext.complete();  
  32.                     }  
  33.                 }  
  34.             }  
  35.         }  
  36.     });  
  37.   
  38.     if(asyncListener == null) {  
  39.         asyncListener = new AsyncListener() {  
  40.             @Override  
  41.             public void onComplete(AsyncEvent event) throws IOException {  
  42.             }  
  43.   
  44.             @Override  
  45.             public void onTimeout(AsyncEvent event) throws IOException {  
  46.                 AsyncContext asyncContext = event.getAsyncContext();  
  47.                 try {  
  48.                     String uri = (String) asyncContext.getRequest().getAttribute("uri");  
  49.                     Map params = (Map) asyncContext.getRequest().getAttribute("params");  
  50.                     LOG.error("async request timeout, uri : {}, params : {}", uri, JSONUtils.toJSON(params));  
  51.                 } catch (Exception e) {}  
  52.                 try {  
  53.                     HttpServletResponse resp = (HttpServletResponse) asyncContext.getResponse();  
  54.                     resp.setStatus(HttpServletResponse.SC_INTERNAL_SERVER_ERROR);  
  55.                 } finally {  
  56.                     asyncContext.complete();  
  57.                 }  
  58.             }  
  59.   
  60.             @Override  
  61.             public void onError(AsyncEvent event) throws IOException {  
  62.                 AsyncContext asyncContext = event.getAsyncContext();  
  63.                 try {  
  64.                     String uri = (String) asyncContext.getRequest().getAttribute("uri");  
  65.                     Map params = (Map) asyncContext.getRequest().getAttribute("params");  
  66.                     LOG.error("async request error, uri : {}, params : {}", uri, JSONUtils.toJSON(params));  
  67.                 } catch (Exception e) {}  
  68.                 try {  
  69.                     HttpServletResponse resp = (HttpServletResponse) asyncContext.getResponse();  
  70.                     resp.setStatus(HttpServletResponse.SC_INTERNAL_SERVER_ERROR);  
  71.                 } finally {  
  72.                     asyncContext.complete();  
  73.                 }  
  74.             }  
  75.   
  76.             @Override  
  77.             public void onStartAsync(AsyncEvent event) throws IOException {  
  78.   
  79.             }  
  80.         };  
  81.     }  
  82.   
  83. }  
@Override    public void afterPropertiesSet() throws Exception {        String[] poolSizes = poolSize.split("-");        //初始线程池大小        int corePoolSize = Integer.valueOf(poolSizes[0]);        //最大线程池大小        int maximumPoolSize = Integer.valueOf(poolSizes[1]);        queue = new LinkedBlockingDeque
      
       (queueCapacity);        executor = new ThreadPoolExecutor(                corePoolSize, maximumPoolSize,                keepAliveTimeInSeconds, TimeUnit.SECONDS,                queue);        executor.allowCoreThreadTimeOut(true);        executor.setRejectedExecutionHandler(new RejectedExecutionHandler() {            @Override            public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {                if(r instanceof CanceledCallable) {                    CanceledCallable cc = ((CanceledCallable) r);                    AsyncContext asyncContext = cc.asyncContext;                    if(asyncContext != null) {                        try {                            String uri = (String) asyncContext.getRequest().getAttribute("uri");                            Map params = (Map) asyncContext.getRequest().getAttribute("params");                            LOG.error("async request rejected, uri : {}, params : {}", uri, JSONUtils.toJSON(params));                        } catch (Exception e) {}                        try {                            HttpServletResponse resp = (HttpServletResponse) asyncContext.getResponse();                            resp.setStatus(HttpServletResponse.SC_INTERNAL_SERVER_ERROR);                        } finally {                            asyncContext.complete();                        }                    }                }            }        });        if(asyncListener == null) {            asyncListener = new AsyncListener() {                @Override                public void onComplete(AsyncEvent event) throws IOException {                }                @Override                public void onTimeout(AsyncEvent event) throws IOException {                    AsyncContext asyncContext = event.getAsyncContext();                    try {                        String uri = (String) asyncContext.getRequest().getAttribute("uri");                        Map params = (Map) asyncContext.getRequest().getAttribute("params");                        LOG.error("async request timeout, uri : {}, params : {}", uri, JSONUtils.toJSON(params));                    } catch (Exception e) {}                    try {                        HttpServletResponse resp = (HttpServletResponse) asyncContext.getResponse();                        resp.setStatus(HttpServletResponse.SC_INTERNAL_SERVER_ERROR);                    } finally {                        asyncContext.complete();                    }                }                @Override                public void onError(AsyncEvent event) throws IOException {                    AsyncContext asyncContext = event.getAsyncContext();                    try {                        String uri = (String) asyncContext.getRequest().getAttribute("uri");                        Map params = (Map) asyncContext.getRequest().getAttribute("params");                        LOG.error("async request error, uri : {}, params : {}", uri, JSONUtils.toJSON(params));                    } catch (Exception e) {}                    try {                        HttpServletResponse resp = (HttpServletResponse) asyncContext.getResponse();                        resp.setStatus(HttpServletResponse.SC_INTERNAL_SERVER_ERROR);                    } finally {                        asyncContext.complete();                    }                }                @Override                public void onStartAsync(AsyncEvent event) throws IOException {                }            };        }    }

 

 

3、业务处理

执行bookService.getBook(skuId, cat1, cat2)进行业务处理。

 

4、返回响应

在之前封装的异步线程池上下文中直接返回。

 

5、Tomcat server.xml的配置

Java代码  
  1. <Connector port="1601" asyncTimeout="10000" acceptCount="10240" maxConnections="10240" acceptorThreadCount="1"  minSpareThreads="1" maxThreads="1" redirectPort="8443" processorCache="1024" URIEncoding="UTF-8" protocol="org.apache.coyote.http11.Http11NioProtocol" enableLookups="false"/>  

我们升级到了jdk1.8.0_51 +tomcat 8.0.26，在使用Http11Nio2Protocol时遇到一些问题，暂时还是使用的Http11Nio1Protocol。此处可以看到Tomcat线程池我们配置了maxThreads=1，即一个线程进行请求解析。

 

 

 

4、一些Servlet3异步化压测数据

压测机器基本环境：32核CPU、32G内存；jdk1.7.0_71 + tomcat 7.0.57，服务响应时间在20ms+，使用最简单的单个URL压测吞吐量：

 

1、使用同步方式压测

Java代码  
  1. siege-3.0.7]# ./src/siege -c100 -t60s -b http://***.item.jd.com/981821   
  2. Transactions: 279187 hits  
  3. Availability: 100.00 %  
  4. Elapsed time: 59.33 secs  
  5. Data transferred: 1669.41 MB  
  6. Response time: 0.02 secs  
  7. Transaction rate: 4705.66 trans/sec  
  8. Throughput: 28.14 MB/sec  
  9. Concurrency: 99.91  
  10. Successful transactions: 279187  
  11. Failed transactions: 0  
  12. Longest transaction: 1.04  
  13. Shortest transaction: 0.00  
siege-3.0.7]# ./src/siege -c100 -t60s -b http://***.item.jd.com/981821 Transactions: 279187 hitsAvailability: 100.00 %Elapsed time: 59.33 secsData transferred: 1669.41 MBResponse time: 0.02 secsTransaction rate: 4705.66 trans/secThroughput: 28.14 MB/secConcurrency: 99.91Successful transactions: 279187Failed transactions: 0Longest transaction: 1.04Shortest transaction: 0.00

 

2.1、 使用Servlet3异步化压测 100并发、60秒：

Java代码  
  1. siege-3.0.7]# ./src/siege -c100 -t60s -b http://***.item.jd.com/981821 .  
  2. Transactions: 337998 hits  
  3. Availability: 100.00 %  
  4. Elapsed time: 59.09 secs  
  5. Data transferred: 2021.07 MB  
  6. Response time: 0.03 secs  
  7. Transaction rate: 5720.05 trans/sec  
  8. Throughput: 34.20 MB/sec  
  9. Concurrency: 149.79  
  10. Successful transactions: 337998  
  11. Failed transactions: 0  
  12. Longest transaction: 1.07  
  13. Shortest transaction: 0.00  
siege-3.0.7]# ./src/siege -c100 -t60s -b http://***.item.jd.com/981821 .Transactions: 337998 hitsAvailability: 100.00 %Elapsed time: 59.09 secsData transferred: 2021.07 MBResponse time: 0.03 secsTransaction rate: 5720.05 trans/secThroughput: 34.20 MB/secConcurrency: 149.79Successful transactions: 337998Failed transactions: 0Longest transaction: 1.07Shortest transaction: 0.00

 

2.2、使用Servlet3异步化压测 600并发、60秒：

Java代码  
  1. siege-3.0.7]# ./src/siege -c600 -t60s -b http://***.item.jd.com/981821   
  2. Transactions: 370985 hits  
  3. Availability: 100.00 %  
  4. Elapsed time: 59.16 secs  
  5. Data transferred: 2218.32 MB  
  6. Response time: 0.10 secs  
  7. Transaction rate: 6270.88 trans/sec  
  8. Throughput: 37.50 MB/sec  
  9. Concurrency: 598.31  
  10. Successful transactions: 370985  
  11. Failed transactions: 0  
  12. Longest transaction: 1.32  
  13. Shortest transaction: 0.00  
siege-3.0.7]# ./src/siege -c600 -t60s -b http://***.item.jd.com/981821 Transactions: 370985 hitsAvailability: 100.00 %Elapsed time: 59.16 secsData transferred: 2218.32 MBResponse time: 0.10 secsTransaction rate: 6270.88 trans/secThroughput: 37.50 MB/secConcurrency: 598.31Successful transactions: 370985Failed transactions: 0Longest transaction: 1.32Shortest transaction: 0.00

 

  

可以看出异步化之后吞吐量提升了，但是响应时间长了，也就是异步化并不会提升响应时间，但是会增加吞吐量和增加我们需要的灵活性。

 

 

通过异步化我们不会获得更快的响应时间，但是我们获得了整体吞吐量和我们需要的灵活性：请求解析和业务处理线程池分离；根据业务重要性对业务分级，并分级线程池；对业务线程池进行监控、运维、降级等处理。

 

 

Servlet3相关资料

  

 

 

原文地址：http://jinnianshilongnian.iteye.com/blog/2245925

转载地址：http://damtx.baihongyu.com/

你可能感兴趣的文章
不忘初心，努力做最好的自己
查看>>
走在浪潮尖端——为什么选择云计算
查看>>
重启nagios有异常提示Starting nagios:This account is currently not available
查看>>
全面、经典视频教程(php、java、jsp、ajax、linux...)
查看>>
Windows 7/Windows 8都有上帝模式
查看>>
华为云计算FusionCompute环境部署实验之使用批量部署工具安装
查看>>
静态路由的配置实验
查看>>
linux for循环
查看>>
关卡CyclicBarrier的使用
查看>>
用好这6个新功能，让你的项目设计事半功倍
查看>>
第一篇，试试功能如何
查看>>
成为男人眼中魅力女人的十大要素
查看>>
Python(四）IP代理
查看>>
louboutin france jambe allongement lignes
查看>>
TestBird频现国内手游“盛宴” 开发商互相介绍用得很赞
查看>>
以太坊PHP离线交易开发包
查看>>
linux运维人员必须熟悉的运维工具汇总
查看>>
在PHPStorm中支持ThinkPHP代码提示
查看>>
Linux文件压缩与归档
查看>>
Linux下LAMP环境的安装之编译安装过程(五)
查看>>
喝酒易醉，品茶养心，人生如梦，品茶悟道，何以解忧？唯有杜康！-- 愿君每日到此一游！
当前时间: 2024-12-28 15:17:18   当前IP: 18.220.134.161   联系邮箱:javaeecc@qq.com     Copyright © 2020 - 2022 baihongyu.com 京ICP备2021015314号-2
强烈建议你试试无所不能的CHAT-GPT，快点击我
强烈建议你试试无所不能的CHAT-GPT，快点击我