经验首页 前端设计 程序设计 Java相关 移动开发 数据库/运维 软件/图像 大数据/云计算 其他经验
当前位置：技术经验 » Java相关 » Java » 查看文章
简单设计一个JAVA并行处理工具类
来源：cnblogs　　作者：狂盗一枝梅　　时间：2024/8/7 8:51:11　　对本文有异议

在工作中，我们肯定遇到过一个接口要处理N多事项导致接口响应速度很慢的情况，通常我们会综合使用两种方式来提升接口响应速度

  1. 优化查询SQL，提升查询效率
  2. 开启多线程并发处理业务数据

这里讨论第二种方案：使用多线程并发处理业务数据，最后处理完成以后，拼装起来返回给前端，每个人的实现方案都不一样，我在工作的这几年也经历了几种写法。

一、几种常见的并行处理写法

方法一：Future写法

其代码形式如下

  1. @Test
  2. public void test1() {
  3.     //定义线程池
  4.     ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(2, 4, 30,
  5.                     TimeUnit.SECONDS,
  6.                     new ArrayBlockingQueue<>(10),
  7.                     Executors.defaultThreadFactory(),
  8.                     new ThreadPoolExecutor.DiscardPolicy());
  9.     //异步执行
  10.     Future<String> getUserName = threadPoolExecutor.submit(() -> {
  11.         //do something...
  12.         return "kdyzm";
  13.     });
  14.     //异步执行
  15.     Future<Integer> getUserAge = threadPoolExecutor.submit(() -> {
  16.         //do something...
  17.         return 12;
  18.     });
  19.     //拼装回调结果
  20.     try {
  21.         UserInfo user = new UserInfo();
  22.         user.setName(getUserName.get());
  23.         user.setAge(getUserAge.get());
  24.         log.info(JsonUtils.toPrettyString(user));
  25.     } catch (InterruptedException | ExecutionException e) {
  26.         e.printStackTrace();
  27.     }
  28. }
  30. @Data
  31. static class UserInfo {
  32.     private String name;
  33.     private Integer age;
  34. }

多几个submit一起执行，最后集中get获取最终结果。

这种方式任务一旦多了，就会显得代码很乱，一堆的变量名会让代码可读性很差。

方法二：CompletableFuture.allOf写法

其代码形式如下

  1. @Test
  2. public void test2() {
  3.     try {
  4.         UserInfo userInfo = new UserInfo();
  5.         
  6.         CompletableFuture.allOf(
  7.             	//异步执行
  8.                 CompletableFuture.runAsync(() -> {
  9.                     userInfo.setName("kdyzm");
  10.                 }),
  11.             	//异步执行
  12.                 CompletableFuture.runAsync(() -> {
  13.                     userInfo.setAge(12);
  14.                 })
  15.         //同步返回
  16.         ).get();
  18.         log.info(JsonUtils.toPrettyString(userInfo));
  19.     } catch (InterruptedException | ExecutionException e) {
  20.         e.printStackTrace();
  21.     }
  22. }
  24. @Data
  25. static class UserInfo {
  26.     private String name;
  27.     private Integer age;
  28. }

这种方法使用了CompletableFuture的API，通过将多个异步任务收集起来统一调度最后通过一个get方法同步到主线程。比直接使用Future简化了些。

方法三：CompletableFuture::join写法

其代码形式如下

  1. @Test
  2. public void test3(){
  3.     UserInfo userInfo = new UserInfo();
  4.     Arrays.asList(
  5. 			//异步执行
  6.             CompletableFuture.supplyAsync(()->{
  7.                 return "kdyzm";
  8.             //回调执行
  9.             }).thenAccept(name->{
  10.                 userInfo.setName(name);
  11.             }),
  13.         	//异步执行
  14.             CompletableFuture.supplyAsync(()->{
  15.                 return 12;
  16.             //回调执行
  17.             }).thenAccept(age->{
  18.                 userInfo.setAge(age);
  19.             })
  20.         
  21.         //等待所有线程执行完毕
  22.     ).forEach(CompletableFuture::join);
  24.     log.info(JsonUtils.toPrettyString(userInfo));
  26. }
  28. @Data
  29. static class UserInfo {
  30.     private String name;
  31.     private Integer age;
  32. }

