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/project01
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.basics.designpatterns._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.basics.designpatterns._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.basics.designpatterns._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 01 2 3Data Structure & Algorithms & Design Patterns 4 5[https://gitorko.github.io/grokking-the-coding-interview/](https://gitorko.github.io/grokking-the-coding-interview/) 6[https://gitorko.github.io/design-patterns/](https://gitorko.github.io/design-patterns/) 7 8### Version 9 10Check version 11 12```bash 13$java --version 14openjdk version "21.0.3" 2024-04-16 LTS 15``` 16 17### Online code editor 18 19https://rustpad.io/ 20 21https://collabedit.com/ 22 23https://app.coderpad.io/ 24 25https://codeshare.io/ 26 27### Topic 28 2901. Number 3002. String 3103. Map & Set 3204. Heap 3305. Sliding window / Two pointer 3406. Matrix / Grid 3507. Backtracking 3608. Pre-Sum 3709. DP 3810. Link List 3911. Binary Tree / BST 4012. Interval 4113. Binary Search 4214. Topological Sort 4315. Stack & Monotonic Stack & Queue 4416. Graphs 4517. Thread 4618. Greedy 4719. Segment Tree 4820. Prefix Tree / Trie 4921. Cyclic sort 5022. Bit Manipulation 5125. Generic comments powered by Disqus