Deferred实现异步调用

2022-07-24

场景

生产中，往往会遇到这样的情景，controller 中调用的方法很多都是和第三方有关的，例如 JMS，定时任务，队列等，拿 JMS 来说，比如 controller 里面的服务需要从 JMS 中拿到返回值，才能给客户端返回，而从 JMS 拿值这个过程也是异步的，这个时候，我们就可以通过 Deferred 来实现整个的异步调用。

如果只使用 Async 方式，虽然开启了异步处理方式，但是拿不到返回结果。

而 Deferred 方式，不仅通过 Async 开启了异步线程，而且还能获取到异步处理结果，一定程度提高了系统的吞吐量。

实现

首先，模拟一个长时间调用的任务，代码如下：

@Component  
public class LongTimeTask {  
    private final Logger logger = LoggerFactory.getLogger(this.getClass());  
    @Async  
    public void execute(DeferredResult<String> deferred){  
        logger.info(Thread.currentThread().getName() + "进入 taskService 的 execute 方法");  
        try {  
            // 模拟长时间任务调用，睡眠 2s  
            TimeUnit.SECONDS.sleep(2);  
            // 2s 后给 Deferred 发送成功消息，告诉 Deferred，我这边已经处理完了，可以返回给客户端了  
            deferred.setResult("world");  
        } catch (InterruptedException e) {  
            e.printStackTrace();  
        }  
    }  
}

Tips：启动类需要增加 @EnableAsync 注解

接着，实现异步调用，controller 如下：

@RestController  
public class AsyncDeferredController {  
    private final Logger logger = LoggerFactory.getLogger(this.getClass());  
    private final LongTimeTask taskService;  
      
    @Autowired  
    public AsyncDeferredController(LongTimeTask taskService) {  
        this.taskService = taskService;  
    }  
      
    @GetMapping("/deferred")  
    public DeferredResult<String> executeSlowTask() {  
        logger.info(Thread.currentThread().getName() + "进入 executeSlowTask 方法");  
        DeferredResult<String> deferredResult = new DeferredResult<>();  
        // 调用长时间执行任务  
        taskService.execute(deferredResult);  
        // 当长时间任务中使用 deferred.setResult ("world"); 这个方法时，会从长时间任务中返回，继续 controller 里面的流程  
        logger.info(Thread.currentThread().getName() + "从 executeSlowTask 方法返回");  
        // 超时的回调方法  
        deferredResult.onTimeout(new Runnable(){  
          
            @Override  
            public void run() {  
                logger.info(Thread.currentThread().getName() + " onTimeout");  
                // 返回超时信息  
                deferredResult.setErrorResult("time out!");  
            }  
        });  
          
        // 处理完成的回调方法，无论是超时还是处理成功，都会进入这个回调方法  
        deferredResult.onCompletion(new Runnable(){  
          
            @Override  
            public void run() {  
                logger.info(Thread.currentThread().getName() + " onCompletion");  
            }  
        });  
          
        return deferredResult;  
    }  
}

异步调用可以使用 AsyncHandlerInterceptor 进行拦截，使用示例如下：

@Component  
public class MyAsyncHandlerInterceptor implements AsyncHandlerInterceptor {  
      
    private static final Logger logger = LoggerFactory.getLogger(MyAsyncHandlerInterceptor.class);  
  
    @Override  
    public boolean preHandle(HttpServletRequest request, HttpServletResponse response, Object handler)  
            throws Exception {  
        return true;  
    }  
  
    @Override  
    public void postHandle(HttpServletRequest request, HttpServletResponse response, Object handler,  
            ModelAndView modelAndView) throws Exception {  
				//      HandlerMethod handlerMethod = (HandlerMethod) handler;  
        logger.info(Thread.currentThread().getName()+ "服务调用完成，返回结果给客户端");  
    }  
  
    @Override  
    public void afterCompletion(HttpServletRequest request, HttpServletResponse response, Object handler, Exception ex)  
            throws Exception {  
        if(null != ex){  
            System.out.println("发生异常:"+ex.getMessage());  
        }  
    }  
  
    @Override  
    public void afterConcurrentHandlingStarted(HttpServletRequest request, HttpServletResponse response, Object handler)  
            throws Exception {  
          
        // 拦截之后，重新写回数据，将原来的 hello world 换成如下字符串  
        String resp = "my name is chhliu!";  
        response.setContentLength(resp.length());  
        response.getOutputStream().write(resp.getBytes());  
          
        logger.info(Thread.currentThread().getName() + " 进入 afterConcurrentHandlingStarted 方法");  
    }  
  
}

执行结果如下：

2017-12-07 19:25:40.192  INFO 6196 --- [nio-8060-exec-7] c.t.controller.AsyncDeferredController   : http-nio-8060-exec-7 进入 executeSlowTask 方法  
2017-12-07 19:25:40.193  INFO 6196 --- [nio-8060-exec-7] .s.a.AnnotationAsyncExecutionInterceptor : No TaskExecutor bean found for async processing  
2017-12-07 19:25:40.194  INFO 6196 --- [nio-8060-exec-7] c.t.controller.AsyncDeferredController   : http-nio-8060-exec-7 从 executeSlowTask 方法返回  
2017-12-07 19:25:40.198  INFO 6196 --- [nio-8060-exec-7] c.t.i.MyAsyncHandlerInterceptor          : http-nio-8060-exec-7 进入 afterConcurrentHandlingStarted 方法  
2017-12-07 19:25:40.202  INFO 6196 --- [cTaskExecutor-1] com.travelsky.controller.LongTimeTask    : SimpleAsyncTaskExecutor-1 进入 taskService 的 execute 方法  
2017-12-07 19:25:42.212  INFO 6196 --- [nio-8060-exec-8] c.t.i.MyAsyncHandlerInterceptor          : http-nio-8060-exec-8 服务调用完成，返回结果给客户端  
2017-12-07 19:25:42.213  INFO 6196 --- [nio-8060-exec-8] c.t.controller.AsyncDeferredController   : http-nio-8060-exec-8 onCompletion

从上面的执行结果不难看出，容器线程会立刻返回，应用程序使用线程池里面的 cTaskExecutor-1 线程来完成长时间任务的调用，当调用完成后，容器又启了一个连接线程，来返回最终的执行结果。

这种异步调用，在容器线程资源非常宝贵的时候，能够大大的提高整个系统的吞吐量。