首页 > 编程语言 >简单设计一个JAVA并行处理工具类

简单设计一个JAVA并行处理工具类

时间：2024-08-06 10:07:58浏览次数：10  
标签：JAVA 并行处理 private item CompletableFuture userInfo return 工具 public

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

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

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

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

方法一：Future写法

其代码形式如下

@Test
public void test1() {
    //定义线程池
    ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(2, 4, 30,
                    TimeUnit.SECONDS,
                    new ArrayBlockingQueue<>(10),
                    Executors.defaultThreadFactory(),
                    new ThreadPoolExecutor.DiscardPolicy());
    //异步执行
    Future<String> getUserName = threadPoolExecutor.submit(() -> {
        //do something...
        return "kdyzm";
    });
    //异步执行
    Future<Integer> getUserAge = threadPoolExecutor.submit(() -> {
        //do something...
        return 12;
    });
    //拼装回调结果
    try {
        UserInfo user = new UserInfo();
        user.setName(getUserName.get());
        user.setAge(getUserAge.get());
        log.info(JsonUtils.toPrettyString(user));
    } catch (InterruptedException | ExecutionException e) {
        e.printStackTrace();
    }
}

@Data
static class UserInfo {
    private String name;
    private Integer age;
}

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

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

方法二：CompletableFuture.allOf写法

其代码形式如下

@Test
public void test2() {
    try {
        UserInfo userInfo = new UserInfo();
        
        CompletableFuture.allOf(
            	//异步执行
                CompletableFuture.runAsync(() -> {
                    userInfo.setName("kdyzm");
                }),
            	//异步执行
                CompletableFuture.runAsync(() -> {
                    userInfo.setAge(12);
                })
        //同步返回
        ).get();

        log.info(JsonUtils.toPrettyString(userInfo));
    } catch (InterruptedException | ExecutionException e) {
        e.printStackTrace();
    }
}

@Data
static class UserInfo {
    private String name;
    private Integer age;
}

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

方法三：CompletableFuture::join写法

其代码形式如下

@Test
public void test3(){
    UserInfo userInfo = new UserInfo();
    Arrays.asList(
			//异步执行
            CompletableFuture.supplyAsync(()->{
                return "kdyzm";
            //回调执行
            }).thenAccept(name->{
                userInfo.setName(name);
            }),

        	//异步执行
            CompletableFuture.supplyAsync(()->{
                return 12;
            //回调执行
            }).thenAccept(age->{
                userInfo.setAge(age);
            })
        
        //等待所有线程执行完毕
    ).forEach(CompletableFuture::join);

    log.info(JsonUtils.toPrettyString(userInfo));

}

@Data
static class UserInfo {
    private String name;
    private Integer age;
}

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

总结

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

二、并行处理工具类设计

1、设计模式选型

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

2、第一版代码

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

import lombok.Data;
import lombok.extern.slf4j.Slf4j;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.function.Consumer;
import java.util.function.Supplier;

/**
 * @author kdyzm
 */
@Slf4j
public class ConcurrentWorker {

    private List<Task> workers = new ArrayList<>();

    public static ConcurrentWorker runner() {
        return new ConcurrentWorker();
    }

    public <R> ConcurrentWorker addTask(Consumer<? super R> action, Supplier<R> value) {
        Task<R> worker = new Task<>(action, value);
        this.workers.add(worker);
        return this;
    }

    public void run() {
        workers.forEach(item -> {
            CompletableFuture completableFuture = CompletableFuture.supplyAsync(item.getValue());
            item.setCompletableFuture(completableFuture);
        });
        workers
                .stream()
                .map(
                        item -> {
                            return item.completableFuture.thenAccept(item.getAction());
                        }
                )
                .forEach(CompletableFuture::join);
    }

    @Data
    public static class Task<R> {
        private Consumer<? super R> action;
        private Supplier<R> value;
        private CompletableFuture<R> completableFuture;

        public Task(Consumer<? super R> action, Supplier<R> value) {
            this.action = action;
            this.value = value;
        }
    }
}

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

