在Java中如何确保在多线程情况下，所有线程都完成再继续执行任务？

╰+攻爆jí腚メ 2023-10-10 17:25 27阅读 0赞

#### 1.使用awaitTermination ####

awaitTermination是executorService自带的方法，用来确保所有的线程任务都执行完毕。

例如下使用线程池来执行100个不同的 SQL 语句，将查询结果存储在一个 List 集合中，具体实现如下：

1.  定义一个 Runnable 接口的实现类，用来执行 SQL 查询任务，该类需要在构造函数中接收 SQL 语句和存储结果的 List 集合对象。

public class SqlQueryTask implements Runnable {
        private final String sql;
        private final List<Map<String, Object>> resultList;
    
        public SqlQueryTask(String sql, List<Map<String, Object>> resultList) {
            this.sql = sql;
            this.resultList = resultList;
        }
    
        @Override
        public void run() {
            // 执行 SQL 查询，并将结果存储在 resultList 中
            Connection conn = null;
            Statement stmt = null;
            ResultSet rs = null;
            try {
                conn = DriverManager.getConnection("jdbc:mysql://localhost:3306/test", "root", "");
                stmt = conn.createStatement();
                rs = stmt.executeQuery(sql);
                while (rs.next()) {
                    Map<String, Object> rowMap = new HashMap<>();
                    // 获取查询结果的每一列的名称和值，并将其存储在 Map 中
                    ResultSetMetaData metaData = rs.getMetaData();
                    int count = metaData.getColumnCount();
                    for (int i = 1; i <= count; i++) {
                        String columnName = metaData.getColumnName(i);
                        Object columnValue = rs.getObject(i);
                        rowMap.put(columnName, columnValue);
                    }
                    // 将该行数据添加到结果 List 集合中
                    resultList.add(rowMap);
                }
            } catch (SQLException e) {
                e.printStackTrace();
            } finally {
                try {
                    if (rs != null) rs.close();
                    if (stmt != null) stmt.close();
                    if (conn != null) conn.close();
                } catch (SQLException e) {
                    e.printStackTrace();
                }
            }
    
        }
    }

*  2、在主线程中，创建一个 List 集合对象存储查询结果，再创建一个线程池对象，然后将100个 SQL 查询任务提交给线程池。

public static void main(String[] args) throws InterruptedException {
        // 创建一个 List 集合存储查询结果
        List<Map<String, Object>> resultList = new ArrayList<>();
    
        // 创建一个线程池对象
        ExecutorService executorService = Executors.newFixedThreadPool(10);
    
        // 循环提交100个 SQL 查询任务给线程池
        for (int i = 1; i <= 100; i++) {
            String sql = "SELECT * FROM table" + i + ";";
            SqlQueryTask task = new SqlQueryTask(sql, resultList);
            executorService.execute(task);
        }
    
        // 等待所有任务执行完成
        executorService.shutdown();
        executorService.awaitTermination(Long.MAX_VALUE, TimeUnit.NANOSECONDS);
    
        // 输出查询结果
        for (Map<String, Object> rowMap : resultList) {
            System.out.println(rowMap);
        }
    }

#### 2、使用CountDownLatch ####

在多线程环境下，需要保证线程安全，避免出现线程竞争和线程之间的互相干扰。针对这种需要保证多线程运行情况下执行 1000 条 SQL 语句的需求，可以使用 CountDownLatch 来实现线程同步和确保所有的 SQL 查询任务都已经执行完成后再执行后面的代码逻辑。

1.  在主线程中，创建一个 CountDownLatch 对象并设置计数器为 1000，再创建一个 List 集合对象存储查询结果。
2.  在每个 SQL 查询任务中，执行 SQL 查询，并将查询结果存储在 List 集合中，最后将 CountDownLatch 的计数器减 1。
3.  在主线程中，调用 CountDownLatch 的 await() 方法在所有 SQL 查询任务执行完成前阻塞线程，直到计数器减为 0 时，所有任务都已经执行完成可以继续往下执行，此时可以对查询结果进行汇总处理。

下面是一个示例代码：

import java.sql.*;
    import java.util.*;
    import java.util.concurrent.*;
    
    public class SqlQueryDemo {
        private static final int TASK_COUNT = 1000;
    
        public static void main(String[] args) throws InterruptedException {
            String url = "jdbc:mysql://localhost:3306/test";
            String user = "root";
            String password = "";
            int threads = 10;
    
            // 创建 CountDownLatch 对象
            CountDownLatch latch = new CountDownLatch(TASK_COUNT);
    
            // 创建线程池
            ExecutorService executorService = Executors.newFixedThreadPool(threads);
    
            // 创建 List 存储查询结果
            List<Map<String, Object>> resultList = new ArrayList<>();
    
            // 提交 SQL 查询任务
            for (int i = 1; i <= TASK_COUNT; i++) {
                String sql = "SELECT * FROM table" + i + ";";
                executorService.execute(new SqlQueryTask(url, user, password, sql, resultList, latch));
            }
    
            // 等待所有任务执行完成
            latch.await();
    
            // 关闭线程池
            executorService.shutdown();
    
            // 合并查询结果
            List<Map<String, Object>> totalResult = new ArrayList<>();
            for (Map<String, Object> result : resultList) {
                totalResult.addAll(result.values());
            }
            // TODO: 执行后面的代码逻辑，使用 totalResult 对查询结果进行汇总处理
        }
    }
    
    class SqlQueryTask implements Runnable {
        private String url;
        private String user;
        private String password;
        private String sql;
        private List<Map<String, Object>> resultList;
        private CountDownLatch latch;
    
        public SqlQueryTask(String url, String user, String password, String sql, List<Map<String, Object>> resultList, CountDownLatch latch) {
            this.url = url;
            this.user = user;
            this.password = password;
            this.sql = sql;
            this.resultList = resultList;
            this.latch = latch;
        }
    
        @Override
        public void run() {
            Connection conn = null;
            Statement stmt = null;
            ResultSet rs = null;
            try {
                // 执行 SQL 查询，并将查询结果存储在 List 中
                conn = DriverManager.getConnection(url, user, password);
                stmt = conn.createStatement();
                rs = stmt.executeQuery(sql);
                Map<String, Object> result = new HashMap<>();
                while (rs.next()) {
                    ResultSetMetaData metaData = rs.getMetaData();
                    int count = metaData.getColumnCount();
                    Map<String, Object> row = new HashMap<>();
                    for (int i = 1; i <= count; i++) {
                        String columnName = metaData.getColumnName(i);
                        Object columnValue = rs.getObject(i);
                        row.put(columnName, columnValue);
                    }
                    result.put(sql, row);
                }
                resultList.add(result);
                // CountDownLatch 计数器减 1
                latch.countDown();
            } catch (SQLException e) {
                e.printStackTrace();
            } finally {
                try {
                    if (rs != null) rs.close();
                    if (stmt != null) stmt.close();
                    if (conn != null) conn.close();
                } catch (SQLException e) {
                    e.printStackTrace();
                }
            }
        }
    }

在以上示例代码中，当每个 SQL 查询任务执行完毕后都将 CountDownLatch 计数器减 1，当计数器减为 0 时，主线程的 await() 方法返回，程序继续执行后面的逻辑，汇总所有 SQL 查询结果并进行处理。