Web服务耗时接口同步请求异步处理解决方案

2021-09-16 本文已影响0人蜀山_竹君子

背景

生产环境存在一些接口，其因为后端服务涉及到大量数据库读写操作，因此接口非常耗时。比如商品导入功能，经过事务拆分、拆分查询并组装数据等手段对功能进行了优化，在不改变业务设计的前提下几乎无任何优化空间。
容器(tomcat)中线程的数量是一定的，容器处理大量耗时请求时，势必会影响其他接口的正常访问。
因此，在不改变业务的前提下，在高并发场景提高商品服务的吞吐率优化是非常必要的。

技术实现方案

采用Spring MVC异步处理方案（适用于耗时同步交易场景）。Spring MVC异步处理实现方案通常支持3种方式：

Callable实现
WebAsyncTask实现
DefferedResult实现

我们使用DefferedResult + 线程池 + 阻塞队列LinkedBlockingQueue 实现请求的异步处理同步响应。

关于DeferredResult

DeferredResult从 Spring 3.2 开始可用，有助于将长时间运行的计算从 http-worker 线程卸载到单独的线程。
尽管另一个线程会占用一些资源进行计算，但工作线程在此期间不会被阻塞并且可以处理传入的其他客户端请求。
异步请求处理模型非常有用，因为它有助于在高负载期间很好地扩展应用程序，尤其是对于 IO 密集型操作。

DeferredResult处理流程

DeferredResult的处理过程与Callback类似，不一样的地方在于它的结果不是DeferredResult直接返回的，而是由其它线程通过同步的方式设置到该对象中。它的执行过程如下所示：

客户端请求服务
SpringMVC调用Controller，Controller返回一个DeferredResult对象
SpringMVC调用ruquest.startAsync
DispatcherServlet以及Filters等从应用服务器线程中结束（释放容器线程），但Response仍旧是打开状态，也就是说暂时还不返回给客户端
异步线程处理实际业务并将结果设置到DeferredResult中，SpringMVC将请求发送给应用服务器继续处理
DispatcherServlet再次被调用并且继续处理DeferredResult中的结果，最终将其返回给客户端。

重要技术点

线程池

线程池维护多个线程，等待监督管理者分配可并发执行的任务。这种做法，一方面避免了处理任务时创建销毁线程开销的代价，另一方面避免了线程数量膨胀导致的过分调度问题，保证了对内核的充分利用。

LinkedBlockingQueue

LinkedBlockingQueue实现是线程安全的，实现了先进先出等特性，是作为生产者消费者的首选，LinkedBlockingQueue 可以指定容量，也可以不指定，不指定的话，默认最大是Integer.MAX_VALUE，其中主要用到put和take方法，put方法在队列满的时候会阻塞直到有队列成员被消费，take方法在队列空的时候会阻塞，直到有队列成员被放进来。这里需要注意直接使用LinkedBlockingQueue阻塞队列作为线程池会存在一个问题，当workcount > corePool时优先进入队列排队，因此当请求并发过多会导致请求缓慢，甚至因为队列过多出现内存溢出（JDK是先排队再涨线程池）

网络上基本找得到的DeferredResult相关的技术博客或文章都是直接创建线程或使用LinkedBlockingQueue的线程池，实际在压力测试过程当模拟接口并发到50就出现大量延迟，比不优化时性能还差。有兴趣的可以直接使用LinkedBlockingQueue作为线程池队列压力测试看看效果。

Tomcat的线程池

org.apache.tomcat.util.threads.TaskQueue
org.apache.tomcat.util.threads.ThreadPoolExecutor

因为我们优化的是web接口请求，不能因为LinkedBlockingQueue的排队导致接口出现大量延迟和缓慢，因此我们在实现过程不直接使用LinkedBlockingQueue作为线程池的阻塞队列，而是使用tomcat的线程池TaskQueue，TaskQueue继承了JDK的LinkedBlockingQueue 并扩展了JDK线程池的功能，主要体现在两点：

Tomcat的ThreadPoolExecutor使用的TaskQueue，是无界的LinkedBlockingQueue，但是通过taskQueue的offer方法覆盖了LinkedBlockingQueue的offer方法，改写了规则，使得线程池能在任务较多的情况下增长线程池数量——JDK是先排队再涨线程池，Tomcat则是先涨线程池再排队。
Tomcat的ThreadPoolExecutor改写了execute方法，当任务被reject时，捕获异常，并强制入队。

代码实现

创建处理耗时任务的线程池

public static ThreadPoolExecutor executor = null;

  private TaskQueue taskqueue;

  protected int maxQueueSize = Integer.MAX_VALUE;

  protected int threadPriority = 5;
  protected boolean daemon = true;
  protected String namePrefix = "testsleep-";
  protected int minSpareThreads = 25;
  protected int maxThreads = 200;
  protected int maxIdleTime = 60000;
  protected long threadRenewalDelay = 1000L;
  protected boolean prestartminSpareThreads = false;


  /**
   * 初始化时启动监听请求队列
   */
  @PostConstruct
  public void init() {
    /*cachedThreadPool = new ThreadPoolExecutor(4,
        50,
        0,
        TimeUnit.MILLISECONDS,
        new LinkedBlockingQueue<>(50),
        r -> new Thread(r));*/

    // 任务队列:这里你看到的是一个无界队列，但是队列里面进行了特殊处理
    taskqueue = new TaskQueue(maxQueueSize);
    TaskThreadFactory tf = new TaskThreadFactory(namePrefix,daemon, threadPriority);
    // 创建线程池，这里的ThreadPoolExecutor是Tomcat继承自JDK的ThreadPoolExecutor
    executor = new ThreadPoolExecutor(
        minSpareThreads, maxThreads, // 核心线程数与最大线程数
        maxIdleTime, TimeUnit.MILLISECONDS, // 默认6万毫秒的超时时间，也就是一分钟
        taskqueue, tf); // 玄机在任务队列的设置
    executor.setThreadRenewalDelay(threadRenewalDelay);
    if (prestartminSpareThreads) {
      executor.prestartAllCoreThreads(); // 预热所有的核心线程
    }
    taskqueue.setParent(executor);


  }

重构请求，使用DeferredResult实现异步处理

这里我们直接使用Thread.sleep模拟一个耗时任务

@GetMapping("/users-anon/test/{testkey}")
  public DeferredResult<Result<String>> testSleep(@PathVariable String testkey) {

    //return service.testSleep(); //直接调用耗时业务处理

    DeferredResult<Result<String>> output = new DeferredResult<>(1000 * 30L);

    output.onTimeout(() ->
        output.setErrorResult(
            ResponseEntity.status(HttpStatus.REQUEST_TIMEOUT)
                .body("Request timeout occurred.")));

    log.info("[ TestSleepController ] 接到请求");

    //转到后台线程
    QueueListener.executor.execute(() -> {
      log.info("开始执行耗时任务：{}", System.currentTimeMillis());
      try {
        Thread.sleep(6000);
      } catch (InterruptedException e) {
      }
      log.info("执行耗时任务结束：{}", System.currentTimeMillis());
      output.setResult(Result.succeed());
    });


    log.info("[ TestSleepController ] 返回DeferredResult，并释放容器线程.");

    return output;


  }

压力测试

机器参数

CentOS Linux release 7.3.1611 (Core) 1核 2.30GHz 8G内存

服务部署

Docker容器

测试结果

单接口测试

在测试DeferredResult前，我们先模拟一个耗时接口，并对接口进行压力测试：
100线程循环一次

200线程循环一次

500线程循环一次

当耗时接口在100、200并发下接口基本正常，当达到500时接口响应出现明显的迟缓。

混合接口测试

我们建立两个线程组，一个是正常的耗时任务A（线程组2），一个是使用DeferredResult优化的耗时任务B（线程组1）。
任务A、B 各50线程循环一次
压测结果如下：

整体响应都差不多。

任务A 50线程循环一次任务B 500线程循环一次
DeferredResult优化的耗时任务压测结果：

正常任务压测结果：

当任务BDeferredResult优化后的接口并发爆发式增长后，接口的响应仍和优化前一样出现大范围的延迟，但是任务A的接口响应并未收到影响。

结果分析

通过压测结果分析我们可以得出DeferredResult优化的耗时任务虽然不能提示耗时接口本身的响应速度，但是能极大减少耗时任务对服务容器线程的占用，提升应用在高并发场景下本身的吞吐量。