线程复用助力之家出行平台运力提升的应用实践

之家出行平台起源

2020年疫情开始，为响应国家号召减少公共交通，避免人多聚集，虽出行人数减少，但网约车用户不降反增；为了更好的保障用户安心到店，同时打通用户线上询价与线下到店看车行为关联，更好的服务之家用户，之家出行平台诉求迫在眉睫；

既然需求已经很明确，接下来就使围绕着怎么构建之家出行，经过调查发现网约车细分如下：

（网约车市场细分,摘自：2022年网约车行业研究报告）

经过分析，以下两种方式更适合之家出行平台要求，但依然各有千秋：

1）网约车 S2B2C

优点：服务、数据安全有保证；

缺点：需网约车运营资质，成本高，运营难度大；

2）聚合平台 S2B2C

优点：依赖流量优势，低成本，易运营；

缺点：存在数据安全风险，打车成功率以及打车体验受限；

之家构建出行平台初衷是为了保障用户安心到店，对抢占目前网约车市场并无兴趣，在保证数据安全，用户用车到店安全的前提下，用车成本是必须考虑的因素；相比网约车的高成本高运营，聚合平台的优势相对突出，无需关心具体运力公司的成本以及运营，只需使用他们的服务，同时之家的用车流量也是各运力公司拓展渠道的香饽饽；

出行平台系统要求

100+主要城市全部运力覆盖，叫车成功率在60%以上；保证之家业务覆盖到的城市均可以让用户使用出行平台打车到店，平均叫车成功率达到60%以上，这样才能保证用户的叫车体验，太低会使用户体验降低，失去信任，提升叫车成功率是用车体验重要一步；

要实现该目标需要：

（一）统一运力接入流程，配置优化管理，快速接入；

① 分析接入流程：

② 针对运力公司和聚合平台接入特点，分别针对运力公司以及聚合平台进行流程细化，寻找共同点以及差异点，针对共同点进行功能设计，抽取，减少重复造轮子，针对差异点进行功能定制；通过模板方法设计模式、适配器模式、装饰器模式、工厂模式、单例模式、策略模式等设计模式提升运力接入功能抽象，功能统一同时数据隔离，保证服务安全可靠；

（二）无论同时选择多少个运力叫车，都不能影响预估以及下单性能；

经过上一步我们使用了统一的接入流程，虽然能够通过功能抽象、抽取，实现运力公司以及聚合平台两种快速接入。但用户预估时每选择一个运力，对应的就会多一套运力接口调用，以下用户预估时分析：

① 运力下单前，需要先对用车（时间、车型、起点、终点等）行程进行预估，使用获取到的有效预估信息，才能调用运力下单；

② 预估时需要调用C端选择的每个运力的预估接口，由于木桶效应，最长耗时的运力拉长整个预估接口响应时长；

③ 随着运力接入数量的增多，每次预估调用运力预估接口也成倍增加，没有上限，对服务器资源要求高；

分析之后有解决办法如下

运力预估返回数据是存在时效（基本在10分钟左右），可以对预估数据进行缓存，针对同一个用户，服务类型、用车类型以及上下车地点等信息不变的情况下可以对运力返回预估数据进行缓存，这样可以减少用户频繁预估时无效调用，提升性能，但解决同时请求多套运力接口并保证总体RT的问题主要解决办法如下：

㊀ 线程复用，减少创建、销毁、调度线程的开销，譬如JDK自带的J.U.C包线程池就是一个不错的选择；

线程池解决方案

jdk1.8对于Worker类的官方注释

 /**
 * Class Worker mainly maintains interrupt control state for
 * threads running tasks, along with other minor bookkeeping.
 * This class opportunistically extends AbstractQueuedSynchronizer
 * to simplify acquiring and releasing a lock surrounding each
 * task execution.  This protects against interrupts that are
 * intended to wake up a worker thread waiting for a task from
 * instead interrupting a task being run.  We implement a simple
 * non-reentrant mutual exclusion lock rather than use
 * ReentrantLock because we do not want worker tasks to be able to
 * reacquire the lock when they invoke pool control methods like
 * setCorePoolSize.  Additionally, to suppress interrupts until
 * the thread actually starts running tasks, we initialize lock
 * state to a negative value, and clear it upon start (in
 * runWorker).
 */

线程池配置

由于本身业务为IO密集型的，并且使用之家云容器Docker配置：4核心8G内存；

详细配置：coreSize:10,maxSize:20,rejectedExecutionHandler:ThreadPoolExecutor.CallerRunsPolicy()

摘自：Java Concurrency In Practice

具体实现

/** 价格预估伪代码；**/  
public JSONArray estimatePrice(EstimateQuery estimateQuery) {
        /* .. */
        /* 查询可用运力 */
        List<Integer> channelIdList = channelMapper.getAllChannelIds();
        /** 构建多个异步调用任务；**/
        CompletableFuture[] cfs = channelIdList.stream().map(i-> CompletableFuture
                .supplyAsync(()-> request(i), taskExecutor )
                .whenComplete((u,e) -> {if(null != u){array.add(u);}})).toArray(CompletableFuture[]::new);
        /** 开始执行任务，阻塞主线程，等待各任务全部处理完成； **/
        CompletableFuture.allOf(cfs).join();
      /** 后续处理 .. */    
}

/**
 * 调用运力对应的预估接口获取预估数据；
 */
private JSONObject request(int channeId){
  switch (channelId){
    case Constant.SQ_CHANNEL_ID:
      /** 首汽预估；*/
      return sqService.getEstimatePrice(estimateQuery);
    case xx:
      /* 其他预估；省略.. */
  }
  return null;
}

通过设置HTTP请求的connectTime，requestTime控制接口调用时长，以及使用Alibaba Sentinel对各运力封装接口进行熔断处理，从而保证接口整体响应时长；

