How to create multiple threads for each request item

Question

I am trying to process the below code using multithreading at the order level.

List<String> orders = Arrays.asList("order1", "order2", 
                   "order3", "order4", "order1");

Current sequential execution:

orders.stream().forEach(order -> {
    rules.forEach(rule -> {
        finalList.add(beanMapper.getBean(rule)
                .applyRule(createTemplate.apply(getMetaData.apply(rule), command),
                           order));
    });
});

I have tried using:

orders.parallelStream().forEach(order -> {}} // code snippet.

But it is changing the rules.forEach(rule -> {}} order.

For example:
Input :

 List<String> orders = Arrays.asList("order1", "order2", 
                         "order3", "order4", "order1");
 List<String> rules = Arrays.asList("rule1", "rule2", "rule3");

Expected output:

order1 with rule1, rule2, rule3
order2 with rule1, rule2, rule3

Actual output with parallelStream():

order1 with rule3, rule1, rule2
order1 with rule2, rule1, rule3

I am not bothered about the order of orders, but I am bothered about eh order of rules. Orders can process in any order, but rules should execute in the same order for each order.

Please help.

Answer 1

You can use :

orders.stream().parallel().forEachOrdered(// Your rules logic goes here. )

ForEachOrdered guarantees to maintain the order of the Stream.

So for your reference:

orders.stream().parallel().forEachOrdered( order -> {

            rules.stream().parallel().forEachOrdered ( rule -> {

                 System.out.println( " Order : " + order + " rule :" + rule);
            });

        });

Note : While we can do this , performance should be closely watched because parellelism and order do not marry each other very well!

Output

 Order : order1 rule :rule1
 Order : order1 rule :rule2
 Order : order1 rule :rule3
 Order : order2 rule :rule1
 Order : order2 rule :rule2
 Order : order2 rule :rule3
 Order : order3 rule :rule1
 Order : order3 rule :rule2
 Order : order3 rule :rule3
 Order : order4 rule :rule1
 Order : order4 rule :rule2
 Order : order4 rule :rule3
 Order : order1 rule :rule1
 Order : order1 rule :rule2
 Order : order1 rule :rule3

Answer 2

You add elements to the finalList from different threads at the same time. This is causing mixing results of applying rules to different orders (rules aren't being grouped by their orders).

You can fix it by creating a temporary list for each order and then synchronously merging all temporary lists in a finalList.

Here is how you can do it using Stream-API (Java 9+):

List<AppliedRule> finalList = orders.parallelStream().map(order ->
        rules.stream().map(rule -> applyRule(order, rule)).collect(Collectors.toList())
).collect(Collectors.flatMapping(Collection::stream, Collectors.toList()));

Note: Collectors.flatMapping() is used here instead of simple flatMap to run flat mapping synchronously during stream collection.

---

Java 8 analog:

List<AppliedRule> finalList = orders.parallelStream().map(order ->
        rules.stream().map(rule -> applyRule(order, rule)).collect(Collectors.toList())
).collect(Collectors.toList())
        .stream()
        .flatMap(Collection::stream)
        .collect(Collectors.toList());