CompletableFuture ve Structured Concurrency

1
// The bank's FX desk buys currency from several liquidity providers (other banks)
2
// and gives the customer the best ask it can get.
3
public interface LiquidityProvider {
4

5
    String name();
6

7
    /** Blocking call to the provider's pricing API. */
8
    Quote quote(String pair, BigDecimal amount);
9
}
10

11
record Quote(String provider, String pair, BigDecimal ask) {
12
}

1
public class BestRateWithFutures {
2

3
    private final List<LiquidityProvider> providers;
4
    // Never the common ForkJoinPool for blocking I/O: its few threads would all wait.
5
    private final ExecutorService io = Executors.newVirtualThreadPerTaskExecutor();
6

7
    public BestRateWithFutures(List<LiquidityProvider> providers) {
8
        this.providers = providers;
9
    }
10

11
    public Optional<Quote> best(String pair, BigDecimal amount) {
12
        List<CompletableFuture<Optional<Quote>>> calls = providers.stream()
13
                .map(provider -> CompletableFuture
14
                        .supplyAsync(() -> Optional.of(provider.quote(pair, amount)), io)
15
                        .orTimeout(800, TimeUnit.MILLISECONDS)         // a quote older than this is useless
16
                        .exceptionally(ex -> Optional.empty()))          // a failed provider is just "no quote"
17
                .toList();
18

19
        CompletableFuture.allOf(calls.toArray(CompletableFuture[]::new)).join();
20

21
        return calls.stream()
22
                .map(CompletableFuture::join)                           // already complete: no waiting
23
                .flatMap(Optional::stream)
24
                .min(Comparator.comparing(Quote::ask));
25
    }
26
}

1
// Preview API in Java 21–24 (compile and run with --enable-preview).
2
public class BestRateStructured {
3

4
    private final List<LiquidityProvider> providers;
5

6
    public BestRateStructured(List<LiquidityProvider> providers) {
7
        this.providers = providers;
8
    }
9

10
    // Here every provider is required (say, a large trade that must be split across
11
    // all of them): if one fails, the others are cancelled and the whole call fails.
12
    // The subtasks can never outlive this method.
13
    public Quote best(String pair, BigDecimal amount) throws InterruptedException, ExecutionException {
14
        try (var scope = new StructuredTaskScope.ShutdownOnFailure()) {
15
            List<StructuredTaskScope.Subtask<Quote>> tasks = providers.stream()
16
                    .map(provider -> scope.fork(() -> provider.quote(pair, amount)))
17
                    .toList();
18

19
            scope.joinUntil(Instant.now().plusMillis(800));   // one deadline for all of them
20
            scope.throwIfFailed();
21

22
            return tasks.stream()
23
                    .map(StructuredTaskScope.Subtask::get)
24
                    .min(Comparator.comparing(Quote::ask))
25
                    .orElseThrow();
26
        } catch (TimeoutException ex) {
27
            throw new ExecutionException("providers too slow", ex);
28
        }
29
    }
30
}

1
public class Main {
2

3
    public static void main(String[] args) throws Exception {
4
        List<LiquidityProvider> providers = List.of(
5
                new FakeProvider("banka-a", "38.4150", Duration.ofMillis(50)),
6
                new FakeProvider("banka-b", "38.3900", Duration.ofSeconds(3)),   // misses the 800 ms budget
7
                new FakeProvider("banka-c", "38.4020", Duration.ofMillis(200)));
8
        BigDecimal amount = new BigDecimal("250000");                        // EUR
9

10
        System.out.println(new BestRateWithFutures(providers).best("EURTRY", amount));
11
        // Optional[Quote[provider=banka-c, pair=EURTRY, ask=38.4020]] — the slow, cheaper one timed out
12

13
        try {
14
            new BestRateStructured(providers).best("EURTRY", amount);
15
        } catch (ExecutionException ex) {
16
            System.out.println("structured: " + ex.getMessage());
17
            // structured: providers too slow — and the slow call was cancelled, not left running
18
        }
19
    }
20
}