【已解决】利用 Java 多线程并发编程提高数据处理效率

简介: 【已解决】利用 Java 多线程并发编程提高数据处理效率

🎉工作场景中遇到这样一个需求:根据主机的 IP 地址联动更新其他模型的相关信息。需求很简单,只涉及一般的数据库联动查询以及更新操作,然而在编码实现过程中发现,由于主机的数量很多,导致循环遍历查询、更新时花费很长的时间,调用一次接口大概需要 30-40 min 时间才能完成操作。

💡因此,为了有效缩短接口方法的执行时间,便考虑使用多线程并发编程方法,利用多核处理器并行执行的能力,通过异步处理数据的方式,便可以大大缩短执行时间,提高执行效率。


📍这里使用可重用固定线程数的线程池 FixedThreadPool,并使用 CountDownLatch 并发工具类提供的并发流程控制工具作为配合使用,保证多线程并发编程过程中的正常运行:


首先,通过 Runtime.getRuntime().availableProcessors() 方法获取运行机器的 CPU 线程数,用于后续设置固定线程池的线程数量。

其次,判断任务的特性,如果为计算密集型任务则设置线程数为 CPU 线程数+1,如果为 IO 密集型任务则设置线程数为 2 * CPU 线程数,由于在方法中需要与数据库进行频繁的交互,因此属于 IO 密集型任务。

之后,对数据进行分组切割,每个线程处理一个分组的数据,分组的组数与线程数保持一致,并且还要创建计数器对象 CountDownLatch,调用构造函数,初始化参数值为线程数个数,保证主线程等待所有子线程运行结束后,再进行后续的操作。

然后,调用 executorService.execute() 方法,重写 run 方法编写业务逻辑与数据处理代码,执行完当前线程后记得将计数器减1操作。

最后,当所有子线程执行完成后,关闭线程池。

✨在省略工作场景中的业务逻辑代码后,通用的处理方法示例如下所示:

