Scatter Gather Pattern
Overview
Scatter Gather enterprise integration pattern is used for scenarios such as "best quote", where we need to request information from several suppliers and decide which one provides us with the best price for the requested item.
Github: https://github.com/gitorko/project62
Scatter Gather Pattern
So we have a book product and we need to fetch the price from various sources and at max we can wait for 3 seconds. You could use a Thread.sleep or Threads join() method but then if the tasks complete before 3 seconds the tasks will still wait for 3 seconds before returning.
Code
We can use a CountDownLatch to wait for the prices to be fetched. It will wait only for 3 seconds and return the prices fetched.
1package com.demo.project62._24_scattergather.latch;
2
3import java.util.Map;
4import java.util.concurrent.ConcurrentHashMap;
5import java.util.concurrent.CountDownLatch;
6import java.util.concurrent.ExecutorService;
7import java.util.concurrent.Executors;
8import java.util.concurrent.TimeUnit;
9
10import lombok.AllArgsConstructor;
11import lombok.SneakyThrows;
12import org.junit.jupiter.api.Test;
13
14public class ScatterGatherLatchTest {
15
16 ExecutorService threadPool = Executors.newCachedThreadPool();
17
18 @Test
19 public void test() {
20 Map<String, Float> book1Prices = new ScatterGatherLatchTest().getPrices("book1");
21 System.out.println(book1Prices);
22 }
23
24 @SneakyThrows
25 private Map<String, Float> getPrices(String productId) {
26 Map<String, Float> prices = new ConcurrentHashMap<>();
27 CountDownLatch latch = new CountDownLatch(3);
28 threadPool.submit(new FetchData("http://amazon", productId, prices, latch));
29 threadPool.submit(new FetchData("http://ebay", productId, prices, latch));
30 threadPool.submit(new FetchData("http://flipkart", productId, prices, latch));
31 latch.await(3, TimeUnit.SECONDS);
32 threadPool.shutdown();
33 return prices;
34 }
35
36 @AllArgsConstructor
37 class FetchData implements Runnable {
38
39 String url;
40 String productId;
41 Map<String, Float> prices;
42 CountDownLatch latch;
43
44 @SneakyThrows
45 @Override
46 public void run() {
47 if (url.contains("amazon")) {
48 //http fetch from amazon
49 System.out.println("Fetching price from amazon!");
50 TimeUnit.SECONDS.sleep(2);
51 prices.put("amazon", 2.35f);
52 latch.countDown();
53 }
54
55 if (url.contains("ebay")) {
56 System.out.println("Fetching price from ebay!");
57 //http fetch from ebay
58 TimeUnit.SECONDS.sleep(4);
59 prices.put("ebay", 2.30f);
60 latch.countDown();
61 }
62
63 if (url.contains("flipkart")) {
64 System.out.println("Fetching price from flipkart!");
65 //http fetch from flipkart
66 TimeUnit.SECONDS.sleep(1);
67 prices.put("flipkart", 2.10f);
68 latch.countDown();
69 }
70 }
71 }
72}
73
We can also use the invokeAll method
1package com.demo.project62._24_scattergather.invoke;
2
3import java.util.ArrayList;
4import java.util.List;
5import java.util.Map;
6import java.util.concurrent.Callable;
7import java.util.concurrent.ConcurrentHashMap;
8import java.util.concurrent.ExecutorService;
9import java.util.concurrent.Executors;
10import java.util.concurrent.TimeUnit;
11
12import lombok.AllArgsConstructor;
13import lombok.SneakyThrows;
14import org.junit.jupiter.api.Test;
15
16public class ScatterGatherInvokeTest {
17 ExecutorService threadPool = Executors.newCachedThreadPool();
18
19 @Test
20 public void test() {
21 Map<String, Float> book1Prices = new ScatterGatherInvokeTest().getPrices("book1");
22 System.out.println(book1Prices);
23 }
24
25 @SneakyThrows
26 private Map<String, Float> getPrices(String productId) {
27 Map<String, Float> prices = new ConcurrentHashMap<>();
28 List<Callable<Void>> tasks = new ArrayList<>();
29
30 tasks.add(new FetchData("http://amazon", productId, prices));
31 tasks.add(new FetchData("http://ebay", productId, prices));
32 tasks.add(new FetchData("http://flipkart", productId, prices));
33 threadPool.invokeAll(tasks, 3, TimeUnit.SECONDS);
34 threadPool.shutdown();
35 return prices;
36 }
37
38 @AllArgsConstructor
39 class FetchData implements Callable<Void> {
40
41 String url;
42 String productId;
43 Map<String, Float> prices;
44
45 @Override
46 @SneakyThrows
47 public Void call() throws Exception {
48 if (url.contains("amazon")) {
49 //http fetch from amazon
50 System.out.println("Fetching price from amazon!");
51 TimeUnit.SECONDS.sleep(2);
52 prices.put("amazon", 2.35f);
53 }
54
55 if (url.contains("ebay")) {
56 System.out.println("Fetching price from ebay!");
57 //http fetch from ebay
58 TimeUnit.SECONDS.sleep(4);
59 prices.put("ebay", 2.30f);
60 }
61
62 if (url.contains("flipkart")) {
63 System.out.println("Fetching price from flipkart!");
64 //http fetch from flipkart
65 TimeUnit.SECONDS.sleep(1);
66 prices.put("flipkart", 2.10f);
67 }
68 return null;
69 }
70 }
71}
72
73
We can also use the CompletableFuture.
1package com.demo.project62._24_scattergather.completable;
2
3import java.util.Map;
4import java.util.concurrent.CompletableFuture;
5import java.util.concurrent.ConcurrentHashMap;
6import java.util.concurrent.ExecutorService;
7import java.util.concurrent.Executors;
8import java.util.concurrent.TimeUnit;
9import java.util.concurrent.TimeoutException;
10
11import lombok.AllArgsConstructor;
12import lombok.SneakyThrows;
13import org.junit.jupiter.api.Test;
14
15public class ScatterGatherCompletableTest {
16 ExecutorService threadPool = Executors.newCachedThreadPool();
17
18 @Test
19 public void test() {
20 Map<String, Float> book1Prices = new ScatterGatherCompletableTest().getPrices("book1");
21 System.out.println(book1Prices);
22 }
23
24 @SneakyThrows
25 private Map<String, Float> getPrices(String productId) {
26 Map<String, Float> prices = new ConcurrentHashMap<>();
27
28 CompletableFuture<Void> task1 = CompletableFuture.runAsync(new FetchData("http://amazon", productId, prices));
29 CompletableFuture<Void> task2 = CompletableFuture.runAsync(new FetchData("http://ebay", productId, prices));
30 CompletableFuture<Void> task3 = CompletableFuture.runAsync(new FetchData("http://flipkart", productId, prices));
31
32 CompletableFuture<Void> allTasks = CompletableFuture.allOf(task1,task2,task3);
33 try {
34 allTasks.get(3, TimeUnit.SECONDS);
35 } catch (TimeoutException ex) {
36 //Do Nothing!
37 }
38 return prices;
39 }
40
41 @AllArgsConstructor
42 class FetchData implements Runnable {
43
44 String url;
45 String productId;
46 Map<String, Float> prices;
47
48 @Override
49 @SneakyThrows
50 public void run() {
51 if (url.contains("amazon")) {
52 //http fetch from amazon
53 System.out.println("Fetching price from amazon!");
54 TimeUnit.SECONDS.sleep(2);
55 prices.put("amazon", 2.35f);
56 }
57
58 if (url.contains("ebay")) {
59 System.out.println("Fetching price from ebay!");
60 //http fetch from ebay
61 TimeUnit.SECONDS.sleep(4);
62 prices.put("ebay", 2.30f);
63 }
64
65 if (url.contains("flipkart")) {
66 System.out.println("Fetching price from flipkart!");
67 //http fetch from flipkart
68 TimeUnit.SECONDS.sleep(1);
69 prices.put("flipkart", 2.10f);
70 }
71 }
72 }
73}
74
Result
1{amazon=2.35, flipkart=2.1}
Setup
1# Project 62
2
3Java Design Patterns
4
5[https://gitorko.github.io/design-patterns/](https://gitorko.github.io/design-patterns/)
6
7### Version
8
9Check version
10
11```bash
12$java --version
13openjdk version "21.0.3" 2024-04-16 LTS
14```
comments powered by Disqus