/**
 *  通过 @SentinelResource 为运力的预估接口增加 熔断机制，一旦接口达到熔断条件（访问超时占比超过设定阈值）则不在调用运力http接口，以提升性能；
 *  通过 CONNECT_TIMEOUT、REQUEST_TIMEOUT 设置http接口请求的超时时间，接口超时直接放弃结果，以提升接口整体性能；
 */
/**sqService**/
@SentinelResource(value = "sqEstimate", blockHandler = "sqEstimateBlockHandler", fallback = "orderFallback")
public JSONObject getEstimatePrice(EstimateQuery estimateQuery){
   /** 调用接口数据请求转换 **/
   Map<?,?> requestMap = this.convertData(estimateQuery);
   /** 调用接口请求数据，通过 CONNECT_TIMEOUT，REQUEST_TIMEOUT限制请求时间 **/
   String result = HttpUtils.post(this.REQUEST_URL,CONNECT_TIMEOUT,REQUEST_TIMEOUT);
   /** 对结果数据进行封装处理 **/
   return this.convertResult(result);
}

用户在下单时，会尽可能的勾选更多运力以及车型，这样能够保证下单的成功率。但每次价格预估需要同步调用的运力接口是不同，可以对齐每次预估调用时申请线程的数量，复用已有线程，但线程池的资源是有限的，仍然出现线程池线程全部占用，等待的情况；

设置每个预估请求的运力数为channelCount，预估最大允许的线程数量为maxThread，一个请求周期为requestPeriod；

① 预估请求运力数channelCount <= maxThread，有几个运力从线程池获取几个线程进行数据并行请求；

② 预估请求运力数channelCount > maxThread，则对运力按照设置的优先级进行分组排序，对运力分成maxThread个组，每个组内的运力按照优先级再次划分梯队，第一梯队中的运力调用优先级最高，首选获取线程并且执行，当一个梯队中的调用超时或者数据返回失败之后线程并不会释放，会执行当前分组中的第二个梯队中的运力调用，对数据进行补全，同时也能实现调用的优先级；

每个预估都会创建一个数组，数组存放的是单链表结构运力调用任务，任务节点数据结构：

//数据结构
public class ChannelTaskNode{
  
  private boolean header; //头节点
  private int priority;   //优先级
  private int groupIndex; //group
  private Long channelId; // 运力ID
  private ChannelTaskNode next;   // 下一个节点

}

对任务的初始化以及分组处理如下：

//每次预估对应的任务分组
ChannelTaskNode[] channelTaskArray = new ChannelTaskNode[maxThread];
//按照优先级获取对应的运力列表；
int[] channelArray = queryAllOpenChannelSortedByPriority();
//初始化
for(int i=0;i<channelArray.length;i++){
   int idx = i % maxThread;
   ChannelTaskNode node = channelTaskArray[idx];
   if( null != node ){
      node.next = new ChannelTaskNode(channelArray[i],null,priority,false);
   }else{
     channelTaskArray[i] = new ChannelTaskNode(channelArray[i],null,priority,true);
   }
}

对分组后的任务按照阶梯进行任务分层复用追加：

/** 对任务进行拼接 **/
CompletableFuture[] allTasks = new CompletableFuture[channelTaskArray.length] ;
 for(ChannelTaskNode node:channelTaskArray){
    ChannelTaskNode temp = node;
    CompletableFuture tempFuture = null;
    while(null != temp){
        long channelId = temp.getChannelId();
        if(temp.isHeader()){
            tempFuture = CompletableFuture.supplyAsync(()-> service.estimatePrice(channelId));
        }else{
            tempFuture = tempFuture.handle((obj, error) -> {
                if(null != error ){
                    return service.estimatePrice(channelId);
                }
                return obj;
            });
        }
        temp = node.next;
    }
    allTasks[node.groupIndex] = tempFuture;
}

/**  开始调用 **/
CompletableFuture.allOf(allTasks).join();

/** 对结果进行处理 **/
//省略

运行结果对比（横轴为预估运力数量，纵轴为响应时间毫秒）；

优化多线程方案，虽然会随着预估时运力增多带来总体耗时增加，但在总RT控制的前提下，又能保证线程池资源的合理了分配，又能使得各运力能够按照概率的配置被调用到，总体效果还是显而易见；

本文主要讲解了，出行平台的来源，系统构建时的思考，以及在设计时遇到的问题；前期主要是以接入更多的运力，来保证系统的覆盖以及用户用车体验，由于网约车系统的规范性每个运力的接入基本都是设计到数据适配、接口对接、订单系统集成、费用结算集成等节点；为了降低开发成本，快速接入，提供了一套接入的统一的接入流程，并对接入做统一处理，这样单运力接入工作缩短一倍以上；由于出行平台为聚合型平台，用车功能基本都是需要调用各接入的运力公司的接口，预估接口又是整个叫车环节中前置的重要入口，在对接口性能以及用车体验双重考验之下，我们运用了线程池的方式在预估时对多个运力做并发调用，但由于预估时选择运力的不确定性，为了保证每个预估调用的公平，预估多个运力时，按照运力进行分组，复用当前从线程池中获取的线程，同时也实现运力的优先级；为了保证整个预估接口的RT，每个运力调用时均存在connectTimeOut以及requestTimeOut，同时为了避免接口性能问题对整个接口影响，还引入了Alibaba Sentinel对各运力封装接口进行熔断处理，从而保证接口整体响应时长；经过以上处理，价格预估阶段不再受限与接入运力的多少，不再成为整个用车环节中的瓶颈，但HTTP请求的超时限制并不严格准确，后续会使用时间轮算法对该部分功能进行持续完善；