1

I have a List of TicketDTO objects where every TicketDTO needs to go through a function to convert the data to TicketDataDTO. What I want here is to reduce the time it takes for this code to run because when the list size is bigger, it takes a lot of time to convert it and it's unacceptable for fetching the data through a GET mapping. However, when I try to implement ForkJoinPool along with the parallelStream) (code below) to get it done, my return List` is empty. Can someone tell me what am I doing wrong?

@Override
public List<TicketDataDTO> getOtrsTickets(String value, String startDate, String endDate, String product, String user) {
  // TODO Implement threads
  List<TicketDTO> tickets = ticketDao.findOtrsTickets(value, startDate, endDate, product, user);
  Stream<TicketDTO> ticketsStream = tickets.parallelStream();
  List<TicketDataDTO> data = new ArrayList<TicketDataDTO>();
  
  ForkJoinPool forkJoinPool = new ForkJoinPool(6);
  
  forkJoinPool.submit(() -> {
    try {
      ticketsStream.forEach(ticket -> data.add(createTicketData(ticket)));
    } catch (Exception e) {
      throw new RuntimeException(e);
    }
  });
  
  forkJoinPool.shutdown();
  
  //ticketsStream.forEach(ticket -> data.add(createTicketData(ticket)));

  return data;

createTicketData is just a function with two for loops and one switch loop to create some new columns I need as an output.

martinspielmann
  • 536
  • 6
  • 19
madkingaerys
  • 25
  • 6
  • 1
    You're modifiying a single ArrayList from multiple threads without doing any kind of synchronization. Even once you fix your immediate issue this will result in errors. I recommend you read https://docs.oracle.com/en/java/javase/11/docs/api/java.base/java/util/stream/package-summary.html and swap to using a non-side-effectful approach using `collect(Collectors.toList())`. Additionally the use of custom ForkJoinPools in parallel streams is not a supported feature. Better to explicitly call `ExecutorService.submit` or `invokeAll` to get a list of Futures and wait on them. – MikeFHay Nov 24 '20 at 11:51

1 Answers1

2

Additional to calling shutdown() on the ForkJoinPool, you have to wait for its termination like

forkJoinPool.awaitTermination(Long.MAX_VALUE, TimeUnit.NANOSECONDS);

If you do not wait for the termination, data will be returned before the threads have the chance to add their results to it.

See How to wait for all threads to finish, using ExecutorService? for more details

martinspielmann
  • 536
  • 6
  • 19