这种写法和上面的写法相比具有更高的可读性，但是它也有缺点：thenAccept只能接收一个返回值，如果想处理多个值，则没有办法，只能使用方法2。

总结

几种写法中第二、三种写法比较常见，使用起来也更加方便，两者各有优缺点：方法2能处理多个返回值，方法3可读性更高。但是无论是方法2还是方法3，它们的使用总是要记住相关的API，使用起来总不是很顺手，可读性虽然方法3更强一些，但是总还是差点意思。此时我就有了自己设计一个简单的并行处理工具类的想法，既要易用，还要可读性高。

二、并行处理工具类设计

1、设计模式选型

因为平时比较喜欢链式调用的API，所以一开始一开始设计，我就想用建造者模式来实现这个工具类。关于建造者模式，详情可以看我之前的文章：设计模式（六）：建造者模式 。建造者模式在实际应用中的特点就是链式调用，无论是StringBuilder还是lombok的@Data注解，都使用了建造者模式。

2、第一版代码

仿照方法三，我开发了第一版代码

  1. import lombok.Data;
  2. import lombok.extern.slf4j.Slf4j;
  4. import java.util.ArrayList;
  5. import java.util.List;
  6. import java.util.concurrent.CompletableFuture;
  7. import java.util.function.Consumer;
  8. import java.util.function.Supplier;
  10. /**
  11.  * @author kdyzm
  12.  */
  13. @Slf4j
  14. public class ConcurrentWorker {
  16.     private List<Task> workers = new ArrayList<>();
  18.     public static ConcurrentWorker runner() {
  19.         return new ConcurrentWorker();
  20.     }
  22.     public <R> ConcurrentWorker addTask(Consumer<? super R> action, Supplier<R> value) {
  23.         Task<R> worker = new Task<>(action, value);
  24.         this.workers.add(worker);
  25.         return this;
  26.     }
  28.     public void run() {
  29.         workers.forEach(item -> {
  30.             CompletableFuture completableFuture = CompletableFuture.supplyAsync(item.getValue());
  31.             item.setCompletableFuture(completableFuture);
  32.         });
  33.         workers
  34.                 .stream()
  35.                 .map(
  36.                         item -> {
  37.                             return item.completableFuture.thenAccept(item.getAction());
  38.                         }
  39.                 )
  40.                 .forEach(CompletableFuture::join);
  41.     }
  43.     @Data
  44.     public static class Task<R> {
  45.         private Consumer<? super R> action;
  46.         private Supplier<R> value;
  47.         private CompletableFuture<R> completableFuture;
  49.         public Task(Consumer<? super R> action, Supplier<R> value) {
  50.             this.action = action;
  51.             this.value = value;
  52.         }
  53.     }
  54. }

这段代码一共不到60行，使用了Lambda表达式和函数式编程相关的API对方法三进行改造，最终使用效果如下

  1. @Test
  2.     public void test() {
  4.         UserInfo userInfo = new UserInfo();
  6.         ConcurrentWorker.runner()
  7.             	//添加任务
  8.                 .addTask(userInfo::setName, () -> {
  9.                     //延迟1000毫秒打印线程执行情况
  10.                     try {
  11.                         Thread.sleep(1000);
  12.                     } catch (InterruptedException e) {
  13.                         e.printStackTrace();
  14.                     }
  15.                     log.info(Thread.currentThread().getName()+"-name");
  16.                     return "张三";
  17.                 })
  18.             	//添加任务
  19.                 .addTask(userInfo::setAge, () -> {
  20.                     //延迟1000毫秒打印线程执行情况
  21.                     try {
  22.                         Thread.sleep(1000);
  23.                     } catch (InterruptedException e) {
  24.                         e.printStackTrace();
  25.                     }
  26.                     log.info(Thread.currentThread().getName()+"-age");
  27.                     return 13;
  28.                 })
  29.             	//执行任务
  30.                 .run();
  31.         log.info(JsonUtils.toPrettyString(userInfo));
  32.     }
  34.     @Data
  35.     static class UserInfo {
  36.         private String name;
  37.         private Integer age;
  38.         private String sex;
  39.     }

