转载

Spring / Spring boot 异步任务编程 WebAsyncTask

今天一起学习下如何在Spring中进行异步编程。我们都知道，web服务器处理请求 request 的线程是从线程池中获取的，这也不难解释，因为当web请求并发数非常大时，如何一个请求进来就创建一条处理线程，由于创建线程和线程上下文切换的开销是比较大的，web服务器最终将面临崩溃。另外，web服务器创建的处理线程从头到尾默认是同步执行的，也就是说，假如处理线程A负责处理请求B，那么当B没有 return 之前，处理线程A是不可以脱身去处理别的请求的，这将极大限制了web服务器的并发处理能力。

因此线程池解决了线程可循环利用的问题，那同步处理请求怎么去解决呢？答案是异步处理。什么是异步处理呢？异步处理主要是让上面的B请求处理完成之前，能够将A线程空闲出来继续去处理别的请求。那么我们可以这样做，在A线程内部重新开启一个线程C去执行任务，让A直接返回给web服务器，继续接受新进来的请求。

在开始下面的讲解之前，我在这里先区别下两个概念：

1、处理线程

处理线程属于web服务器，负责处理用户请求，采用线程池管理
2、异步线程

异步线程属于用户自定义的线程，可采用线程池管理

spring中提供了对异步任务的支持，采用 WebAsyncTask 类即可实现异步任务，同时我们也可以对异步任务设置相应的回调处理，如当任务超时、抛出异常怎么处理等。异步任务通常非常实用，比如我们想让一个可能会处理很长时间的操作交给异步线程去处理，又或者当一笔订单支付完成之后，开启异步任务查询订单的支付结果。

一、正常异步任务

为了演示方便，异步任务的执行采用 Thread.sleep(long) 模拟，现在假设用户请求以下接口 :

http://localhost:7000/demo/getUserWithNoThing.json

异步任务接口定义如下：

/**
 * 测试没有发生任何异常的异步任务
 */
@RequestMapping(value = "getUserWithNoThing.json", method = RequestMethod.GET)
public WebAsyncTask<String> getUserWithNoThing() {
    // 打印处理线程名
    System.err.println("The main Thread name is " + Thread.currentThread().getName());
    
    // 此处模拟开启一个异步任务,超时时间为10s
    WebAsyncTask<String> task1 = new WebAsyncTask<String>(10 * 1000L, () -> {
    	System.err.println("The first Thread name is " + Thread.currentThread().getName());
    	// 任务处理时间5s,不超时
    	Thread.sleep(5 * 1000L);
    	return "任务1顺利执行成功！任何异常都没有抛出！";
    });
    
    // 任务执行完成时调用该方法
    task1.onCompletion(() -> {
    	System.err.println("任务1执行完成啦！");
    });
    
    System.err.println("task1继续处理其他事情！");
    return task1;
}

控制台打印如下：

The main Thread name is http-nio-7000-exec-1
task1继续处理其他事情！
The first Thread name is MvcAsync1
任务1执行完成啦！

浏览器结果如下：

Spring / Spring boot 异步任务编程 WebAsyncTask

二、抛异常异步任务

接口调用： http://localhost:7000/demo/getUserWithError.json

/**
 * 测试发生error的异步任务
 * @return
 */
@RequestMapping(value = "getUserWithError.json", method = RequestMethod.GET)
public WebAsyncTask<String> getUserWithError() {
	System.err.println("The main Thread name is " + Thread.currentThread().getName());

	// 此处模拟开启一个异步任务
	WebAsyncTask<String> task3 = new WebAsyncTask<String>(10 * 1000L, () -> {
		System.err.println("The second Thread name is " + Thread.currentThread().getName());
		// 此处抛出异常
		int num = 9 / 0;
		System.err.println(num);
		return "";
	});

	// 发生异常时调用该方法
	task3.onError(() -> {
		System.err.println("====================================" + Thread.currentThread().getName()
				+ "==============================");
		System.err.println("任务3发生error啦！");
		return "";
	});
	// 任务执行完成时调用该方法
	task3.onCompletion(() -> {
		System.err.println("任务3执行完成啦！");
	});

	System.err.println("task3继续处理其他事情！");
	return task3;
}

控制台输出如下：

