Skip to content

Commit 2df15af

Browse files
ylcn91ylcn91
committed
feat: add Scatter-Gather pattern (#3577)
1 parent 41625d8 commit 2df15af

17 files changed

Lines changed: 1341 additions & 0 deletions

‎pom.xml‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -260,6 +260,7 @@
260260
<module>rate-limiting-pattern</module>
261261
<module>fallback</module>
262262
<module>onion-architecture</module>
263+
<module>scatter-gather</module>
263264
</modules>
264265
<repositories>
265266
<repository>

‎scatter-gather/README.md‎

Lines changed: 237 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,237 @@
1+
---
2+
title: "Scatter-Gather Pattern in Java: Broadcasting One Request and Aggregating Many Replies"
3+
shortTitle: Scatter-Gather
4+
description: "Learn the Scatter-Gather pattern in Java. Send one request to several independent services at once, gather the replies that arrive in time, and aggregate them into a single answer while tolerating slow or failed recipients."
5+
category: Concurrency
6+
language: en
7+
tag:
8+
- Asynchronous
9+
- Decoupling
10+
- Integration
11+
- Messaging
12+
- Scalability
13+
---
14+
15+
## Also known as
16+
17+
* Broadcast and Aggregate
18+
* Request-Reply Broadcast
19+
20+
## Intent of Scatter-Gather Design Pattern
21+
22+
Send the same request to a number of independent recipients concurrently, collect the replies that arrive within a deadline, and combine them into one result. The caller pays roughly the latency of the slowest tolerated recipient instead of the sum of all latencies, and a single slow or failing recipient does not prevent an answer.
23+
24+
## Detailed Explanation of Scatter-Gather Pattern with Real-World Examples
25+
26+
Real-world example
27+
28+
> A travel site has to show the price of a hotel stay. It does not own a single price list; several rate providers each have their own. Asking them one after another would make the page as slow as all of them together, and one provider that is down would block the result. Instead the site scatters the same request to every provider at the same time, waits a few hundred milliseconds for their replies, drops the ones that did not answer in time, and shows the cheapest of the quotes it gathered.
29+
30+
In plain words
31+
32+
> Ask everybody the same question at once, wait a bounded time, and aggregate the answers you got.
33+
34+
Enterprise Integration Patterns says
35+
36+
> Use a Scatter-Gather that broadcasts a message to multiple recipients and re-aggregates the responses back into a single message.
37+
38+
How it differs from Fan-Out/Fan-In
39+
40+
> [Fan-Out/Fan-In](../fanout-fanin) splits one job into sub-tasks of the same kind and needs all of them back to build the result. Scatter-Gather sends the same message to different, independent services, expects heterogeneous replies, and is designed to produce a result even when some recipients are slow or unavailable.
41+
42+
Sequence diagram
43+
44+
```mermaid
45+
sequenceDiagram
46+
participant Client
47+
participant ScatterGather
48+
participant Atlas as Atlas Hotels
49+
participant Harbor as Harbor Stays
50+
participant Sleepy as Sleepy Suites (slow)
51+
participant Flaky as Flaky Inns (down)
52+
53+
Client->>ScatterGather: scatter(request)
54+
par broadcast
55+
ScatterGather->>Atlas: quote(request)
56+
ScatterGather->>Harbor: quote(request)
57+
ScatterGather->>Sleepy: quote(request)
58+
ScatterGather->>Flaky: quote(request)
59+
end
60+
Atlas-->>ScatterGather: 387.00
61+
Harbor-->>ScatterGather: 295.50
62+
Flaky-->>ScatterGather: error
63+
Note over ScatterGather,Sleepy: timeout expires, reply dropped
64+
ScatterGather->>ScatterGather: gather() keeps 2 of 4 replies
65+
ScatterGather->>Client: aggregate() returns Harbor Stays 295.50
66+
```
67+
68+
## Programmatic Example of Scatter-Gather Pattern in Java
69+
70+
The request is a plain immutable value. Every recipient receives exactly the same instance.
71+
72+
```java
73+
public record RateRequest(String city, LocalDate checkIn, int nights) {}
74+
75+
public record RateQuote(String provider, BigDecimal total) {}
76+
```
77+
78+
Each recipient implements `RateProvider`. In a real system these would be remote services with their own latency and failure profile. The demo ships a fast in-memory provider, a decorator that delays any provider, and a provider that is down.
79+
80+
```java
81+
public interface RateProvider {
82+
String name();
83+
84+
RateQuote quote(RateRequest request);
85+
}
86+
```
87+
88+
The `Aggregator` reduces the gathered replies. Because it is a separate strategy the same coordinator can serve callers that want the cheapest quote, an average, or the full list.
89+
90+
```java
91+
@FunctionalInterface
92+
public interface Aggregator<T, R> {
93+
R aggregate(List<T> replies);
94+
95+
static Aggregator<RateQuote, Optional<RateQuote>> cheapestQuote() {
96+
return replies -> replies.stream().min(Comparator.comparing(RateQuote::total));
97+
}
98+
}
99+
```
100+
101+
`ScatterGather` implements the three phases. The scatter phase submits one asynchronous call per provider and attaches the timeout to each future. Nothing is awaited yet.
102+
103+
```java
104+
public List<PendingReply> scatter(RateRequest request, List<RateProvider> providers) {
105+
var pending = new ArrayList<PendingReply>();
106+
for (var provider : providers) {
107+
var reply =
108+
CompletableFuture.supplyAsync(() -> provider.quote(request), executor)
109+
.orTimeout(timeout.toMillis(), TimeUnit.MILLISECONDS);
110+
pending.add(new PendingReply(provider, reply));
111+
}
112+
return pending;
113+
}
114+
```
115+
116+
The gather phase waits until every future has settled, then keeps the successful replies. Timeouts and failures are logged and dropped, which is what allows the caller to get an answer from the providers that did respond.
117+
118+
```java
119+
public List<RateQuote> gather(List<PendingReply> pending) {
120+
CompletableFuture.allOf(pending.stream().map(PendingReply::reply).toArray(CompletableFuture[]::new))
121+
.exceptionally(ex -> null)
122+
.join();
123+
var quotes = new ArrayList<RateQuote>();
124+
for (var entry : pending) {
125+
try {
126+
quotes.add(entry.reply().join());
127+
} catch (CompletionException e) {
128+
if (e.getCause() instanceof TimeoutException) {
129+
LOGGER.warn("Dropping {}: no reply within {} ms", entry.provider().name(), timeout.toMillis());
130+
} else {
131+
LOGGER.warn("Dropping {}: {}", entry.provider().name(), e.getCause().getMessage());
132+
}
133+
}
134+
}
135+
return quotes;
136+
}
137+
```
138+
139+
A convenience method chains the phases together.
140+
141+
```java
142+
public <R> R scatterGather(
143+
RateRequest request, List<RateProvider> providers, Aggregator<RateQuote, R> aggregator) {
144+
return aggregator.aggregate(gather(scatter(request, providers)));
145+
}
146+
```
147+
148+
The demo application asks four providers for the same three-night stay. One provider needs two seconds while the gather timeout is 300 milliseconds, and one provider is down. The site still answers with the cheapest of the two quotes it gathered.
149+
150+
```java
151+
var request = new RateRequest("Lisbon", LocalDate.of(2026, 10, 3), 3);
152+
var providers =
153+
List.<RateProvider>of(
154+
new InMemoryRateProvider("Atlas Hotels", new BigDecimal("129.00")),
155+
new InMemoryRateProvider("Harbor Stays", new BigDecimal("98.50")),
156+
new DelayedRateProvider(
157+
new InMemoryRateProvider("Sleepy Suites", new BigDecimal("75.00")),
158+
Duration.ofSeconds(2)),
159+
new FailingRateProvider("Flaky Inns"));
160+
161+
try (var scatterGather =
162+
new ScatterGather(Executors.newFixedThreadPool(providers.size()), Duration.ofMillis(300))) {
163+
var pending = scatterGather.scatter(request, providers);
164+
var quotes = scatterGather.gather(pending);
165+
Aggregator.cheapestQuote()
166+
.aggregate(quotes)
167+
.ifPresentOrElse(
168+
best -> LOGGER.info("Best offer: {} at {}", best.provider(), best.total()),
169+
() -> LOGGER.info("No provider answered in time"));
170+
}
171+
```
172+
173+
Running the program produces output similar to the following.
174+
175+
```
176+
Scatter phase: broadcasting the same request to every provider
177+
Scattering request for 3 nights in Lisbon to 4 providers
178+
Atlas Hotels quotes 387.00 for 3 nights in Lisbon
179+
Harbor Stays quotes 295.50 for 3 nights in Lisbon
180+
Gather phase: collecting replies that arrive within the timeout
181+
Sleepy Suites is slow and will need 2000 ms to answer
182+
Gathered quote 387.00 from Atlas Hotels
183+
Gathered quote 295.50 from Harbor Stays
184+
Dropping Sleepy Suites: no reply within 300 ms
185+
Dropping Flaky Inns: Flaky Inns is unavailable
186+
Gathered 2 of 4 replies
187+
Aggregate phase: choosing the cheapest of 2 quotes
188+
Best offer: Harbor Stays at 295.50
189+
```
190+
191+
## Class diagram
192+
193+
See [scatter-gather.urm.puml](./etc/scatter-gather.urm.puml) for the PlantUML class diagram.
194+
195+
## When to Use the Scatter-Gather Pattern in Java
196+
197+
* The same question has to be answered by several independent services, such as price comparison, search federation, or quorum reads.
198+
* The latency of asking recipients sequentially is unacceptable.
199+
* A partial answer built from the recipients that replied in time is more valuable than no answer.
200+
* The recipients are unknown or change at runtime, so the caller should only depend on a common contract.
201+
202+
## Real-World Applications of Scatter-Gather Pattern in Java
203+
204+
* Travel and shopping comparison sites that query many suppliers for the same product.
205+
* Distributed search engines that broadcast a query to every index shard and merge the ranked results.
206+
* Quorum reads in replicated data stores that ask several replicas and accept the first consistent majority.
207+
* [Apache Camel Scatter-Gather EIP](https://camel.apache.org/components/latest/eips/scatter-gather.html)
208+
* [Spring Integration Scatter-Gather](https://docs.spring.io/spring-integration/reference/scatter-gather.html)
209+
* [Akka scatter-gather with `ask` and `Future.sequence`](https://doc.akka.io/docs/akka/current/futures.html)
210+
211+
## Benefits and Trade-offs of Scatter-Gather Pattern
212+
213+
Benefits:
214+
215+
* **Lower latency**: recipients are called concurrently, so the caller waits for the slowest tolerated reply rather than the sum of all replies.
216+
* **Resilience**: a timeout bounds the wait and a failed recipient is simply left out of the aggregate.
217+
* **Decoupling**: the caller depends only on the recipient contract and the aggregation strategy, not on the number or identity of recipients.
218+
219+
Trade-offs:
220+
221+
* **Partial results**: the caller must be able to live with an answer built from a subset of recipients, and the aggregator must handle an empty set.
222+
* **Resource usage**: every request occupies one thread or connection per recipient; a dropped reply may still be computed by the recipient.
223+
* **Tuning**: the timeout is a compromise between completeness and responsiveness and usually needs measurement to get right.
224+
225+
## Related Java Design Patterns
226+
227+
* [Fan-Out/Fan-In](../fanout-fanin): splits one task into homogeneous sub-tasks and waits for all of them; Scatter-Gather broadcasts one request to heterogeneous recipients and tolerates missing replies.
228+
* [Microservices Aggregator](../microservices-aggregrator): a service that composes the responses of several downstream services; Scatter-Gather is a way to fetch those responses concurrently.
229+
* [Async Method Invocation](../async-method-invocation): the mechanism used to call each recipient without blocking the caller.
230+
* [Promise](../promise): each pending reply is a promise that either completes with a quote or fails.
231+
* Timeout: bounds how long the gather phase waits for each recipient.
232+
233+
## References and Credits
234+
235+
* [Enterprise Integration Patterns](https://www.amazon.com/gp/product/0321200683) (Gregor Hohpe and Bobby Woolf)
236+
* [Scatter-Gather at enterpriseintegrationpatterns.com](https://www.enterpriseintegrationpatterns.com/patterns/messaging/BroadcastAggregate.html)
237+
* [Java Concurrency in Practice](https://www.amazon.com/gp/product/0321349601) (Brian Goetz)

‎scatter-gather/etc/scatter-gather.urm.puml‎

Lines changed: 73 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,73 @@
1+
@startuml
2+
package com.iluwatar.scattergather {
3+
class RateRequest {
4+
+ RateRequest(city : String, checkIn : LocalDate, nights : int)
5+
+ city() : String
6+
+ checkIn() : LocalDate
7+
+ nights() : int
8+
}
9+
class RateQuote {
10+
+ RateQuote(provider : String, total : BigDecimal)
11+
+ provider() : String
12+
+ total() : BigDecimal
13+
}
14+
interface RateProvider {
15+
+ name() : String {abstract}
16+
+ quote(request : RateRequest) : RateQuote {abstract}
17+
}
18+
class InMemoryRateProvider {
19+
- name : String
20+
- nightlyRate : BigDecimal
21+
+ InMemoryRateProvider(name : String, nightlyRate : BigDecimal)
22+
+ name() : String
23+
+ quote(request : RateRequest) : RateQuote
24+
}
25+
class DelayedRateProvider {
26+
- delegate : RateProvider
27+
- delay : Duration
28+
+ DelayedRateProvider(delegate : RateProvider, delay : Duration)
29+
+ name() : String
30+
+ quote(request : RateRequest) : RateQuote
31+
}
32+
class FailingRateProvider {
33+
- name : String
34+
+ FailingRateProvider(name : String)
35+
+ name() : String
36+
+ quote(request : RateRequest) : RateQuote
37+
}
38+
interface Aggregator<T, R> {
39+
+ aggregate(replies : List<T>) : R {abstract}
40+
+ cheapestQuote() : Aggregator<RateQuote, Optional<RateQuote>> {static}
41+
}
42+
class PendingReply {
43+
+ PendingReply(provider : RateProvider, reply : CompletableFuture<RateQuote>)
44+
+ provider() : RateProvider
45+
+ reply() : CompletableFuture<RateQuote>
46+
}
47+
class ScatterGather {
48+
- executor : ExecutorService
49+
- timeout : Duration
50+
+ ScatterGather(executor : ExecutorService, timeout : Duration)
51+
+ scatter(request : RateRequest, providers : List<RateProvider>) : List<PendingReply>
52+
+ gather(pending : List<PendingReply>) : List<RateQuote>
53+
+ scatterGather(request : RateRequest, providers : List<RateProvider>, aggregator : Aggregator<RateQuote, R>) : R
54+
+ close() : void
55+
}
56+
class App {
57+
+ App()
58+
+ main(args : String[]) : void
59+
}
60+
}
61+
InMemoryRateProvider ..|> RateProvider
62+
DelayedRateProvider ..|> RateProvider
63+
DelayedRateProvider --> RateProvider
64+
FailingRateProvider ..|> RateProvider
65+
RateProvider ..> RateRequest
66+
RateProvider ..> RateQuote
67+
PendingReply --> RateProvider
68+
PendingReply ..> RateQuote
69+
ScatterGather ..> PendingReply
70+
ScatterGather ..> Aggregator
71+
ScatterGather --> "*" RateProvider
72+
App ..> ScatterGather
73+
@enduml

‎scatter-gather/pom.xml‎

Lines changed: 70 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,70 @@
1+
<?xml version="1.0" encoding="UTF-8"?>
2+
<!--
3+
4+
This project is licensed under the MIT license. Module model-view-viewmodel is using ZK framework licensed under LGPL (see lgpl-3.0.txt).
5+
6+
The MIT License
7+
Copyright © 2014-2022 Ilkka Seppälä
8+
9+
Permission is hereby granted, free of charge, to any person obtaining a copy
10+
of this software and associated documentation files (the "Software"), to deal
11+
in the Software without restriction, including without limitation the rights
12+
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
13+
copies of the Software, and to permit persons to whom the Software is
14+
furnished to do so, subject to the following conditions:
15+
16+
The above copyright notice and this permission notice shall be included in
17+
all copies or substantial portions of the Software.
18+
19+
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
20+
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
21+
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
22+
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
23+
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
24+
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
25+
THE SOFTWARE.
26+
27+
-->
28+
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
29+
<modelVersion>4.0.0</modelVersion>
30+
<parent>
31+
<groupId>com.iluwatar</groupId>
32+
<artifactId>java-design-patterns</artifactId>
33+
<version>1.26.0-SNAPSHOT</version>
34+
</parent>
35+
<artifactId>scatter-gather</artifactId>
36+
<dependencies>
37+
<dependency>
38+
<groupId>org.slf4j</groupId>
39+
<artifactId>slf4j-api</artifactId>
40+
</dependency>
41+
<dependency>
42+
<groupId>ch.qos.logback</groupId>
43+
<artifactId>logback-classic</artifactId>
44+
</dependency>
45+
<dependency>
46+
<groupId>org.junit.jupiter</groupId>
47+
<artifactId>junit-jupiter-engine</artifactId>
48+
<scope>test</scope>
49+
</dependency>
50+
</dependencies>
51+
<build>
52+
<plugins>
53+
<plugin>
54+
<groupId>org.apache.maven.plugins</groupId>
55+
<artifactId>maven-assembly-plugin</artifactId>
56+
<executions>
57+
<execution>
58+
<configuration>
59+
<archive>
60+
<manifest>
61+
<mainClass>com.iluwatar.scattergather.App</mainClass>
62+
</manifest>
63+
</archive>
64+
</configuration>
65+
</execution>
66+
</executions>
67+
</plugin>
68+
</plugins>
69+
</build>
70+
</project>

0 commit comments

Comments
 (0)