Fork/Join框架分析

案例代码

/**
 * @Auther: honghe
 * @Date: 2019/10/24 11:31
 * @Description:多个订单每个金额都加10000。然后把订单列表返回
 */
public class OrderForkTask extends RecursiveTask<List<OrderForkTask.Order>> {
    private static final int THREAD_HOLD = 10;

    private List<OrderForkTask.Order> orderDTOS;
    private Long amount;

    public OrderForkTask(List<OrderForkTask.Order> orderDTOS, Long amount) {
        this.orderDTOS = orderDTOS;
        this.amount = amount;
    }

    @Override
    protected List<OrderForkTask.Order> compute() {

        List<OrderForkTask.Order> result = new ArrayList<>();
        int orderSize = orderDTOS.size();

        if (orderSize <= THREAD_HOLD) {
            System.out.println(String.format("当前数量[%s]不会进行fork", orderSize));
            result = orderDTOS.stream().map(x -> {
                x.setAmount(x.getAmount() + amount);
                return x;
            }).collect(Collectors.toList());
        } else {
            System.out.println(String.format("当前数量[%s]继续执行fork", orderSize));
            int middle = orderSize / 2;

            System.out.println(String.format("数据被切成了[%s-%s] and [%s-%s]", 0, middle, middle,
                orderSize));

            OrderForkTask left = new OrderForkTask(orderDTOS.subList(0, middle), amount);
            OrderForkTask right = new OrderForkTask(orderDTOS.subList(middle, orderSize), amount);
            //执行子任务
            left.fork();
            right.fork();
            //获取子任务结果
            List<OrderForkTask.Order> lResult = left.join();
            List<OrderForkTask.Order> rResult = right.join();
            result.addAll(lResult);
            result.addAll(rResult);
        }
        return result;
    }

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        List<Order> list = new ArrayList<>();
        for (int i = 1; i < 22; i++) {

            list.add(Order.builder().id(i).amount(0).build());
        }
        ForkJoinPool forkJoinPool = new ForkJoinPool();
        OrderForkTask forkTask = new OrderForkTask(list, 10000L);
        Future<List<Order>> result = forkJoinPool.submit(forkTask);
        List<Order> res = result.get();
        res.stream().forEach(x ->
            System.out.println(String.format("订单[%s]", JSONObject.toJSONString(x)))
        );
    }

    @Data
    @Builder
    @AllArgsConstructor
    @NoArgsConstructor
    public static class Order {
        private int id;
        private long amount;
    }
}




//结果打印
当前数量[21]继续执行fork
数据被切成了[0-10] and [10-21]
当前数量[10]不会进行fork
当前数量[11]继续执行fork
数据被切成了[0-5] and [5-11]
当前数量[5]不会进行fork
当前数量[6]不会进行fork
订单[{"amount":10000,"id":1}]
订单[{"amount":10000,"id":2}]
订单[{"amount":10000,"id":3}]
订单[{"amount":10000,"id":4}]
订单[{"amount":10000,"id":5}]
订单[{"amount":10000,"id":6}]
订单[{"amount":10000,"id":7}]
订单[{"amount":10000,"id":8}]
订单[{"amount":10000,"id":9}]
订单[{"amount":10000,"id":10}]
订单[{"amount":10000,"id":11}]
订单[{"amount":10000,"id":12}]
订单[{"amount":10000,"id":13}]
订单[{"amount":10000,"id":14}]
订单[{"amount":10000,"id":15}]
订单[{"amount":10000,"id":16}]
订单[{"amount":10000,"id":17}]
订单[{"amount":10000,"id":18}]
订单[{"amount":10000,"id":19}]
订单[{"amount":10000,"id":20}]
订单[{"amount":10000,"id":21}]

Process finished with exit code 0