ExecutorService 类型中的方法 invokeAll 不适用于参数错误

Question

我正在做一个项目，我将在其中拥有不同的捆绑包。举个例子，假设我有 5 个捆绑包，每个捆绑包都有一个方法名称 process。

以下是我应该做的事情-

1. 我需要使用多线程代码并行调用所有这 5 个捆绑包process 方法，然后写入数据库。我不确定这样做的正确方法是什么？我应该有五个线程吗？每个捆绑一个线程？但是在那种情况下会发生什么，假设如果我有 50 个包，那么我将有 50 个线程？
2. 而且，我也想拥有超时功能。如果任何捆绑包花费的时间超过了我们设置的阈值，那么它应该会超时并记录为该捆绑包花费了很多时间的错误。

我所做的以下尝试很可能存在缺陷，并且错误处理绝不是完整的。但不知何故，我总是在这一行出错-

pool.invokeAll

错误是-

The method invokeAll(Collection<? extends Callable<T>>, long, TimeUnit) in the type ExecutorService is not applicable for the arguments (List<ModelFramework.ProcessBundleHolderEntry>, int, TimeUnit)

下面是我的方法，它将以多线程方式调用所有捆绑包的process method。

public void processEvents(final Map<String, Object> eventData) {
    ExecutorService pool = Executors.newFixedThreadPool(5);
    List<ProcessBundleHolderEntry> entries = new ArrayList<ProcessBundleHolderEntry>();

    Map<String, String> outputs = (Map<String, String>)eventData.get(BConstants.EVENT_HOLDER);

    for (BundleRegistration.BundlesHolderEntry entry : BundleRegistration.getInstance()) {
        ProcessBundleHolderEntry processBundleHolderEntry = new ProcessBundleHolderEntry(entry, outputs);
        entries.add(processBundleHolderEntry);
    }

    try {

        // somehow I always get an error at invokeAll method. Is there anything wrong?
        List<Future<Object>> futures = pool.invokeAll(entries, 30, TimeUnit.SECONDS);
        for (int i = 0; i < futures.size(); i++) {
            // This works since the list of future objects are in the
            // same sequential order as the list of entries
            Future<Object> future = futures.get(i);
            ProcessBundleHolderEntry entry = entries.get(i);
            if (!future.isDone()) {
                // log error for this entry
            }
        }
    } catch (InterruptedException e) {
        // handle this exception!
    }
}

其次，我在 ModelFramework 类中添加的线程的 Callable 实现

public class ProcessBundleHolderEntry implements Callable {
    private BundleRegistration.BundlesHolderEntry entry;
    private Map<String, String> outputs;

    public ProcessBundleHolderEntry(BundleRegistration.BundlesHolderEntry entry, Map<String, String> outputs) {
        this.entry = entry;
        this.outputs = outputs;
    }

    public Object call() throws Exception {
        final Map<String, String> response = entry.getPlugin().process(outputs);
        // write to the database.
        System.out.println(response);
        return response;
    }
}

谁能帮我解决我遇到的错误？还有谁能告诉我上述方法是否有任何问题，或者有没有更好更有效的方法来做同样的事情？

对此的任何帮助将不胜感激。

Answer 1

我所做的以下尝试很可能存在缺陷，并且错误处理绝不是完整的。但不知何故，我总是在这一行出错-

我认为问题在于，如果您希望 invokeAll(...) 返回 List<Future<Object>>，ProcessBundleHolderEntry 应该实现 Callable<Object>。我刚刚编译了你的代码，这解决了问题。

真的在我看来它应该实现Callable<Map<String, String>>。那么调用方法应该返回正确的类型：

public Map<String, String> call() throws Exception {

然后invokeAll(...) 方法将返回正确的List<Future<Map<String, String>>。

一个不同的（虽然有点奇怪）的想法是从call() 方法中返回this。拥有ProcessBundleHolderEntry implements Callable<ProcessBundleHolderEntry> 并在条目中记录响应，然后从call() 返回this。然后你不需要在条目上做一个get(i) 来匹配它。然后，您将条目、输出和响应放在一个对象中。

Answer 2

没关系，我发现了实际问题

public class ProcessBundleHolderEntry implements Callable {

应该定义为

public class ProcessBundleHolderEntry implements Callable<Object> {

为了匹配下面声明中的类型Object

List<Future<Object>> futures = pool.invokeAll(entries, 30, TimeUnit.SECONDS);

invokeAll 方法的签名是

<T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks,
                              long timeout, TimeUnit unit)