商品详情页系统的Servlet3异步化实践

标签: 商品 系统 servlet3 | 发表时间:2015-09-24 20:16 | 作者:
出处:http://www.iteye.com

在京东工作的这一年多时间里,我在整个商品详情页系统(后端数据源)及商品详情页统一服务系统(页面中异步加载的很多服务,如库存服务、图书相关服务、延保服务等)中使用了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

          // 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的使用,可以参考我之前的博客:

   Servlet3.1规范(最终版)中文版下载 

   Servlet3.1学习示例

 

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

 

1、接收请求 

    @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、业务线程池封装

    public void submitFuture(final HttpServletRequest req, final Callable<Object> task) {
        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);
        }
        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<Object> future = (CompletableFuture<Object>)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;
            }
        });
    }
    private void callBack(AsyncContext asyncContext, Object result, String uri, Map<String, String[]> 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();
        }
    }

 

  

线程池的初始化

    @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<Runnable>(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的配置

<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、使用同步方式压测

siege-3.0.7]# ./src/siege -c100 -t60s -b http://***.item.jd.com/981821 
Transactions: 279187 hits
Availability: 100.00 %
Elapsed time: 59.33 secs
Data transferred: 1669.41 MB
Response time: 0.02 secs
Transaction rate: 4705.66 trans/sec
Throughput: 28.14 MB/sec
Concurrency: 99.91
Successful transactions: 279187
Failed transactions: 0
Longest transaction: 1.04
Shortest transaction: 0.00

 

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

siege-3.0.7]# ./src/siege -c100 -t60s -b http://***.item.jd.com/981821 .
Transactions: 337998 hits
Availability: 100.00 %
Elapsed time: 59.09 secs
Data transferred: 2021.07 MB
Response time: 0.03 secs
Transaction rate: 5720.05 trans/sec
Throughput: 34.20 MB/sec
Concurrency: 149.79
Successful transactions: 337998
Failed transactions: 0
Longest transaction: 1.07
Shortest transaction: 0.00

 

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

siege-3.0.7]# ./src/siege -c600 -t60s -b http://***.item.jd.com/981821 
Transactions: 370985 hits
Availability: 100.00 %
Elapsed time: 59.16 secs
Data transferred: 2218.32 MB
Response time: 0.10 secs
Transaction rate: 6270.88 trans/sec
Throughput: 37.50 MB/sec
Concurrency: 598.31
Successful transactions: 370985
Failed transactions: 0
Longest transaction: 1.32
Shortest transaction: 0.00

 

  

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

 

 

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

 

 

Servlet3相关资料

  Servlet3.1规范(最终版)中文版下载 

  Servlet3.1学习示例

 



已有 0 人发表留言,猛击->> 这里<<-参与讨论


ITeye推荐



相关 [商品 系统 servlet3] 推荐:

商品详情页系统的Servlet3异步化实践

- - ITeye博客
在京东工作的这一年多时间里,我在整个商品详情页系统(后端数据源)及商品详情页统一服务系统(页面中异步加载的很多服务,如库存服务、图书相关服务、延保服务等)中使用了Servlet3请求异步化模型,总结了Servlet3请求异步化的一些经验和想法跟大家分享和交流. 我将从如下几点阐述Servlet3异步化之后的好处:.

spring mvc使用Servlet3异步要注意的几个问题

- - 企业架构 - ITeye博客
1、注意添加 true. 在web.xml中对DispatcherServlet添加true. 2、如果集成了shiro一定要注意在mapping中增加dispatcher项,否则会抛出org.apache.shiro.UnavailableSecurityManagerException异常.

商品搜索引擎—推荐系统设计

- - ImportNew
结合目前已存在的商品推荐设计(如淘宝、京东等),推荐系统主要包含系统推荐和个性化推荐两个模块. 系统推荐: 根据大众行为的推荐引擎,对每个用户都给出同样的推荐,这些推荐可以是静态的由系统管理员人工设定的,或者基于系统所有用户的反馈统计计算出的当下比较流行的物品. 个性化推荐:对不同的用户,根据他们的口味和喜好给出更加精确的推荐,这时,系统需要了解需推荐内容和用户的特质,或者基于社会化网络,通过找到与当前用户相同喜好的用户,实现推荐.