The main Thread name is http-nio-7000-exec-1
task3继续处理其他事情！
The second Thread name is MvcAsync1
2018-06-15 09:40:13.538 ERROR 9168 --- [nio-7000-exec-2] o.a.c.c.C.[.[.[.[dispatcherServlet]      : Servlet.service() for servlet [dispatcherServlet] threw exception

java.lang.ArithmeticException: / by zero
	at com.example.demo.controller.GetUserInfoController.lambda$5(GetUserInfoController.java:93) ~[classes/:na]
	at org.springframework.web.context.request.async.WebAsyncManager.lambda$startCallableProcessing$4(WebAsyncManager.java:317) ~[spring-web-5.0.6.RELEASE.jar:5.0.6.RELEASE]
	at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) ~[na:1.8.0_161]
	at java.util.concurrent.FutureTask.run(FutureTask.java:266) ~[na:1.8.0_161]
	at java.lang.Thread.run(Thread.java:748) [na:1.8.0_161]

2018-06-15 09:40:13.539 ERROR 9168 --- [nio-7000-exec-2] o.a.c.c.C.[.[.[.[dispatcherServlet]      : Servlet.service() for servlet [dispatcherServlet] in context with path [/demo] threw exception [Request processing failed; nested exception is java.lang.ArithmeticException: / by zero] with root cause

java.lang.ArithmeticException: / by zero
	at com.example.demo.controller.GetUserInfoController.lambda$5(GetUserInfoController.java:93) ~[classes/:na]
	at org.springframework.web.context.request.async.WebAsyncManager.lambda$startCallableProcessing$4(WebAsyncManager.java:317) ~[spring-web-5.0.6.RELEASE.jar:5.0.6.RELEASE]
	at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) ~[na:1.8.0_161]
	at java.util.concurrent.FutureTask.run(FutureTask.java:266) ~[na:1.8.0_161]
	at java.lang.Thread.run(Thread.java:748) [na:1.8.0_161]

====================================http-nio-7000-exec-2==============================
任务3发生error啦！
任务3执行完成啦！

当然你也可以对上面做一些异常处理，不至于在用户看来显得不友好，关于异常处理，可以查看我的另一篇博文 Spring boot/Spring 统一错误处理方案的使用

浏览器输出结果：

三、超时异步任务

接口调用： http://localhost:7000/demo/getUserWithTimeOut.json

/**
 * 测试发生任务超时的异步任务
 * @return
 */
@RequestMapping(value = "getUserWithTimeOut.json", method = RequestMethod.GET)
public WebAsyncTask<String> getUserWithTimeOut() {
    System.err.println("The main Thread name is " + Thread.currentThread().getName());
    
    // 此处模拟开启一个异步任务,超时10s
    WebAsyncTask<String> task2 = new WebAsyncTask<String>(10 * 1000L, () -> {
    	System.err.println("The second Thread name is " + Thread.currentThread().getName());
    	Thread.sleep(20 * 1000L);
    	return "任务2执行超时！";
    });
    
    // 任务超时调用该方法
    task2.onTimeout(() -> {
    	System.err.println("====================================" + Thread.currentThread().getName()
    			+ "==============================");
    	return "任务2发生超时啦！";
    });
    
    // 任务执行完成时调用该方法
    task2.onCompletion(() -> {
    	System.err.println("任务2执行完成啦！");
    });
    
    System.err.println("task2继续处理其他事情！");
    return task2;
}

控制台执行结果：

The main Thread name is http-nio-7000-exec-4
task2继续处理其他事情！
The second Thread name is MvcAsync2
====================================http-nio-7000-exec-5==============================
任务2执行完成啦！

浏览器执行结果：

四、线程池异步任务

上面的三种情况中的异步任务默认不是采用线程池机制进行管理的，也就是说，一个请求进来，虽然释放了处理线程，但是系统依旧会为每个请求创建一个异步任务线程，也就是上面我们看到的 MvcAsync 开头的异步任务线程，那这样不行啊，开销特别大呀！所以我们可以采用线程池进行管理，直接在 WebAsyncTask 类构造器传入一个 ThreadPoolTaskExecutor 对象实例即可。

下面我们先看看，当对上面第一种情况执行并发请求时会出现什么情况(此处模拟对 http://localhost:7000/demo/getUserWithNoThing.json 进行并发调用)：

控制台输出如下：

The first Thread name is MvcAsync57
The first Thread name is MvcAsync58
The first Thread name is MvcAsync59
The first Thread name is MvcAsync60
The first Thread name is MvcAsync61
The first Thread name is MvcAsync62
The first Thread name is MvcAsync63
The first Thread name is MvcAsync64
The first Thread name is MvcAsync65
The first Thread name is MvcAsync66
The first Thread name is MvcAsync67
The first Thread name is MvcAsync68
The first Thread name is MvcAsync69
The first Thread name is MvcAsync70
The first Thread name is MvcAsync71
The first Thread name is MvcAsync72
The first Thread name is MvcAsync73
The first Thread name is MvcAsync74
The first Thread name is MvcAsync76
The first Thread name is MvcAsync75
The first Thread name is MvcAsync77
The first Thread name is MvcAsync78
The first Thread name is MvcAsync79
The first Thread name is MvcAsync80

由于没有加入线程池，所以100个请求将开启100个异步任务线程，开销特别大，不推荐。

下面是采用线程池的实现：

调用接口： http://localhost:7000/demo/getUserWithExecutor.json

/**
 * 测试线程池
 * @return
 */
@RequestMapping(value = "getUserWithExecutor.json", method = RequestMethod.GET)
public WebAsyncTask<String> getUserWithExecutor() {
    System.err.println("The main Thread name is " + Thread.currentThread().getName());
    
    // 此处模拟开启一个异步任务,此处传入一个线程池
    WebAsyncTask<String> task1 = new WebAsyncTask<String>(10 * 1000L, executor, () -> {
    	System.err.println("The first Thread name is " + Thread.currentThread().getName());
    	Thread.sleep(5000L);
    	return "任务4顺利执行成功！任何异常都没有抛出！";
    });
    
    // 任务执行完成时调用该方法
    task1.onCompletion(() -> {
    	System.err.println("任务4执行完成啦！");
    });
    
    System.err.println("task4继续处理其他事情！");
    return task1;
}

线程池定义如下：

@Configuration
public class MyExecutor {
    
    @Bean
    public static ThreadPoolTaskExecutor getExecutor() {
    	ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
    	taskExecutor.setCorePoolSize(30);
    	taskExecutor.setMaxPoolSize(30);
    	taskExecutor.setQueueCapacity(50);
    	taskExecutor.setThreadNamePrefix("huang");// 异步任务线程名以 huang 为前缀
    	return taskExecutor;
    }
}

对上面进行并发测试，可以得出下面结果 :