public ResponseData updateHostDept() {
    // ...
    List<Map> hostMapList = mongoTemplate.find(query, Map.class, "host");
        // split the hostMapList for the following multi-threads task
        // return the number of logical CPUs
        int processorsNum = Runtime.getRuntime().availableProcessors();
        // set the threadNum as 2*(the number of logical CPUs) for handling IO Tasks,
        // if Computing Tasks set the threadNum as (the number of logical  CPUs) + 1
        int threadNum = processorsNum * 2;  
        // the number of each group data 
        int eachGroupNum = hostMapList.size() / threadNum; 
        List<List<Map>> groupList = new ArrayList<>();
        for (int i = 0; i < threadNum; i++) {
            int start = i * eachGroupNum;
            if (i == threadNum - 1) {
                int end = mapList.size();
                groupList.add(hostMapList.subList(start, end));
            } else {
                int end = (i+1) * eachGroupNum;
                groupList.add(hostMapList.subList(start, end));
            }
        }
        // update data by using multi-threads asynchronously
        ExecutorService executorService = Executors.newFixedThreadPool(threadNum/2);
        CountDownLatch countDownLatch = new CountDownLatch(threadNum);
        for (List<Map> group : groupList) {
            executorService.execute(()->{
                try {
                    for (Map map : group) {
                      // update the data in mongodb
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                } finally {
                  // let counter minus one 
                    countDownLatch.countDown();  
                }
            });
        }
        try {
          // main thread donnot execute until all child threads finish
            countDownLatch.await();  
        } catch (Exception e) {
            e.printStackTrace();
        }
        // remember to shutdown the threadPool
        executorService.shutdown();  
        return ResponseData.success();
}

🎉那么在使用多线程异步更新的策略后,从当初调用接口所需的大概时间为 30-40 min 下降到了 8-10 min,大大提高了执行效率。


💡需要注意的是,这里使用的 newFixedThreadPool 创建线程池,它有一个缺陷就是,它的阻塞队列默认是一个无界队列,默认值为 Integer.MAX_VALUE 极有可能会造成 OOM 问题。因此,一般可以使用 ThreadPoolExecutor 来创建线程池,自己可以指定等待队列中的线程个数,避免产生 OOM 问题。

public ResponseData updateHostDept() {
    // ...
    List<Map> hostMapList = mongoTemplate.find(query, Map.class, "host");
        // split the hostMapList for the following multi-threads task
        // return the number of logical CPUs
        int processorsNum = Runtime.getRuntime().availableProcessors();
        // set the threadNum as 2*(the number of logical CPUs) for handling IO Tasks,
        // if Computing Tasks set the threadNum as (the number of logical  CPUs) + 1
        int threadNum = processorsNum * 2;  
        // the number of each group data 
        int eachGroupNum = hostMapList.size() / threadNum; 
        List<List<Map>> groupList = new ArrayList<>();
        for (int i = 0; i < threadNum; i++) {
            int start = i * eachGroupNum;
            if (i == threadNum - 1) {
                int end = mapList.size();
                groupList.add(hostMapList.subList(start, end));
            } else {
                int end = (i+1) * eachGroupNum;
                groupList.add(hostMapList.subList(start, end));
            }
        }
        // update data by using multi-threads asynchronously
        ThreadPoolExecutor executor = new ThreadPoolExecutor(5, 8, 30L, TimeUnit.SECONDS, 
                new ArrayBlockingQueue<>(100));
        CountDownLatch countDownLatch = new CountDownLatch(threadNum);
        for (List<Map> group : groupList) {
            executor.execute(()->{
                try {
                    for (Map map : group) {
                      // update the data in mongodb
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                } finally {
                  // let counter minus one 
                    countDownLatch.countDown();  
                }
            });
        }
        try {
          // main thread donnot execute until all child threads finish
            countDownLatch.await();  
        } catch (Exception e) {
            e.printStackTrace();
        }
        // remember to shutdown the threadPool
        executor.shutdown();  
        return ResponseData.success();
}

在上述的代码中,核心线程数和最大线程数分别为 5 和 8,并没有设置的很大的值,因为如果如果设置的很大,线程间频繁的上下文切换也会增加时间消耗,反而不能最大程度上发挥多线程的优势。至于如何选择合适的参数,需要根据机器的参数以及任务的类型综合考虑决定。


🎉最后补充一点,如果想要通过非编码的方式获取机器的 CPU 线程个数也很简单,windows 系统通过任务管理器,选择 “性能”,便可以查看 CPU 线程个数的情况,如下图所示:2e718f6de6d94987ae2f2182c528d9c8.png

🎉从上图可以看到,我的机器中内核是八个 CPU,但是通过超线程技术一个物理的 CPU 核心可以模拟成两个逻辑 CPU 线程,因此我的机器是支持8核16线程的。

相关文章
|
16小时前
|
Java 网络安全
Java中的Socket编程详解
Java中的Socket编程详解
|
18小时前
|
安全 Java 开发者
Java并发编程之深入理解synchronized关键字
【6月更文挑战第26天】在Java并发编程的世界中,synchronized关键字扮演着守护者的角色,它保护着共享资源的完整性。本文将深入探讨synchronized关键字的内部机制、使用方法及其对性能的影响,帮助开发者更好地利用这一工具来构建线程安全的应用。
|
18小时前
|
监控 Java 调度
Java并发编程:深入理解线程池
【6月更文挑战第26天】在Java并发编程的世界中,线程池是提升应用性能、优化资源管理的关键组件。本文将深入探讨线程池的内部机制,从核心概念到实际应用,揭示如何有效利用线程池来处理并发任务,同时避免常见的陷阱和错误实践。通过实例分析,我们将了解线程池配置的策略和对性能的影响,以及如何监控和维护线程池的健康状况。
6 1
|
1天前
|
存储 Java 调度
深入探索Java并发编程:ConcurrentSkipListSet的高效使用与实现原理
深入探索Java并发编程:ConcurrentSkipListSet的高效使用与实现原理
17 0
|
1天前
|
存储 安全 Java
深入探索Java并发编程:ArrayBlockingQueue详解
深入探索Java并发编程:ArrayBlockingQueue详解
12 5
|
1天前
|
存储 设计模式 并行计算
CopyOnWriteArrayList:深入理解Java中的线程安全List原理和应用
CopyOnWriteArrayList:深入理解Java中的线程安全List原理和应用
7 0
|
1天前
|
Java 关系型数据库 开发者
Java编程设计原则:构建稳健、可维护的软件基石
Java编程设计原则:构建稳健、可维护的软件基石
6 1
|
1天前
|
缓存 并行计算 安全
【并发编程系列一】并发编年史:线程的双刃剑——从优势到风险的全面解析
【并发编程系列一】并发编年史:线程的双刃剑——从优势到风险的全面解析
|
1天前
|
Java 测试技术 开发者
Java并发编程:深入理解线程池
本文将带领读者深入了解Java中的线程池,探索其内部机制、使用场景以及如何有效地利用线程池来提高程序的性能和可维护性。我们将通过实例演示如何创建和配置线程池,并讨论常见的并发模式和最佳实践。文章旨在为开发者提供实用的线程池应用知识,帮助他们在面对多线程编程挑战时,能够设计出更加高效和稳定的系统。
|
1天前
|
安全 Java 开发者
如何在Java中实现线程安全的单例模式
如何在Java中实现线程安全的单例模式