【多线程实例】

使用场景

拿到了一个商品的list,然后要循环list去获取每个商品的明细，由于调用api很依赖于网络，一个个执行速度慢，所以考虑使用线程去解决。

//根据机器id  获取 所有商品信息
public List<ProductResponse> productList(MachineConfigRequest.Code request) {
    //一次性查询数据库机器对应的商品list
    BoolQuery boolQuery = new BoolQuery();
    List<Query> queryList = new ArrayList<>();
    TermQuery termQuery1 = new TermQuery();
    termQuery1.setFieldName(ProductListConfigEnum.MACHINE_ID.getValue());
    termQuery1.setTerm(ColumnValue.fromString(request.getMachine_id()));
    queryList.add(termQuery1);


    boolQuery.setMustQueries(queryList);
    MachineCommonTableStore tableStore = new MachineCommonTableStore();
    SearchQuery searchQuery = new SearchQuery();
    searchQuery.setQuery(boolQuery);
    List<ProductResponse> searchResponse = tableStore.getRowList(searchQuery, ProductResponse.class, ModelEnum.PRODUCT_LIST.getModel(), ModelEnum.PRODUCT_LIST_INDEX.getModel());


    //根据机器id 和 获取所有的货道信息  key:slot value productInfo
    Map<String, ProductSlotInfo> map = this.getSlotInfo(request.getMachine_id());
    //转换为 key：productId value:slotList
    Map<String, List<ProductSlotInfo>> productSlotMap = this.getProductSlotList(map);


    //不需要更新货架图 直接返回库里面存的
    if (request.getLabel() == 0) {
        searchResponse.forEach(productResponse -> {
            //根据商品ID获取商品对应的货道list
            List<ProductSlotInfo> productSlotInfos = productSlotMap.get(productResponse.getProduct_id());
            //塞货道信息和库存
            this.setSlotAndQuantity(productResponse, productSlotInfos);
        });
        return searchResponse.parallelStream()
                .sorted(Comparator.comparing(ProductResponse::getSlot_info))
                .collect(Collectors.toList());
    }


    //根据机器id 调用API 获取所有商品id
    MachineService service = new MachineService();
    List<ProductResponse> list = service.getProductInventory(request.getMachine_id(), ProductResponse.class);


    //循环对比 把之前已经配置过赏级的商品的id和赏级塞进去
    list.forEach(response -> searchResponse.stream().filter(row -> response.getProduct_id().equals(row.getProduct_id())).forEach(row -> {
        response.setId(row.getId());
        response.setMarket(row.getMarket());
    }));


    List<ProductResponse> responseList = Lists.newArrayList();
    if (CollectionUtils.isNotEmpty(list)) {
        ThreadPoolExecutor pool = new ThreadPoolExecutor(50, 50
                , 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(100), (r, executor) -> {
            try {
                executor.getQueue().put(r);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });


        list.forEach(response -> pool.submit(() -> {
            //调用API 获取商品信息
            JSONObject object = new JSONObject();
            object.put(CeresonApiEnum.product_id.getValue(), response.getProduct_id());
            String json = RobotShopClient.callApiGet(CallMachineShopApiConstants.GET_PRODUCT_BY_ID, object.toJSONString());
            JSONObject jsonObject = JSON.parseObject(json);
            String data = jsonObject.getString(CeresonApiEnum.data.getValue());
            JSONObject js = JSON.parseObject(data);
            String productStr = js.getString(CeresonApiEnum.product.getValue());


            //把API获取的商品信息转换为Obj
            ProductResponse productResponse = JSON.parseObject(productStr, ProductResponse.class);
            //根据商品ID获取商品对应的货道list
            List<ProductSlotInfo> productSlotInfos = productSlotMap.get(response.getProduct_id());
            //塞货道信息和库存
            this.setSlotAndQuantity(productResponse, productSlotInfos);


            responseList.add(productResponse);
        }));
        boolean allThreadsIsDone = pool.getTaskCount() == pool.getCompletedTaskCount();
        while (!allThreadsIsDone) {
            allThreadsIsDone = pool.getTaskCount() == pool.getCompletedTaskCount();
        }
        pool.shutdown();
    }


    return responseList.parallelStream()
            .sorted(Comparator.comparing(ProductResponse::getSlot_info))
            .collect(Collectors.toList());
}

public PageResponse<AuctionListByUserIdResponse> getAuctionListByUserId(AuctionRequest.GetAuctionByUserId request) {


    AuctionTableStore auctionTableStore = new AuctionTableStore();


    List<Query> list = new ArrayList<>();
    BoolQuery boolQuery = new BoolQuery();
    BoolQuery shouldQuery = new BoolQuery();


    List<Sort.Sorter> sorter = TableStoreTemplate.getSorter(request.getSort());
    PageResponse<AuctionListByUserIdResponse> pageResponse = new PageResponse<>();
    try {
        pageResponse = auctionTableStore.queryAll(request.getNextToken(), sorter, boolQuery, request.getLimit(),
                AuctionListByUserIdResponse.class);
    } catch (Exception e) {
        log.error("getAuctionList failed exception = {}", e);
        throw new BaseException(ResponseCode.Bad_Request.getCode(), "查询失败！");
    }
    List<AuctionListByUserIdResponse> auctionListByUserIdResponses = pageResponse.getData();


    if (null != auctionListByUserIdResponses && auctionListByUserIdResponses.size() > 0) {
        ThreadPoolExecutor pool = new ThreadPoolExecutor(auctionListByUserIdResponses.size(), auctionListByUserIdResponses.size()
                , 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(), new RejectedExecutionHandler() {
            @Override
            public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
                try {
                    executor.getQueue().put(r);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });
        String robotSign = "##robot";
        auctionListByUserIdResponses.forEach(response -> {
            pool.submit(new Runnable() {
                @Override
                public void run() {
                    String sellerId = response.getSellerId();
                    String buyerId = response.getBuyerId();
                    String auctionId = response.getId();
                    String twr = "";
                    /* 处理是否为机器人 robotSign */
                    if (buyerId.endsWith(robotSign) || sellerId.endsWith(robotSign)) {
                        twr = "是";
                    } else {
                        twr = "否";
                    }
                    response.setWhetherRobot(twr);
                    response.setAuctionGoodsList(getGoodsRangeByAuctionId(sellerId, auctionId, DeleteFlagEnum.DELETE));
                    response.setAuctionBidGoodsList(getGoodsRangeByAuctionId(buyerId, auctionId, DeleteFlagEnum.DELETE));
                    response.setCostPrice(getTotalCostPrice(response.getAuctionGoodsList()));
                }
            });
        });
        boolean allThreadsIsDone = pool.getTaskCount() == pool.getCompletedTaskCount();
        while (!allThreadsIsDone) {
            allThreadsIsDone = pool.getTaskCount() == pool.getCompletedTaskCount();
        }
        pool.shutdown();
    }
    for (AuctionListByUserIdResponse info : pageResponse.getData()){
        System.out.println(info.getTransactionPrice() + "," + info.getCostPrice());
    }
    return pageResponse;
}