We have added two new classes to the ones used in the previous section. These classes are:

Let's see the details of both classes.

We have implemented our algorithm in the ConcurrentMainRecommendation class to obtain the list of recommended products to a customer. This class implements the main() method that receives as a parameter the ID of the customer whose recommended products we want to obtain. We have the following code:

    public static void main(String[] args) {
        String user = args[0];
        Path file = Paths.get("data");
        try {
            Date start, end;
            start=new Date();

We have used different stream to transform the data in the final solution. The first one loads the whole list of the Product objects from its files:

            List<Product> productList = Files
                .walk(file, FileVisitOption.FOLLOW_LINKS)
                .parallel()
                .filter(f -> f.toString().endsWith(".txt"))
                .collect(ConcurrentLinkedDeque<Product>::new
                 ,new ConcurrentLoaderAccumulator(), ConcurrentLinkedDeque::addAll);

This stream has the following elements:

Once we have the list of products, we are going to organize those products in a map using the identifier of the customer as the key for that map. We use the ProductReview class to store the information of the customers of the products. We will create as many ProductReview objects as reviews have a Product. We use the following stream to make the transformation:

        Map<String, List<ProductReview>> productsByBuyer=productList
                .parallelStream()
                .<ProductReview>flatMap(p -> p.getReviews().stream().map(r -> new ProductReview(p, r.getUser(), r.getValue())))
                .collect(Collectors.groupingByConcurrent( p -> p.getBuyer()));

This stream has the following elements:

The next stream is the most important stream of this example. We take the products purchased by a customer and generate the recommendations to that customer. It's a two-phase process made by one stream. In the first phase, we obtain the users that purchased the products purchased by the original customer. In the second phase, we generate a map with the products purchased by those customers with all the reviews of the products made by those customers. This is the code for that stream:

            Map<String,List<ProductReview>> recommendedProducts=productsByBuyer.get(user)
                    .parallelStream()
                    .map(p -> p.getReviews())
                    .flatMap(Collection::stream)
                    .map(r -> r.getUser())
                    .distinct()
                    .map(productsByBuyer::get)
                    .flatMap(Collection::stream)
                    .collect(Collectors.groupingByConcurrent(p -> p.getTitle()));

We have the following elements in that stream:

To finish our recommendation algorithm, we need one last step. For every product, we want to calculate its average score in the reviews and sort the list in descending order to show in the first place the top-rated products. To make that transformation, we use an additional stream:

        List<ProductRecommendation> recommendations = recommendedProducts
                    .entrySet()
                    .parallelStream()
                    .map(entry -> new
                     ProductRecommendation(
                         entry.getKey(),
                         entry.getValue().stream().mapToInt(p -> p.getValue()).average().getAsDouble()))
                    .sorted()
                    .collect(Collectors.toList());
            end=new Date();
         recommendations. forEach(pr -> System.out.println (pr.getTitle()+": "+pr.getValue()));

            System.out.println("Execution Time: "+(end.getTime()- start.getTime()));

        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

We process the map obtained in the previous step. For each product, we process its list of reviews generating a ProductRecommendation object. The value of this object is calculated as the average value of each review using a stream using the mapToInt() method to transform the stream of ProductReview objects into a stream of integers and the average() method to get the average value of all the numbers in the string.

Finally, in the recommendations ConcurrentLinkedDeque class, we have a list of ProductRecommendation objects. We sort that list using an other stream with the sorted() method. We use that stream to write the final list in the console.

To implement this example, we have used the ConcurrentLoaderAccumulator class used as the accumulator function in the collect() method that transforms the stream of Path objects with the routes of all the files to process into the ConcurrentLinkedDeque class of Product objects. This is the source code of this class:

public class ConcurrentLoaderAccumulator implements
        BiConsumer<ConcurrentLinkedDeque<Product>, Path> {

    @Override
    public void accept(ConcurrentLinkedDeque<Product> list, Path path) {

        Product product=ProductLoader.load(path);
        list.add(product);
        
    }
}

It implements the BiConsumer interface. The accept() method uses the ProducLoader class (explained earlier in this chapter) to load the product information from the file and add the resultant Product object in the ConcurrentLinkedDeque class received as parameters.

As with other examples in the book, we have implemented a serial version of this example to check that parallel streams improve the performance of the application. To implement this serial version, we have to follow these steps:

You can see the serial version of this example in the source code of the book.

To compare the serial and concurrent versions of our recommendation system, we have obtained the recommended products for three users:

For these three users, we have executed both versions using the JMH framework (http://openjdk.java.net/projects/code-tools/jmh/) that allows you to implement micro benchmarks in Java. Using a framework for benchmarking is a better solution that simply measures time using the methods such as currentTimeMillis() or nanoTime(). We have executed them 10 times in a computer with a four-core processor and calculated the medium execution time of those 10 times. These are the results in milliseconds:

We can draw the following conclusions:

If we compare the concurrent and serial versions, for example, the second user using the speed-up, we obtain the following result: