I'm getting Different sized responses for same request when using parallel stream in Java 8. Can anyone provide insights on how to handle this?

user_5485767 0
Viewed 273

My application structure is as follows: A controller that calls a service class that in turn calls a repository class. The repository(objectRepo) returns a list of Objects. The service class parses thru this list of objects it gets from repository and transforms it into an apiResponseWrapper class. The controller gives this apiResponseWrapper to user without changing it.

In my Service class, I have code that effectively translates to below sample code:

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

When parallelstream is used like above, the size of the response of my API varies on consecutive calls, with the exact same parameters, and on further inspection certain entities are absent in responses of few of those calls. On removal of parallelStream the size of response remains same, and all expected entities are in the result.

I would like to know if there is any specific way/any set of guidelines to be followed that would help me to use parallelstream without any data being dropped.

java-8 lambda java
2 Answers

don't do that... use:

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

After all entityList is not a thread-safe collection that you update from multiple threads, this is also discouraged in the documentation as un-needed side effects.

Also there are very few cases where parallel would improve performance, you would need lots of Objects to have any meaningful effect.

user_10593720

This is tweaking your construct to work. The correct thing to do is what @Eugene says.

You can not use a non thread safe list in a paralel context. Use instead thread safe structure as SynchroneusQueue or alternativly 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;

This is more or less equivalent to:

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;
user_20443020
Related