AsyncResult使用场景

桃扇骨 2023-06-30 10:54 7阅读 0赞

### 场景 ###

从es查询商品信息，分批查询，5000个查询一次，再把每次查询结果汇总，如果分10次查询，正常查询每次返List<Product>，要等前一次查询返回结果后，才能执行下一次查询，汇总结果，是同步执行的过程。试想如果可以同时进行查询，之间互不影响，每个查询返回结构后，直接汇总，这样就大大节约了查询时间。 AsyncResult作用就在这里。

### 借用举例 ###

AsyncResult是异步方式，异步主要用于调用的代码需要长时间运行，才能返回结果的时候，可以不阻塞调用者。

打个比方，同步方式就是你打电话给客服，客服没法立刻解决，客服说你等等，别挂电话，然后等了10分钟，再告诉你。再挂电话。  
此时电话费照收，并且你不能接打别人的电话。

异步方式就是，客服说，我先查查，查到了给你回电话，然后挂断。你干别的事情。等了10分钟，客服给你来电话了，告诉你结果。

### 代码 ###

1，分批查询

List<ResBaseProductShopVO> adapter1(Set<ShopProductDto> source) throws ExecutionException, InterruptedException {
            List<ResBaseProductShopVO> result = new ArrayList<>(source.size() * 100);
            List<List<ShopProductDto>> query = Lists.partition(source.stream().collect(Collectors.toList()), 5000);
            List<Future<List<ResBaseProductShopVO>>> futures = new ArrayList<>((source.size() / 5000) + 1);
            for (List<ShopProductDto> q : query) {
                futures.add(
                        asyncService.searchShopProductAsync(
                                ReqBaseProductShopPageEsQueryVO.createWithShopCodeAndProductCode(
                                        q
                                                .stream()
                                                .filter(r -> (org.apache.commons.lang3.StringUtils.isNotBlank(r.getShopCode()) && org.apache.commons.lang3.StringUtils.isNotBlank(r.getProductCode())))
                                                .map(ShopProductDto::shopCodeConcatProductCode).collect(Collectors.toList())
                                )
                        ));
            }
            obtainFromFuture(futures, result);
            return fillNullParams(result);
        }

2，异步查询

@Override
      @Async
      public Future<List<ResBaseProductShopVO>> searchShopProductAsync(ReqBaseProductShopPageEsQueryVO query) {
        List<ResBaseProductShopVO> res = new ArrayList<>(5000);
        if (query != null) {
          while (true) {
            List<ResBaseProductShopVO> vos = esService.baseProductShopEsTermsSearch(query);
            if (CollectionUtils.isNotEmpty(vos)) {
              res.addAll(vos);
              if (!query.continueSearch(vos.size())) {
                break;
              }
              query.increase();
            } else {
              break;
            }
          }
        }
        return new AsyncResult<>(res);
      }

3，从future获取返回结果

@Override
        public void obtainFromFuture(List<Future<List<ResBaseProductShopVO>>> futures, List<ResBaseProductShopVO> result) throws ExecutionException, InterruptedException {
            Iterator<Future<List<ResBaseProductShopVO>>> iterator = futures.iterator();
            while (iterator.hasNext()) {
                Future<List<ResBaseProductShopVO>> future = iterator.next();
                if (future.isDone()) {
                    List<ResBaseProductShopVO> source = future.get();
                    if (CollectionUtils.isNotEmpty(source)) {
                        result.addAll(source);
                    }
                    iterator.remove();
                }
                if (!iterator.hasNext()) {
                    iterator = futures.iterator();
                }
            }
        }