用于并行处理的 Java Executor Service

Question

在系统上工作以支持并行的多个数据库查询。考虑到要从每个查询中查询大量数据，要求将每个数据库查询与其他查询分开。这意味着，一个数据库/表上的负载不应影响其他表查询。我使用 ExecutorService 在 Java 中开发了一个解决方案。每个数据库使用一个 ExecutorService（固定大小，1 个线程）。我维护一个数据库名称到 ExecutorService 的映射，并在接收查询请求时将调用定向到相应的执行程序服务。考虑到可以并行查询一百个数据库，不确定 ExecutorService 是否是正确的选择......！我做了一些估值，初步结果看起来还不错。我在使用此解决方案时遇到的一个挑战是，当我动态创建 ExecutorServices 时，我很难在应用程序停止时优雅地关闭它们。

解决此问题的其他方法是维护一个全局（即跨所有数据库）查询工作线程池，并将它们随机重用于传入请求。但是，这并不能保证所有数据库查询都具有同等的优先级。

DatasetFactory.java

public class DataSetExecutorFactory {

        private static Map<String, DataSetExecutor> executorMap = Collections.synchronizedMap(new HashMap<String, DataSetExecutor>());
    public static DataSetExecutor getDataSetExecutor(String dbName){
            DataSetExecutor executor = null;

            executor = executorMap.get(dbName);
            if(executor == null){
                executor = new DataSetExecutor(dbName);
                executorMap.put(dbName, executor);
            }
            return executor;
        }
    }
}

DataSetExecutor.java

public class DataSetExecutor {

    private ExecutorService executor = Executors.newFixedThreadPool(1);
    public List<Map<String, Object>> execQuery(String collecName, Map<String, Object> queryParams){
        //Construct Query job. 
        //QueryWorker extends 'Callable' and does the actual query to DB
        QueryWorker queryWorker = new QueryWorker(Map queryParams);

        Future<QueryResult> result = null;
        try{
            result = executor.submit(queryWorker);
        }catch (Exception e){
            //Catch Exception here
            e.printStackTrace();
        }
    }

Answer 1

我认为您误解了 ExecutorService 的工作原理。与其为每个数据库创建 ExecutorService，不如将单个 ExecutorService 作为大小为 n 的 FixedThreadPool（n = 数据库数或最大并行查询数）。线程池将为您完成并行处理工作。您只需将数据库名称作为将提交给 ExecutorService 的 QueryWorker 的一部分进行跟踪。

这也使关闭变得容易，因为 ThreadPool 会自动清理未使用的线程，您只需在应用程序关闭时将其关闭一次。

尽管如此，由于所有这些并行处理都发生在同一个 JVM 和同一台机器上，您可能会遇到内存或 CPU 限制，具体取决于您的查询强度。