Метод invokeAll в типе ExecutorService неприменим для аргументов error

Question

Я работаю над проектом, в котором у меня будут разные Связки. Возьмем пример: предположим, что у меня есть 5 связок, и каждый из этих пучков будет иметь имя метода process.

Ниже перечислены вещи, я должен DO-

1. Мне нужно вызвать все эти 5 Связки process метод параллельного использования многопоточного кода, а затем записать в базу данных. Я не уверен, что это правильный способ сделать это? Должен ли я иметь пять потоков? Один поток для каждого пакета? Но что будет в этом сценарии, предположим, если у меня будет 50 пакетов, тогда у меня будет 50 потоков?
2. А также, я хочу иметь функцию тайм-аута. Если какие-либо расслоения занимают много времени, чем установленная нами установка порога, тогда он должен получить тайм-аут и регистрироваться как ошибка, которую этот пакет занимает много времени.

Следующая попытка, которую я совершил, скорее всего, испорчена, и обработка ошибок отнюдь не завершена. Но так или иначе, я всегда получаю ошибку на этой линии-

pool.invokeAll

И ошибка

IS-

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! 
    } 
} 

Во-вторых, реализация Callable для нитей, которые я добавил в классе ModelFramework

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

Следующая попытка, которую я совершил, скорее всего, является недостатком, и обработка ошибок отнюдь не завершена. Но так или иначе, я всегда получаю ошибку на этой линии-

Я думаю, что проблема в том, что ProcessBundleHolderEntry должен реализовать Callable<Object>, если вы хотите, чтобы invokeAll(...) вернуть List<Future<Object>>. Я только что скомпилировал ваш код, и это устраняет проблему.

Действительно, мне кажется, что он должен реализовать Callable<Map<String, String>>. Тогда вызов метода должен возвращать правильный тип:

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

Тогда метод invokeAll(...) будет возвращать правильный List<Future<Map<String, String>>.

Другая (хотя и немного странная) идея состоит в том, чтобы вернуть this из метода call(). Имейте ProcessBundleHolderEntry implements Callable<ProcessBundleHolderEntry> и запишите ответ в записи перед возвратом this из call(). Тогда вам не нужно делать 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)