前言
通过我之前的tomcat系列文章,相信看我博客的同学对tomcat应该有一个比较清晰的了解了,在前几篇博客我们讨论了tomcat在springboot框架中是如何启动的,讨论了tomcat的内部组件是如何设计以及请求是如何流转的,那么我们这边博客聊聊tomcat的异步servlet,tomcat是如何实现异步servlet的以及异步servlet的使用场景。
手撸一个异步的servlet
我们直接借助springboot框架来实现一个servlet,这里只展示servlet代码:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
|
@webservlet (urlpatterns = "/async" ,asyncsupported = true ) @slf4j public class asyncservlet extends httpservlet { executorservice executorservice =executors.newsinglethreadexecutor(); @override protected void doget(httpservletrequest req, httpservletresponse resp) throws servletexception, ioexception { //开启异步,获取异步上下文 final asynccontext ctx = req.startasync(); // 提交线程池异步执行 executorservice.execute( new runnable() { @override public void run() { try { log.info( "async service 准备执行了" ); //模拟耗时任务 thread.sleep(10000l); ctx.getresponse().getwriter().print( "async servlet" ); log.info( "async service 执行了" ); } catch (ioexception e) { e.printstacktrace(); } catch (interruptedexception e) { e.printstacktrace(); } //最后执行完成后完成回调。 ctx.complete(); } }); } |
上面的代码实现了一个异步的servlet,实现了 doget
方法注意在springboot中使用需要再启动类加上 @servletcomponentscan
注解来扫描servlet。既然代码写好了,我们来看看实际运行效果。
我们发送一个请求后,看到页面有响应,同时,看到请求时间花费了10.05s,那么我们这个servlet算是能正常运行啦。有同学肯定会问,这不是异步servlet吗?你的响应时间并没有加快,有什么用呢?对,我们的响应时间并不能加快,还是会取决于我们的业务逻辑,但是我们的异步servlet请求后,依赖于业务的异步执行,我们可以立即返回,也就是说,tomcat的线程可以立即回收,默认情况下,tomcat的核心线程是10,最大线程数是200,我们能及时回收线程,也就意味着我们能处理更多的请求,能够增加我们的吞吐量,这也是异步servlet的主要作用。
异步servlet的内部原理
了解完异步servlet的作用后,我们来看看,tomcat是如何是先异步servlet的。其实上面的代码,主要核心逻辑就两部分, final asynccontext ctx = req.startasync();
和 ctx.complete();
那我们来看看他们究竟做了什么?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
|
public asynccontext startasync(servletrequest request, servletresponse response) { if (!isasyncsupported()) { illegalstateexception ise = new illegalstateexception(sm.getstring( "request.asyncnotsupported" )); log.warn(sm.getstring( "coyoterequest.noasync" , stringutils.join(getnonasyncclassnames())), ise); throw ise; } if (asynccontext == null ) { asynccontext = new asynccontextimpl( this ); } asynccontext.setstarted(getcontext(), request, response, request==getrequest() && response==getresponse().getresponse()); asynccontext.settimeout(getconnector().getasynctimeout()); return asynccontext; } |
我们发现 req.startasync();
只是保存了一个异步上下文,同时设置一些基础信息,比如 timeout
,顺便提一下,这里设置的默认超时时间是30s,也就是说你的异步处理逻辑超过30s后就会报错,这个时候执行 ctx.complete();
就会抛出illegalstateexception 异常。
我们来看看 ctx.complete();
的逻辑
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
|
public void complete() { if (log.isdebugenabled()) { logdebug( "complete " ); } check(); request.getcoyoterequest().action(actioncode.async_complete, null ); } //类:abstractprocessor public final void action(actioncode actioncode, object param) { case async_complete: { cleardispatches(); if (asyncstatemachine.asynccomplete()) { processsocketevent(socketevent.open_read, true ); } break ; } } //类:abstractprocessor protected void processsocketevent(socketevent event, boolean dispatch) { socketwrapperbase<?> socketwrapper = getsocketwrapper(); if (socketwrapper != null ) { socketwrapper.processsocket(event, dispatch); } } //类:abstractendpoint public boolean processsocket(socketwrapperbase<s> socketwrapper, socketevent event, boolean dispatch) { //省略部分代码 socketprocessorbase<s> sc = null ; if (processorcache != null ) { sc = processorcache.pop(); } if (sc == null ) { sc = createsocketprocessor(socketwrapper, event); } else { sc.reset(socketwrapper, event); } executor executor = getexecutor(); if (dispatch && executor != null ) { executor.execute(sc); } else { sc.run(); } return true ; } |
所以,这里最终会调用 abstractendpoint
的 processsocket
方法,之前看过我前面博客的同学应该有印象, endpoint
是用来接受和处理请求的,接下来就会交给 processor
去进行协议处理。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
|
类:abstractprocessorlight public socketstate process(socketwrapperbase<?> socketwrapper, socketevent status) throws ioexception { //省略部分diam socketstate state = socketstate.closed; iterator<dispatchtype> dispatches = null ; do { if (dispatches != null ) { dispatchtype nextdispatch = dispatches.next(); state = dispatch(nextdispatch.getsocketstatus()); } else if (status == socketevent.disconnect) { } else if (isasync() || isupgrade() || state == socketstate.async_end) { state = dispatch(status); if (state == socketstate.open) { state = service(socketwrapper); } } else if (status == socketevent.open_write) { state = socketstate. long ; } else if (status == socketevent.open_read){ state = service(socketwrapper); } else { state = socketstate.closed; } } while (state == socketstate.async_end || dispatches != null && state != socketstate.closed); return state; } |
这部分是重点, abstractprocessorlight
会根据 socketevent
的状态来判断是不是要去调用 service(socketwrapper)
,该方法最终会去调用到容器,从而完成业务逻辑的调用,我们这个请求是执行完成后调用的,肯定不能进容器了,不然就是死循环了,这里通过 isasync()
判断,就会进入 dispatch(status)
,最终会调用 coyoteadapter
的 asyncdispatch
方法
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
|
public boolean asyncdispatch(org.apache.coyote.request req, org.apache.coyote.response res, socketevent status) throws exception { //省略部分代码 request request = (request) req.getnote(adapter_notes); response response = (response) res.getnote(adapter_notes); boolean success = true ; asynccontextimpl asyncconimpl = request.getasynccontextinternal(); try { if (!request.isasync()) { response.setsuspended( false ); } if (status==socketevent.timeout) { if (!asyncconimpl.timeout()) { asyncconimpl.seterrorstate( null , false ); } } else if (status==socketevent.error) { } if (!request.isasyncdispatching() && request.isasync()) { writelistener writelistener = res.getwritelistener(); readlistener readlistener = req.getreadlistener(); if (writelistener != null && status == socketevent.open_write) { classloader oldcl = null ; try { oldcl = request.getcontext().bind( false , null ); res.onwritepossible(); //这里执行浏览器响应,写入数据 if (request.isfinished() && req.sendalldatareadevent() && readlistener != null ) { readlistener.onalldataread(); } } catch (throwable t) { } finally { request.getcontext().unbind( false , oldcl); } } } } //这里判断异步正在进行,说明这不是一个完成方法的回调,是一个正常异步请求,继续调用容器。 if (request.isasyncdispatching()) { connector.getservice().getcontainer().getpipeline().getfirst().invoke( request, response); throwable t = (throwable) request.getattribute(requestdispatcher.error_exception); if (t != null ) { asyncconimpl.seterrorstate(t, true ); } } //注意,这里,如果超时或者出错,request.isasync()会返回false,这里是为了尽快的输出错误给客户端。 if (!request.isasync()) { //这里也是输出逻辑 request.finishrequest(); response.finishresponse(); } //销毁request和response if (!success || !request.isasync()) { updatewrappererrorcount(request, response); request.recycle(); response.recycle(); } } return success; } |
上面的代码就是 ctx.complete()
执行最终的方法了(当然省略了很多细节),完成了数据的输出,最终输出到浏览器。
这里有同学可能会说,我知道异步执行完后,调用 ctx.complete()
会输出到浏览器,但是,第一次doget请求执行完成后,tomcat是怎么知道不用返回到客户端的呢?关键代码在 coyoteadapter
中的 service
方法,部分代码如下:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
|
postparsesuccess = postparserequest(req, request, res, response); //省略部分代码 if (postparsesuccess) { request.setasyncsupported( connector.getservice().getcontainer().getpipeline().isasyncsupported()); connector.getservice().getcontainer().getpipeline().getfirst().invoke( request, response); } if (request.isasync()) { async = true ; } else { //输出数据到客户端 request.finishrequest(); response.finishresponse(); if (!async) { updatewrappererrorcount(request, response); //销毁request和response request.recycle(); response.recycle(); } |
这部分代码在调用完 servlet
后,会通过 request.isasync()
来判断是否是异步请求,如果是异步请求,就设置 async = true
。如果是非异步请求就执行输出数据到客户端逻辑,同时销毁 request
和 response
。这里就完成了请求结束后不响应客户端的操作。
为什么说spring boot的@enableasync注解不是异步servlet
因为之前准备写本篇文章的时候就查询过很多资料,发现很多资料写springboot异步编程都是依赖于 @enableasync
注解,然后在 controller
用多线程来完成业务逻辑,最后汇总结果,完成返回输出。这里拿一个掘金大佬的文章来举例《新手也能看懂的 springboot 异步编程指南 》,这篇文章写得很通俗易懂,非常不错,从业务层面来说,确实是异步编程,但是有一个问题,抛开业务的并行处理来说,针对整个请求来说,并不是异步的,也就是说不能立即释放tomcat的线程,从而不能达到异步servlet的效果。这里我参考上文也写了一个demo,我们来验证下,为什么它不是异步的。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
|
@restcontroller @slf4j public class testcontroller { @autowired private testservice service; @getmapping ( "/hello" ) public string test() { try { log.info( "testasynch start" ); completablefuture<string> test1 = service.test1(); completablefuture<string> test2 = service.test2(); completablefuture<string> test3 = service.test3(); completablefuture.allof(test1, test2, test3); log.info( "test1=====" + test1.get()); log.info( "test2=====" + test2.get()); log.info( "test3=====" + test3.get()); } catch (interruptedexception e) { e.printstacktrace(); } catch (executionexception e) { e.printstacktrace(); } return "hello" ; } @service public class testservice { @async ( "asyncexecutor" ) public completablefuture<string> test1() throws interruptedexception { thread.sleep(3000l); return completablefuture.completedfuture( "test1" ); } @async ( "asyncexecutor" ) public completablefuture<string> test2() throws interruptedexception { thread.sleep(3000l); return completablefuture.completedfuture( "test2" ); } @async ( "asyncexecutor" ) public completablefuture<string> test3() throws interruptedexception { thread.sleep(3000l); return completablefuture.completedfuture( "test3" ); } } @springbootapplication @enableasync public class tomcatdebugapplication { public static void main(string[] args) { springapplication.run(tomcatdebugapplication. class , args); } @bean (name = "asyncexecutor" ) public executor asyncexecutor() { threadpooltaskexecutor executor = new threadpooltaskexecutor(); executor.setcorepoolsize( 3 ); executor.setmaxpoolsize( 3 ); executor.setqueuecapacity( 100 ); executor.setthreadnameprefix( "asynchthread-" ); executor.initialize(); return executor; } |
这里我运行下,看看效果
这里我请求之后,在调用容器执行业务逻辑之前打了一个断点,然后在返回之后的同样打了一个断点,在 controller
执行完之后,请求才回到了 coyoteadapter
中,并且判断 request.isasync()
,根据图中看到,是为 false
,那么接下来就会执行 request.finishrequest()
和 response.finishresponse()
来执行响应的结束,并销毁请求和响应体。很有趣的事情是,我实验的时候发现,在执行 request.isasync()
之前,浏览器的页面上已经出现了响应体,这是springboot框架已经通过 stringhttpmessageconverter
类中的 writeinternal
方法已经进行输出了。
以上分析的核心逻辑就是,tomcat的线程执行 coyoteadapter
调用容器后,必须要等到请求返回,然后再判断是否是异步请求,再处理请求,然后执行完毕后,线程才能进行回收。而我一最开始的异步servlet例子,执行完doget方法后,就会立即返回,也就是会直接到 request.isasync()
的逻辑,然后整个线程的逻辑执行完毕,线程被回收。
聊聊异步servlet的使用场景
分析了这么多,那么异步servlet的使用场景有哪些呢?其实我们只要抓住一点就可以分析了,就是异步servlet提高了系统的吞吐量,可以接受更多的请求。假设web系统中tomcat的线程不够用了,大量请求在等待,而此时web系统应用层面的优化已经不能再优化了,也就是无法缩短业务逻辑的响应时间了,这个时候,如果想让减少用户的等待时间,提高吞吐量,可以尝试下使用异步servlet。
举一个实际的例子:比如做一个短信系统,短信系统对实时性要求很高,所以要求等待时间尽可能短,而发送功能我们实际上是委托运营商去发送的,也就是说我们要调用接口,假设并发量很高,那么这个时候业务系统调用我们的发送短信功能,就有可能把我们的tomcat线程池用完,剩下的请求就会在队列中等待,那这个时候,短信的延时就上去了,为了解决这个问题,我们可以引入异步servlet,接受更多的短信发送请求,从而减少短信的延时。
总结
这篇文章我从手写一个异步servlet来开始,分析了异步servlet的作用,以及tomcat内部是如何实现异步servlet的,然后我也根据互联网上流行的springboot异步编程来进行说明,其在tomcat内部并不是一个异步的servlet。最后,我谈到了异步servlet的使用场景,分析了什么情况下可以尝试异步servlet。
以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持服务器之家。
原文链接:https://juejin.im/post/5d872a1b6fb9a06ad16fadb0