ForkJoin处理千万条数据的多线程处理

ゞ浴缸里的玫瑰 2022-03-27 05:22 427阅读 0赞

这次我的任务是一次插入几千条用户操作数据，在虽然在Linux环境下执行影响不大，但是但window环境下非常的慢，甚至超过请求响应时间，然后我就用ForkJoin处理多数据操作问题，将大任务分为小任务来执行

Fork/Join框架是Java 7提供的一个用于并行执行任务的框架，是一个把大任务分割成若干个小任务，最终汇总每个小任务结果后得到大任务结果的框架。Fork/Join框架要完成两件事情：

1.任务分割：首先Fork/Join框架需要把大的任务分割成足够小的子任务，如果子任务比较大的话还要对子任务进行继续分割

2.执行任务并合并结果：分割的子任务分别放到双端队列里，然后几个启动线程分别从双端队列里获取任务执行。子任务执行完的结果都放在另外一个队列里，启动一个线程从队列里取数据，然后合并这些数据。

操作流程：

一、建立任务类，必须继承RecursiveTask，实现代码如下

> public class MyTask extends RecursiveAction \{
> 
>     @Autowired
> 
>     private AssemblyLogService assemblyLogService;
> 
>     
> 
>     private  List<AssemblyLog> list;
> 
>     
> 
>     // 每个"小任务"最多执行保存50个数
> 
>     private static int MAX = 50;
> 
>  
> 
>     private int start;
> 
>     private int end;
> 
>     
> 
>     
> 
>     MyTask(int start, int end,List<AssemblyLog> list) \{
> 
>         this.start = start;
> 
>         this.end = end;
> 
>     \}
> 
>     
> 
>     @Override
> 
>     protected void compute() \{
> 
>          // 当end-start的值小于MAX时候，开始执行
> 
>         if ((end - start) < MAX) \{
> 
>             for (int i = start ; i <=end ; i++  ) \{
> 
>                AssemblyLog assemblyLog = list.get(i);
> 
>                  assemblyLogService.insertAssemblyLog(assemblyLog);
> 
>              \}
> 
>         \} else \{
> 
>             // 将大任务分解成两个小任务
> 
>             int middle = (start + end) / 2;
> 
>             MyTask left = new MyTask(start, middle,list);
> 
>             MyTask right = new MyTask(middle+1, end,list);
> 
>  
> 
>             // 并行执行两个小任务
> 
>             left.fork();
> 
>             right.fork();
> 
>         \}
> 
>     \}
> 
> \}
> 
> /\*\*
> 
> 这段代码的精髓在于利用类递归的思想，根据自己定义的算法，去切割任务。注意的是要准确的切割任务，比如以上卖弄的代码为例子，它只有集合数量大于MAX（既上方 50）的时候，线程分割。当子线程的集合也大于50的时候，子线程继续分割，直到集合数量少于50才实现if(...)中的代码，如下图所示\*/

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMyOTM5Njc5_size_16_color_FFFFFF_t_70][]

>  // 创建执行的方法 
> 
>    public void toDo(List<AssemblyLog> list) \{  
>         System.err.println("\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*程序开始执行\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*");  
>         // 创建线程池，包含Runtime.getRuntime().availableProcessors()返回值作为个数的并行线程的ForkJoinPool  
>         ForkJoinPool forkJoinPool = new ForkJoinPool();  
>         long start = System.currentTimeMillis();  
>         System.err.println(start);  
>         try \{  
>             // 提交可拆分的Task任务  
>             forkJoinPool.invoke(new MyTask(0, list.size()-1,list));  
>             //阻塞当前线程直到 ForkJoinPool 中所有的任务都执行结束  
>             forkJoinPool.awaitTermination(2, TimeUnit.SECONDS);  
>         \} catch (InterruptedException e) \{  
>             e.printStackTrace();  
>         \}  
>    
>         // 关闭线程池  
>         forkJoinPool.shutdown();  
>         long end = System.currentTimeMillis();  
>         System.err.println(end);  
>         System.err.println("\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*程序执行结束\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*");
> 
>     \}  
> \}

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMyOTM5Njc5_size_16_color_FFFFFF_t_70]: /images/20220327/95908cd792eb4473afad837ab108a073.png