当前位置:

首页 > 编程开发 > Java多线程并发编程提高数据处理效率是多少

Java多线程并发编程提高数据处理效率是多少

工作场景中遇到这样一个需求:根据主机的IP地址联动更新其他模型的相关信息。需求很简单,只涉及一般的数据库联动查询以及更新操作,然而在编码实现过程中发现,由于主机的数量很多,导致循环遍历查询、更新时花费很长的时间,调用一次接口大概需要30-40min时间才能完成操作。因此,为了有效缩短接口方法的执行时间,便考虑使用多线程并发编程方法,利用多核处理器并行执行的能力,通过异步处理数据的方式,便可以大大缩短执行时间,提高执行效率。这里使用可重用固定线程数的线程池FixedThreadPool,并使用CountDo

本文目录

      工作场景中遇到这样一个需求:根据主机的 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 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> 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 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 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> 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 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 线程个数的情况,如下图所示:

      Java多线程并发编程提高数据处理效率是多少

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

      本站声明:本文内容由网友自发贡献,版权归原作者所有,本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容,请联系bd@zhengruan.com
      作者最新文章
      编程开发 Java
      相关文章 更多
      Java单元测试入门:使用JUnit 5编写第一个测试用例
      Java单元测试入门:使用JUnit 5编写第一个测试用例

      面向初学者的JUnit 5实用入门指南。从测试概念、Maven依赖配置,到手把手编写首个单元测试用例及断言验证,助你掌握Java单元测试的核心流程。

      Java变量未初始化报错解析:局部变量与成员变量默认值差异
      Java变量未初始化报错解析:局部变量与成员变量默认值差异

      深入解析Java中局部变量未初始化编译报错与成员变量自动赋予默认值的机制差异,涵盖JVM内存分配、编译期确定性赋值检查与实际排查修复方案。

      Java获取当前时间并格式化输出:DateTimeFormatter与SimpleDateFormat
      Java获取当前时间并格式化输出:DateTimeFormatter与SimpleDateFormat

      详细讲解在Java中获取当前系统时间并进行格式化输出的两种核心方案:基于Java 8全新日期时间API的DateTimeFormatter与传统遗留的SimpleDateFormat,包含完整代码示例、线程安全机制对比及生产选型建议。

      Java中final与static final常量的定义及初始化规则
      Java中final与static final常量的定义及初始化规则

      深入解析Java中final变量与static final常量的定义、内存语义、初始化时机及合法赋值路径,涵盖编译期常量优化与实战避坑指南。

      Java抽象方法与接口方法的区别及选型建议
      Java抽象方法与接口方法的区别及选型建议

      深入对比Java中抽象类抽象方法与接口方法的语法演进、设计意图及运行期差异,并提供清晰的架构选型决策流程与实战范例。

      Java实现HTTP请求与JSON解析的完整代码示例
      Java实现HTTP请求与JSON解析的完整代码示例

      本文提供Java标准库HttpClient发起HTTP GET/POST请求并结合Jackson进行JSON数据反序列化与序列化的完整代码示例与工程实践指南。

      Java递归出现StackOverflowError的原因分析及栈空间优化方案
      Java递归出现StackOverflowError的原因分析及栈空间优化方案

      详细分析Java递归调用出现StackOverflowError的原因,结合代码说明线程栈、栈帧和递归深度之间的关系,并介绍终止条件排查、迭代改写、减少栈帧开销以及通过-Xss调整JVM栈空间的方法。

      Java空指针异常(NPE)的代码排查与修复实例
      Java空指针异常(NPE)的代码排查与修复实例

      针对Java中常见的NullPointerException(NPE),本文从堆栈日志定位、典型代码触发场景、具体修复代码对比到防御体系进行系统讲解,提供实用且易上手的排查与修复方案。

      selenium上传文件教程本地文件路径设置方法
      selenium上传文件教程本地文件路径设置方法

      深入解析Selenium处理文件上传的技术细节,涵盖标准HTML控件的路径直接注入法与非标准控件的系统级模拟方案,提供Windows与Linux环境下的路径设置最佳实践及常见异常排查。

      大数据开发语言选型:Java对比Scala与Python的优缺点及适用场景
      大数据开发语言选型:Java对比Scala与Python的优缺点及适用场景

      针对大数据工程场景,系统对比Java、Scala与Python在执行性能、生态结合、开发效率和维护成本上的核心差异,给出清晰的架构选型依据与落地方案。

      查看更多
      精品专题 更多
      装机必备
      装机必备

      正软商城装机必备专区,精选办公、浏览器、安全防护、影音播放、压缩解压、设计创作和系统工具等电脑常用正版软件,帮助用户快速完成新电脑软件配置。

      Windows
      Windows

      正软商城Windows软件专区,汇集适用于Windows电脑的办公、设计、安全防护、影音播放、开发工具和系统优化软件,提供软件介绍、系统要求、正版授权及购买下载服务。

      PDF教程
      PDF教程

      正软商城PDF教程频道提供PDF编辑、转换、合并、拆分、压缩及格式处理方法,同时介绍常用PDF软件和工具的使用技巧。

      Mac软件 更多
      讯飞智作 网页版
      讯飞智作 网页版
      Windows/ macOS/ Linux

      讯飞智作是一款基于语音合成技术、虚拟人驱动技术与数字人形象声音构建能力的 AI 数字人音视频创作平台。

      火山引擎 网页版
      火山引擎 网页版
      Windows/ macOS/ Linux

      火山引擎是字节跳动旗下企业级云与AI服务平台。聚焦大模型核心能力,依托字节自研豆包多模态大模型,凭借火山方舟一站式平台,提供模型精调、推理部署、智能应用搭建等全链路服务,结合高性能AI算力,助力各行业快速落地大模型数字化与智能化应用。

      Aperty macOS版
      Aperty macOS版
      macOS

      Aperty 是 Skylum 推出的 AI 人像修图软件,支持自然肤质修饰、瑕疵处理、面部与身体调整、预设及批量同步。

      WINDOWS 更多
      讯飞智作 网页版
      讯飞智作 网页版
      Windows/ macOS/ Linux

      讯飞智作是一款基于语音合成技术、虚拟人驱动技术与数字人形象声音构建能力的 AI 数字人音视频创作平台。

      火山引擎 网页版
      火山引擎 网页版
      Windows/ macOS/ Linux

      火山引擎是字节跳动旗下企业级云与AI服务平台。聚焦大模型核心能力,依托字节自研豆包多模态大模型,凭借火山方舟一站式平台,提供模型精调、推理部署、智能应用搭建等全链路服务,结合高性能AI算力,助力各行业快速落地大模型数字化与智能化应用。

      PixTV 网页版
      PixTV 网页版
      Windows

      PixTV 是一款在浏览器中使用的 AI 视觉内容创作平台,提供图片、视频、音频、文字生成、AI 分镜、局部重绘、提示词反推和无限画布等工具,适合短剧、漫剧、广告及社交媒体内容制作。