当前位置：首页 > news >正文

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

news 来源：原创 2024/5/3 20:13:06

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

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

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

首先，通过 Runtime.getRuntime().availableProcessors() 方法获取运行机器的 CPU 线程数，用于后续设置固定线程池的线程数量。
其次，判断任务的特性，如果为计算密集型任务则设置线程数为 CPU 线程数+1，如果为 IO 密集型任务则设置线程数为 2 * CPU 线程数，由于在方法中需要与数据库进行频繁的交互，因此属于 IO 密集型任务。
之后，对数据进行分组切割，每个线程处理一个分组的数据，分组的组数与线程数保持一致，并且还要创建计数器对象 CountDownLatch，调用构造函数，初始化参数值为线程数个数，保证主线程等待所有子线程运行结束后，再进行后续的操作。
然后，调用 executorService.execute() 方法，重写 run 方法编写业务逻辑与数据处理代码，执行完当前线程后记得将计数器减1操作。
最后，当所有子线程执行完成后，关闭线程池。

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

public ResponseData updateHostDept() {
		//to do something
		List<Map> mapList = mongoTemplate.find(query, Map.class, "host");
        //split the mapList 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;  
        int eachGroupNum = mapList.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(mapList.subList(start, end));
            } else {
                int end = (i+1) * eachGroupNum;
                groupList.add(mapList.subList(start, end));
            }
        }
        //使用线程池创建多线程异步更新数据
        ExecutorService executorService = Executors.newFixedThreadPool(threadNum);
        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 {
                    countDownLatch.countDown();  //计数器减一
                }
            });
        }
        try {
            countDownLatch.await();  //让主线程等待所有子线程执行完成
        } catch (Exception e) {
            e.printStackTrace();
        }
        executorService.shutdown();  //关闭线程池
        return ResponseData.success();
}