java并发编程——Fork/Join

本是古典何须时尚 2022-12-17 01:27 181阅读 0赞

# Fork/Join #

Fork/Join 是 JDK 1.7 加入的新的线程池实现，它体现的是一种分治思想，适用于能够进行任务拆分的 **cpu 密集型  
运算**。

*  所谓的**任务拆分，是将一个大任务拆分为算法上相同的小任务，直至不能拆分可以直接求解**。跟递归相关的一些计算，如归并排序、斐波那契数列、都可以用分治思想进行求解。
 *  Fork/Join 在**分治的基础上加入了多线程，可以把每个任务的分解和合并交给不同的线程来完成**，进一步提升了运算效率。
 *  Fork/Join 默认会创建与 cpu 核心数大小相同的线程池。

# 例子 #

求1~n 之间整数的和。

import lombok.extern.slf4j.Slf4j;
    import java.util.concurrent.ForkJoinPool;
    import java.util.concurrent.RecursiveTask;
    
    @Slf4j(topic = "c.TestForkJoin2")
    public class TestForkJoin2 { 
    
        public static void main(String[] args) { 
            ForkJoinPool pool = new ForkJoinPool(4);
            System.out.println(pool.invoke(new MyTask(5)));
    
            // new MyTask(5) 5+ new MyTask(4) 4 + new MyTask(3) 3 + new MyTask(2) 2 + new MyTask(1)
        }
    }
    
    // 1~n 之间整数的和
    @Slf4j(topic = "c.MyTask")
    class MyTask extends RecursiveTask<Integer> { 
    
        private int n;
    
        public MyTask(int n) { 
            this.n = n;
        }
    
        @Override
        public String toString() { 
            return "{" + n + '}';
        }
    
        @Override
        protected Integer compute() { 
            // 如果 n 已经为 1，可以求得结果了
            if (n == 1) { 
                log.debug("join() {}", n);
                return n;
            }
    
            // 将任务进行拆分(fork)
            AddTask1 t1 = new AddTask1(n - 1);
            t1.fork();
            log.debug("fork() {} + {}", n, t1);
    
            // 合并(join)结果
            int result = n + t1.join();
            log.debug("join() {} + {} = {}", n, t1, result);
            return result;
        }
    }

输出

12:50:07.716 c.MyTask [ForkJoinPool-1-worker-1] - fork() 5 + {4}
    12:50:07.716 c.AddTask [ForkJoinPool-1-worker-0] - fork() 2 + {1}
    12:50:07.716 c.AddTask [ForkJoinPool-1-worker-3] - fork() 3 + {2}
    12:50:07.716 c.AddTask [ForkJoinPool-1-worker-2] - fork() 4 + {3}
    12:50:07.721 c.AddTask [ForkJoinPool-1-worker-0] - join() 1
    12:50:07.722 c.AddTask [ForkJoinPool-1-worker-0] - join() 2 + {1} = 3
    12:50:07.722 c.AddTask [ForkJoinPool-1-worker-3] - join() 3 + {2} = 6
    12:50:07.722 c.AddTask [ForkJoinPool-1-worker-2] - join() 4 + {3} = 10
    12:50:07.722 c.MyTask [ForkJoinPool-1-worker-1] - join() 5 + {4} = 15
    15

图示运行过程  
![在这里插入图片描述][20201021125441878.png_pic_center]  
改进

import lombok.extern.slf4j.Slf4j;
    import java.util.concurrent.ForkJoinPool;
    import java.util.concurrent.RecursiveTask;
    
    public class TestForkJoin { 
    
        public static void main(String[] args) { 
            ForkJoinPool pool = new ForkJoinPool(4);
    // System.out.println(pool.invoke(new AddTask1(5)));
            System.out.println(pool.invoke(new AddTask3(1, 5)));
        }
    }
    
    @Slf4j(topic = "c.AddTask")
    class AddTask1 extends RecursiveTask<Integer> { 
    
        int n;
    
        public AddTask1(int n) { 
            this.n = n;
        }
    
        @Override
        public String toString() { 
            return "{" + n + '}';
        }
    
        @Override
        protected Integer compute() { 
            if (n == 1) { 
                log.debug("join() {}", n);
                return n;
            }
            AddTask1 t1 = new AddTask1(n - 1);
    
            t1.fork();
            log.debug("fork() {} + {}", n, t1);
            int result = n + t1.join();
            log.debug("join() {} + {} = {}", n, t1, result);
            return result;
        }
    }
    
    @Slf4j(topic = "c.AddTask")
    class AddTask2 extends RecursiveTask<Integer> { 
    
        int begin;
        int end;
    
        public AddTask2(int begin, int end) { 
            this.begin = begin;
            this.end = end;
        }
    
        @Override
        public String toString() { 
            return "{" + begin + "," + end + '}';
        }
    
        @Override
        protected Integer compute() { 
            if (begin == end) { 
                log.debug("join() {}", begin);
                return begin;
            }
            if (end - begin == 1) { 
                log.debug("join() {} + {} = {}", begin, end, end + begin);
                return end + begin;
            }
            //int mid = (end + begin) / 2;
            int mid = begin+(end-begin)/2;
    
            AddTask2 t1 = new AddTask2(begin, mid - 1);
            t1.fork();
            AddTask2 t2 = new AddTask2(mid + 1, end);
            t2.fork();
            log.debug("fork() {} + {} + {} = ?", mid, t1, t2);
    
            int result = mid + t1.join() + t2.join();
            log.debug("join() {} + {} + {} = {}", mid, t1, t2, result);
            return result;
        }
    }
    
    @Slf4j(topic = "c.AddTask")
    class AddTask3 extends RecursiveTask<Integer> { 
    
        int begin;
        int end;
    
        public AddTask3(int begin, int end) { 
            this.begin = begin;
            this.end = end;
        }
    
        @Override
        public String toString() { 
            return "{" + begin + "," + end + '}';
        }
    
        @Override
        protected Integer compute() { 
            if (begin == end) { 
                log.debug("join() {}", begin);
                return begin;
            }
            if (end - begin == 1) { 
                log.debug("join() {} + {} = {}", begin, end, end + begin);
                return end + begin;
            }
            int mid = (end + begin) / 2;
    
            AddTask3 t1 = new AddTask3(begin, mid);
            t1.fork();
            AddTask3 t2 = new AddTask3(mid + 1, end);
            t2.fork();
            log.debug("fork() {} + {} = ?", t1, t2);
    
            int result = t1.join() + t2.join();
            log.debug("join() {} + {} = {}", t1, t2, result);
            return result;
        }
    }

图示  
![在这里插入图片描述][20201021130204668.png_pic_center]

[20201021125441878.png_pic_center]: /images/20221123/3ad70c5a021a44f18fb35a4bf324fef4.png
[20201021130204668.png_pic_center]: /images/20221123/27414197554a4022b301c89bd6d76000.png