它的使用方式就是

  1. ConcurrentWorker.runner()
  2.                 .addTask(setter function, return_value function )
  3.     			.addTask(setter function, return_value function)
  4.     			.run()

可以看到易用性够了，可读性也很好，但是它的缺点和方法三一样，都只能接收一个参数，毕竟它是根据方法3封装的，接下来改造代码让它支持多参数处理。

3、第二版代码

已知，第一版代码已经支持了如下形式的功能

  1. ConcurrentWorker.runner()
  2.                 .addTask(setter function, return_value function )
  3.     			.addTask(setter function, return_value function)
  4.     			.run()

现在我想添加以下形式的重载方法

  1. .addTask(handle function)

没错，就一个参数，在这个方法中可以任意设置对象值。最终使用的效果如下

  1. @Test
  2. public void test() {
  4.     UserInfo userInfo = new UserInfo();
  6.     ConcurrentWorker.runner()
  7.             .addTask(userInfo::setName, () -> {
  8.                 try {
  9.                     Thread.sleep(1000);
  10.                     log.info(Thread.currentThread().getName());
  11.                 } catch (InterruptedException e) {
  12.                     e.printStackTrace();
  13.                 }
  14.                 log.info(Thread.currentThread().getName()+"-name");
  15.                 return "张三";
  16.             })
  17.             .addTask(userInfo::setAge, () -> {
  18.                 try {
  19.                     Thread.sleep(1000);
  20.                     log.info(Thread.currentThread().getName());
  21.                 } catch (InterruptedException e) {
  22.                     e.printStackTrace();
  23.                 }
  24.                 log.info(Thread.currentThread().getName()+"-age");
  25.                 return 13;
  26.             })
  27.         	//新方法：处理任意多属性值填充
  28.             .addTask(()->{
  29.                 try {
  30.                     Thread.sleep(1000);
  31.                 } catch (InterruptedException e) {
  32.                     e.printStackTrace();
  33.                 }
  34.                 log.info(Thread.currentThread().getName()+"-sex");
  35.                 userInfo.setSex("男");
  36.             })
  37.             .run();
  38.     log.info(JsonUtils.toPrettyString(userInfo));
  39. }
  41. @Data
  42. static class UserInfo {
  43.     private String name;
  44.     private Integer age;
  45.     private String sex;
  46. }

完整工具类方法如下

  1. import lombok.Data;
  2. import lombok.extern.slf4j.Slf4j;
  4. import java.util.ArrayList;
  5. import java.util.List;
  6. import java.util.concurrent.CompletableFuture;
  7. import java.util.function.Consumer;
  8. import java.util.function.Supplier;
  10. /**
  11.  * @author kdyzm
  12.  */
  13. @Slf4j
  14. public class ConcurrentWorker {
  16.     private List<Task> workers = new ArrayList<>();
  18.     public static ConcurrentWorker runner() {
  19.         return new ConcurrentWorker();
  20.     }
  22.     public <R> ConcurrentWorker addTask(Consumer<? super R> action, Supplier<R> value) {
  23.         Task<R> worker = new Task<>(action, value);
  24.         this.workers.add(worker);
  25.         return this;
  26.     }
  28.     public <R> ConcurrentWorker addTask(Runnable runnable) {
  29.         Task<R> worker = new Task<>(runnable);
  30.         this.workers.add(worker);
  31.         return this;
  32.     }
  34.     public void run() {
  35.         workers.forEach(item -> {
  36.             int taskType = item.getTaskType();
  37.             CompletableFuture completableFuture = null;
  38.             switch (taskType) {
  39.                 case TaskType.RETURN_VALUE:
  40.                     completableFuture = CompletableFuture.supplyAsync(item.getValue());
  41.                     break;
  42.                 case TaskType.VOID_RETURN:
  43.                     completableFuture = CompletableFuture.runAsync(item.getRunnable());
  44.                     break;
  45.                 default:
  46.                     break;
  47.             }
  48.             item.setCompletableFuture(completableFuture);
  49.         });
  50.         workers
  51.                 .stream()
  52.                 .map(
  53.                         item -> {
  54.                             int taskType = item.getTaskType();
  55.                             switch (taskType) {
  56.                                 case TaskType.RETURN_VALUE:
  57.                                     return item.completableFuture.thenAccept(item.getAction());
  58.                                 default:
  59.                                     return item.completableFuture.thenAccept(temp->{
  60.                                         //空
  61.                                     });
  62.                             }
  63.                         }
  64.                 )
  65.                 .forEach(CompletableFuture::join);
  66.     }
  68.     @Data
  69.     public static class Task<R> {
  70.         private Consumer<? super R> action;
  71.         private Supplier<R> value;
  72.         private CompletableFuture<R> completableFuture;
  73.         private Runnable runnable;
  74.         private int taskType;
  76.         public Task(Consumer<? super R> action, Supplier<R> value) {
  77.             this.action = action;
  78.             this.value = value;
  79.             this.taskType = TaskType.RETURN_VALUE;
  80.         }
  83.         public Task(Runnable runnable) {
  84.             this.runnable = runnable;
  85.             this.taskType = TaskType.VOID_RETURN;
  86.         }
  87.     }
  90.     public static class TaskType {
  92.         /**
  93.          * 有返回值的
  94.          */
  95.         public static final int RETURN_VALUE = 1;
  97.         /**
  98.          * 没有返回值的
  99.          */
  100.         public static final int VOID_RETURN = 2;
  101.     }
  102. }

