Я получаю ответы разного размера на один и тот же запрос при использовании параллельного потока в Java 8. Может кто-нибудь дать представление о том, как с этим справиться?

Question

Моя структура приложения выглядит следующим образом: Контроллер, который вызывает класс обслуживания, который, в свою очередь, вызывает класс репозитория. Хранилище (objectRepo) возвращает список объектов. Класс обслуживания анализирует этот список объектов, которые он получает из репозитория, и преобразует его в класс apiResponseWrapper. Контроллер передает этот apiResponseWrapper пользователю, не меняя его.

В моем классе обслуживания у меня есть код, который эффективно переводится в приведенный ниже пример кода:

List<Object> objectList = objectRepo.getList(paramMap);
List<Entity> entityList = new ArrayList<>();
objectList.parallelStream().forEach(object->{
//code to transform object
entityList.add(transformedObject);
});
return entityList;

Когда используется параллельный поток, как описано выше, размер ответа моего API изменяется при последовательных вызовах с одинаковыми параметрами, и при дальнейшей проверке некоторые объекты отсутствуют в ответах на несколько таких вызовов. При удалении parallelStream размер ответа остается тем же, и все ожидаемые объекты находятся в результате.

Я хотел бы знать, существует ли какой-либо конкретный способ / какой-либо набор руководств, которые бы помогли мне использовать параллельный поток без потери данных.

Answer 1

Это настройка вашей конструкции для работы. Правильнее всего то, что говорит @Eugene.

Вы не можете использовать не безопасный поток список в контексте paralel. Вместо этого используйте поточно-ориентированную структуру как SynchroneusQueue или альтернативно ad synchroneus block.

List<Object> objectList = objectRepo.getList(paramMap);
SynchroneusQueue<Entity> entityList = new SynchroneusQueue<>();
objectList.parallelStream().forEach(object->{
//code to transform object
entityList.add(transformedObject);
});
return entityList;

Это более или менее эквивалентно:

List<Object> objectList = objectRepo.getList(paramMap);
List<Entity> entityList = new ArrayList<>();
objectList.parallelStream().forEach(object->{
//code to transform object
synchronized(entityList) {
    entityList.add(transformedObject);
}
});
return entityList;

Answer 2

не делайте этого ... используйте:

List<Entity> entityList = objectList.stream()
          .parallel()
          .map(object -> transform to transformedObject)
          .collect(Collectors.toList())

В конце концов entityList не является потокобезопасной коллекцией, которую вы обновляете из нескольких потоков, это также не рекомендуется в документации как ненужные побочные эффекты.

Также очень мало случаев, когда parallel улучшит производительность, вам понадобится лотов объектов, чтобы иметь какой-либо значимый эффект.