Use of forkjoin

Preface

• ForkJoin yes JDK1.7 Multi thread parallel processing framework added .ForkJoin Use Divide and rule Thought , Divide a big task into small tasks , And then aggregate , Get the final result . It's a bit like Hadoop Medium MapReduce. It also supports work theft .
• Attached below ForkJoin Java Concurrent animation .

This Jar Package download address ： https://sourceforge.net/projects/javaconcurrenta/files/, There are also many interesting animations , Help us learn JUC.

What is job theft ： Suppose there is A、B Two threads execute a task ,A Faster , Finish the work , Now A You can put B Part of the live connection . In general, this will speed up the task execution .

application

demand

• Suppose there is such a need ： I want to count all the amounts in the user table . This watch has 17000003 Data . If I use SQL Statistics are slow , As shown in the figure below .
  
• It cost 4.563 Seconds to find out .
• I found that every time 1000000 It's still very fast , As shown in the figure below .
  
• I just wonder if I can write a program , Split into multiple small tasks , Batch query , Then merge the results .

Use

according to id Range query SUM

... Omit ...
 @Override
    public long sumRecord(int toId, int fromId) {
        QueryWrapper<Users> queryWrapper = new QueryWrapper<>();
        //  use in The statements are combined into one SQL, Avoid repeatedly requesting the database IO
        queryWrapper.ge("id", fromId);
        queryWrapper.le("id", toId);
        queryWrapper.select("IFNULL(SUM(money),0) as money");
        List<Users> users = usersMapper.selectList(queryWrapper);
        if (!CollectionUtils.isEmpty(users)) {
            return users.get(0).getMoney();
        }
        return 0;
    }
... Omit ...

Create task classes and test cases

 ... Omit ...

    @Test
    public void sumTask() {
        long startTime = System.currentTimeMillis();
        ForkJoinPool pool = new ForkJoinPool(Runtime.getRuntime().availableProcessors() * 2);
        //  Simulate tens of millions of data 
        int min = 1;
        int max = 17000003;
        SumTask sumTask = new SumTask(min, max, userService);
        pool.invoke(sumTask);

        System.out.println(" total  " + sumTask.join() +
                "  execution time  " + (System.currentTimeMillis() - startTime));

    }

    public static final Integer THRESHOLD = 1000000;

    public static class SumTask extends RecursiveTask<Long> {

        int fromId;
        int toId;
        private UserService userService;


        public SumTask(int fromId, int toId, UserService userService) {
            this.fromId = fromId;
            this.toId = toId;
            this.userService = userService;
        }

        @Override
        protected Long compute() {
            if (toId - fromId < THRESHOLD) {
                return sumRecord(toId, fromId);
            } else {
                int mid = (fromId + toId) / 2;
                SumTask left = new SumTask(fromId, mid, userService);
                SumTask right = new SumTask(mid + 1, toId, userService);
                invokeAll(left, right);
                return left.join() + right.join();
            }
        }

        public Long sumRecord(int toId, int fromId) {
            System.out.println("  Parameters  " + fromId + " " + toId);
            return userService.sumRecord(toId, fromId);
        }


    }
... Omit ...

Execution results

• The execution speed is obviously faster .
  

Summary

• We can do it in new ForkJoinPool(int parallelism) Number of incoming threads ( The default is CPU The core number ), tuning .
• If it's inheritance RecursiveAction： For tasks that do not return results .

Complete code

• https://gitee.com/apple_1030907690/spring-boot-kubernetes/tree/v1.0.5