Java: результаты заказа, полученные из асинхронных задач

Question

У меня есть вычисление (шифрование CTR), которое требует результатов в точном порядке.

Для этого я создал многопоточный дизайн, который вычисляет указанные результаты, в этом случае результатом является ByteBuffer. Само вычисление, конечно, выполняется асинхронно, поэтому результаты могут быть доступны в любое время и в любом порядке. «Пользователь» - однопоточное приложение, которое использует результаты, вызывая метод, после чего байтовые буферы возвращаются в пул ресурсов указанным методом - управление ресурсами уже обрабатывается (с использованием поточного безопасного стека).

Теперь вопрос: мне нужно что-то, что агрегирует результаты и делает их доступными в в правильном порядке. Если следующий результат недоступен, метод, который должен вызвать пользователь, должен блокироваться до тех пор, пока он не будет. Кто-нибудь знает хорошую стратегию или класс в java.util.concurrent, который может возвращать асинхронно рассчитанные результаты в порядке?

Решение должно быть потокобезопасным. Я хотел бы избежать сторонних библиотек, Thread.sleep()/Thread.wait() и тегов, связанных с другими словами, отличными от «synchronized». Более того, задачи могут быть заданы, например, Исполнитель в правильном порядке, если это требуется. Это для исследований, поэтому не стесняйтесь использовать Java 1.6 или даже 1.7 конструкции.

Примечание: Я отметил те запросы, которые я хочу сохранить в классах, определенных в JRE и [шифровании], поскольку кому-то, возможно, уже пришлось иметь дело с этим, но сам вопрос касается исключительно Java & многопоточность.

Answer 1

Используйте executors framework:

ExecutorService executorService = Executors.newFixedThreadPool(5); 
List<Future> futures = executorService.invokeAll(listOfCallables); 
for (Future future : futures) { 
    //do something with future.get(); 
} 
executorService.shutdown(); 

listOfCallables будет List<Callable<ByteBuffer>>, что вы построили для работы с данными. Например:

list.add(new SubTaskCalculator(1, 20)); 
list.add(new SubTaskCalculator(21, 40)); 
list.add(new SubTaskCalculator(41, 60)); 

(произвольные диапазоны чисел, регулировать, что к вашей задаче)

.get() блоков, пока результат не является полным, но в то же время другие задачи, также работает, поэтому, когда вы достигните их, их .get() будет готов.

Answer 2

возвращение результата в нужном порядке тривиально. По мере того, как каждый результат приходит, храните его в arraylist, и как только у вас есть ВСЕ результаты, просто отберите arraylist. Вы можете использовать PriorityQueue, чтобы сохранить результаты отсортированы в любое время по мере их поступления, но нет смысла делать это, так как вы не будете использовать результаты, прежде чем все они придут в любом случае.

Итак, что вы могли бы сделать это:

Объявите класс «WorkItem», который содержит один из вашего ByteArray, и его порядкового номера, так что они могут быть отсортированы по порядковому номеру.

В ваших рабочих потоков, сделать что-то вроде этого:

...do work and produce a work_item... 

synchronized(LockObject) 
{ 
    ResultList.Add(work_item); 
    number_of_results++; 
    LockObject.notifyAll(); 
} 

В основном потоке, сделать что-то вроде этого:

synchronized(LockObject) 
    while(number_of_results != number_of_items) 
     LockObject.wait(); 
ResultList.Sort(); 
...go ahead and use the results... 

Answer 3

Мой новый ответ после получения лучшего понимания того, что вы хотите

Объявите класс «WorkItem», который содержит один из ваших bytearrays и его порядковый номер, чтобы их можно сортировать по порядковому номеру.

Используйте java.util.PriorityQueue, который сортируется по порядковому номеру. По существу, все, что нам нужно, это то, что первым элементом в очереди приоритетов в любой момент будет следующий элемент для обработки.

Каждый рабочий поток сохраняет свой результат в PriorityQueue и выдает NotifyAll на какой-либо объект блокировки.

Главный поток ждет объекта блокировки, а затем, если в очереди находятся элементы, и если порядковый номер первого (в порядке очереди, не вычеркнутого) элемента в очереди равен количеству обработанных элементов , то он удаляет объект и обрабатывает его. Если нет, он ждет. Если все предметы были произведены и обработаны, это делается.