我将任务类型分为两种，并使用TaskType类封装成常量值：1表示任务执行回调有返回值；2表示任务执行没有返回值，属性填充将在任务执行过程中完成，该类型任务使用Runnable接口实现。

4、工具类jar包

相关代码我已经打包成jar包上传到maven中央仓库，可以通过引入以下maven依赖使用ConcurrentWorker工具类

  1. <dependency>
  2.     <groupId>cn.kdyzm</groupId>
  3.     <artifactId>kdyzm-util</artifactId>
  4.     <version>0.0.2</version>
  5. </dependency>


最后，欢迎关注我的博客：https://blog.kdyzm.cn

END.

原文链接：https://www.cnblogs.com/kuangdaoyizhimei/p/18344600

 友情链接：直通硅谷 　点职佳 　北美留学生论坛

Java热门文章

  • 【IDEA】Application context not configured for this file
  • [IDEA] chapter_reader - idea看小说插件 idea阅读插件 idea摸鱼插件
  • thrift常见异常及原因分析
  • Java 调用 PaddleDetection 模型
  • JVM 详解，大白话带你认识 JVM
  • spring-boot-2.0.3不一样系列之番外篇 - @Configuration、Condition与@Conditional
  • Java Comparator.comparing比较导致空指针异常的解决
  • Java使用EasyExcel进行单元格合并的问题详解
  • 死磕 java集合之ArrayBlockingQueue源码分析
  • Java基础之集合Set详解

    • Java推荐文章

  • 面试场景题：一次关于线程池使用场景的讨论。
  • 前端黑科技：使用 JavaScript 实现网页扫码功能
  • 聊一聊 Netty 数据搬运工 ByteBuf 体系的设计与实现
  • 简单设计一个JAVA并行处理工具类
  • 【杂谈】JPA乐观锁改悲观锁遇到的一些问题与思考
  • 认识netty的基本组件
  • Jenkins如何使用CrumbIssuer防御CSRF攻击
  • 如何平稳地从nacos迁移到r-nacos？
  • 线程池遇到父子任务，有大坑，要注意！
  • Java JVM——12. 垃圾回收理论概述

    • 本站QQ群：前端 618073944 | Java 606181507 | Python 626812652 | C/C++ 612253063 | 微信 634508462 | 苹果 692586424 | C#/.net 182808419 | PHP 305140648 | 运维 608723728

    W3xue 的所有内容仅供测试，对任何法律问题及风险不承担任何责任。通过使用本站内容随之而来的风险与本站无关。
    关于我们  |  意见建议  |  捐助我们  |  报错有奖  |  广告合作、友情链接（目前9元/月）请联系QQ：27243702 沸活量
    皖ICP备17017327号-2　皖公网安备34020702000426号