5-机器学习启蒙- 商品推荐系统1

- - 神刀安全网
有大量的商品和用户,想要推荐一部分商品给用户. 怎么通过机器学习结合你和别人的历史购物记录做出适合你的推荐. 亚马逊重点关注商品推荐,另一个推荐系统流行的例子是2006-2009. 年主办的比赛,100万美金奖励推荐电影系统. 我们在哪里能见到推荐系统. 来看一些推荐系统起到重要作用的领域. 个性化正在改变我们关于世界的经验.

电商商品系统的演进分析

- - 掘金 架构
好久没有给大家带来新的知识分享了,2022的第一篇(是的,你没看错. 第一篇)就给大家讲讲商品模型的演进过程吧. 希望对大家做的工作有所帮助~~. 我们来假设:现在需要从零开始做电商,毫无经验,也没有竞对可参考,你就是全球独一份. 站在技术的视角,要去做一个电商商品系统,毫无疑问你需要一个商品实体,同时为了方便用户在C端筛选浏览,继承自CMS思想(栏目-->文章模型),很容易想到给每类商品增加一个分类,把相似的商品归属到相同的类目下.

壳系统

- Vernsu - It Talks-魏武挥的blog
经常有人被我问到“你用什么浏览器”时的答案是:傲游啦360啦,但事实上,这些都不是真正的浏览器,从技术角度讲,充其量只是在IE浏览器上加一个壳罢了. 在国外,壳浏览器是以“皮肤”的形式存在,纯属为了美化浏览器而用. 但在中国,壳浏览器成了一门生意. 奇虎的主要收入来源并非来自那个由于一场商战而赫赫有名的安全卫士,而是来自于360浏览器(它有两个版本,分别以IE和Chrome为内核).

秒杀系统

- - 开源软件 - ITeye博客
秒杀系统架构分析与实战. (反馈非常好的文章,推荐). (1)查询商品;(2)创建订单;(3)扣减库存;(4)更新订单;(5)付款;(6)卖家发货. (1)低廉价格;(2)大幅推广;(3)瞬时售空;(4)一般是定时上架;(5)时间短、瞬时并发量高;. 假设某网站秒杀活动只推出一件商品,预计会吸引1万人参加活动,也就说最大并发请求数是10000,秒杀系统需要面对的技术挑战有:.

Ext文件系统

- Haides - 博客园-首页原创精华区
  虽然从Ext2到Ext4,找数据的方式发生了变化,但是,磁盘的布局还是非常相似的. 其实这个东西也不需要变化,因为现在也没什么特别巧妙的方式,而且磁盘的吞吐量、效率的瓶颈也不在这里. 当然,这里排除那些根据自身文件特点设计的数据库,毕竟还是为了支持通用文件.   Boot在第一个块,放的应该是引导程序,超级块就放在了第二个块上,如果不是可以在mount的时候通过参数sb来设置.

HBase 系统架构

- - 博客园_首页
HBase是Apache Hadoop的数据库,能够对大型数据提供随机、实时的读写访问. HBase的目标是存储并处理大型的数据. HBase是一个开源的,分布式的,多版本的,面向列的存储模型. 5 可在廉价PC Server搭建大规模结构化存储集群. HBase是Google BigTable的开源实现,其相互对应如下:.

Linux系统监控

- - CSDN博客系统运维推荐文章
查看所有的进程和端口使用情况:. 查看nginx并发(连接数)进程数:. 查看当网络连接状态中,已建立连接的数量:. 查看系统tcp连接中各个状态的连接数. 输出每个ip的连接数,以及总的各个状态的连接数. df -hl 查看磁盘使用情况 . df -hl 查看磁盘剩余空间. df -h 查看每个根路径的分区大小.