@Test
    public void test() {

        UserInfo userInfo = new UserInfo();

        ConcurrentWorker.runner()
            	//添加任务
                .addTask(userInfo::setName, () -> {
                    //延迟1000毫秒打印线程执行情况
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    log.info(Thread.currentThread().getName()+"-name");
                    return "张三";
                })
            	//添加任务
                .addTask(userInfo::setAge, () -> {
                    //延迟1000毫秒打印线程执行情况
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    log.info(Thread.currentThread().getName()+"-age");
                    return 13;
                })
            	//执行任务
                .run();
        log.info(JsonUtils.toPrettyString(userInfo));
    }

    @Data
    static class UserInfo {
        private String name;
        private Integer age;
        private String sex;
    }

它的使用方式就是

ConcurrentWorker.runner()
                .addTask(setter function, return_value function )
    			.addTask(setter function, return_value function)
    			.run()

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

3、第二版代码

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

ConcurrentWorker.runner()
                .addTask(setter function, return_value function )
    			.addTask(setter function, return_value function)
    			.run()

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

.addTask(handle function)

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

@Test
public void test() {

    UserInfo userInfo = new UserInfo();

    ConcurrentWorker.runner()
            .addTask(userInfo::setName, () -> {
                try {
                    Thread.sleep(1000);
                    log.info(Thread.currentThread().getName());
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                log.info(Thread.currentThread().getName()+"-name");
                return "张三";
            })
            .addTask(userInfo::setAge, () -> {
                try {
                    Thread.sleep(1000);
                    log.info(Thread.currentThread().getName());
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                log.info(Thread.currentThread().getName()+"-age");
                return 13;
            })
        	//新方法：处理任意多属性值填充
            .addTask(()->{
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                log.info(Thread.currentThread().getName()+"-sex");
                userInfo.setSex("男");
            })
            .run();
    log.info(JsonUtils.toPrettyString(userInfo));
}

@Data
static class UserInfo {
    private String name;
    private Integer age;
    private String sex;
}

完整工具类方法如下

import lombok.Data;
import lombok.extern.slf4j.Slf4j;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.function.Consumer;
import java.util.function.Supplier;

/**
 * @author kdyzm
 */
@Slf4j
public class ConcurrentWorker {

    private List<Task> workers = new ArrayList<>();

    public static ConcurrentWorker runner() {
        return new ConcurrentWorker();
    }

    public <R> ConcurrentWorker addTask(Consumer<? super R> action, Supplier<R> value) {
        Task<R> worker = new Task<>(action, value);
        this.workers.add(worker);
        return this;
    }

    public <R> ConcurrentWorker addTask(Runnable runnable) {
        Task<R> worker = new Task<>(runnable);
        this.workers.add(worker);
        return this;
    }

    public void run() {
        workers.forEach(item -> {
            int taskType = item.getTaskType();
            CompletableFuture completableFuture = null;
            switch (taskType) {
                case TaskType.RETURN_VALUE:
                    completableFuture = CompletableFuture.supplyAsync(item.getValue());
                    break;
                case TaskType.VOID_RETURN:
                    completableFuture = CompletableFuture.runAsync(item.getRunnable());
                    break;
                default:
                    break;
            }
            item.setCompletableFuture(completableFuture);
        });
        workers
                .stream()
                .map(
                        item -> {
                            int taskType = item.getTaskType();
                            switch (taskType) {
                                case TaskType.RETURN_VALUE:
                                    return item.completableFuture.thenAccept(item.getAction());
                                default:
                                    return item.completableFuture.thenAccept(temp->{
                                        //空
                                    });
                            }
                        }
                )
                .forEach(CompletableFuture::join);
    }

    @Data
    public static class Task<R> {
        private Consumer<? super R> action;
        private Supplier<R> value;
        private CompletableFuture<R> completableFuture;
        private Runnable runnable;
        private int taskType;

        public Task(Consumer<? super R> action, Supplier<R> value) {
            this.action = action;
            this.value = value;
            this.taskType = TaskType.RETURN_VALUE;
        }


        public Task(Runnable runnable) {
            this.runnable = runnable;
            this.taskType = TaskType.VOID_RETURN;
        }
    }


    public static class TaskType {

        /**
         * 有返回值的
         */
        public static final int RETURN_VALUE = 1;

        /**
         * 没有返回值的
         */
        public static final int VOID_RETURN = 2;
    }
}

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

4、工具类jar包

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

<dependency>
    <groupId>cn.kdyzm</groupId>
    <artifactId>kdyzm-util</artifactId>
    <version>0.0.2</version>
</dependency>


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

END.

标签：JAVA,并行处理,private,item,CompletableFuture,userInfo,return,工具,public
From： https://www.cnblogs.com/kuangdaoyizhimei/p/18344600

相关文章

  • Java IO流 详细版
    一、什么是IO流IO：输入/输出(Input/Output)流：是一种抽象概念,是对数据传输的总称.也就是说数据在设备间的传输称为流,流的本质是数据传输IO流就是用来处理设备间数据传输问题的.常见的应用:文件复制;文件上传;文件下载二、IO流的分类按照数据的流向输入流：读数据输出......
  • 「Java开发指南」MyEclipse for Spring参数 - 注释器（一）
    MyEclipsev2024.1离线版下载MyEclipse技术交流群：742336981欢迎一起进群讨论1.Spring开发中的注释器注解器是MyEclipse中带有Spring支持的代码助手的一部分，它帮助开发人员进行基于注解的开发。虽然大多数Java开发人员都了解如何从技术上使用Java注释，但对于特定技术，哪些注释是......
  • java笔记5
    9.抽象类与接口抽象类抽象类的概念抽象类是面向对象编程中不能被实例化的一种类，它通常被用作基类，为其他类提供公共的接口或实现。什么是抽象抽象是将现实世界中的复杂事物简化为基本特征的过程，它关注对象的本质特征，忽略细节,只说明做什么，不说明怎么做。为什么要抽象抽象帮......
  • Hashtable工具类 - C#小函数类推荐
          此文记录的是一个Hashtable的增强版本，以前没泛型的时候笔者喜欢用Hashtable，性能刚刚的。/***Hashtable工具类AustinLiu刘恒辉ProjectManagerandSoftwareDesignerE-Mail:[email protected]:http://lzhdim.cnblogs.comD......
  • 2024，Java开发在中国市场还有发展前景吗？
    随着2024年的到来，Java作为一种经典而强大的编程语言，依然在中国的软件开发市场中扮演着重要角色。然而，许多人对Java的未来发展前景持有不同的看法。让我们来探讨一下当前情况和未来的走向。Java程序员真的过剩了吗？2023年,各大求职网站上的java职位数量相对于其他技术岗位来......
  • Java之类和对象
    目录static关键字1修饰属性2修饰方法 final构造方法基本语法this关键字代码块定义普通代码块 构造代码块静态代码块匿名对象toString 总结static关键字1修饰属性Java的静态属性和类相关,和具体的实例无关.换句话说,同一个类的不同实例共用同一个静......
  • Java调用Linux命令行SpringBoot
    原文链接：https://blog.csdn.net/u014628771/article/details/108308337需求在Linux服务器上的SpringBoot程序中，调用gitclone，之后遍历git仓库中的所有文件。遍历git仓库可以使用File类实现，现在的问题是需要在SpringBoot程序中调用gitclone命令。实现方式使用Javanative的Proc......
  • 基于Java中的SSM框架实现在线音乐网站系统项目【项目源码+论文说明】
    基于Java中的SSM框架实现在线音乐网站系统演示摘要本文讲述了使用JSP语言及HTML5语言及MySql数据库技术开发的音乐网站的设计与实现。本文所讲的JSP音乐系统是通过所学的知识创办一个类似于QQ音乐或者酷狗音乐性质的网站平台，使所有对国内外音乐欣赏感兴趣的人都可以不必再......
  • Apache Flink开发时选择Java还是Scala作为编程语言
    在ApacheFlink的开发过程中，选择Java还是Scala作为编程语言是一个重要的决策点。这两种语言各有其独特的优势和特点，适合不同的开发场景和需求。以下是对这一选择的详细探讨，旨在帮助开发者更好地理解并做出合理的选择。一、ApacheFlink简介ApacheFlink是一个开源的分布式......
  • 【Java基础】03选择结构
    if分支ifif(条件){代码块;}if...else...if(条件){代码块1;}else{代码块2;}if...elseif...else...if(条件1){代码块1;}elseif(条件2){代码块2;//elseif可以写多个}else{代码块3;//else可以省略不写}if嵌套if（条件1){......

赞助商

阅读排行

网站主页关于我们联系我们网站地图

本网站内容转载自其他媒体，侵权联系[admin##ips99.com]。

Copyright © 2020-2023 IPS99 版权所有 IPS99