Virtual Threads vs Platform Threads

1
// Every outgoing international payment is checked against sanctions lists before
2
// it is released. The check is I/O: the payer from our database, the match from
3
// the screening provider's HTTP API.
4
public class PaymentScreening {
5

6
    private final ScreeningClient screening;
7
    private final CustomerRepository customers;
8

9
    public PaymentScreening(ScreeningClient screening, CustomerRepository customers) {
10
        this.screening = screening;
11
        this.customers = customers;
12
    }
13

14
    // Plain blocking code. On a virtual thread, each blocking call unmounts the
15
    // thread and frees its carrier for another task.
16
    public List<ScreeningResult> screen(List<OutgoingPayment> payments) throws Exception {
17
        try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
18
            List<Future<ScreeningResult>> futures = payments.stream()
19
                    .map(payment -> executor.submit(() -> new ScreeningResult(
20
                            payment.id(),
21
                            screening.check(customers.nameOf(payment.customerId()),   // blocking JDBC
22
                                            payment.beneficiaryName(),                // blocking HTTP
23
                                            payment.beneficiaryCountry()))))
24
                    .toList();
25

26
            List<ScreeningResult> result = new ArrayList<>();
27
            for (Future<ScreeningResult> future : futures) result.add(future.get());
28
            return result;   // HIT results go to the compliance queue, CLEAR ones are released
29
        }
30
    }
31
}

1
public class CustomerRepository {
2

3
    // 10,000 virtual threads would happily open 10,000 connections.
4
    // The pool has 20; the semaphore makes the rest wait cheaply instead of timing out.
5
    private final Semaphore dbPermits = new Semaphore(20);
6
    private final DataSource dataSource;
7

8
    public CustomerRepository(DataSource dataSource) {
9
        this.dataSource = dataSource;
10
    }
11

12
    public String nameOf(long customerId) throws SQLException, InterruptedException {
13
        dbPermits.acquire();
14
        try (var connection = dataSource.getConnection();
15
             var statement = connection.prepareStatement("SELECT name FROM customers WHERE id = ?")) {
16
            statement.setLong(1, customerId);
17
            try (var rs = statement.executeQuery()) {
18
                return rs.next() ? rs.getString(1) : "?";
19
            }
20
        } finally {
21
            dbPermits.release();
22
        }
23
    }
24
}

1
public class TokenCache {
2

3
    private final ReentrantLock lock = new ReentrantLock();
4
    private final AuthClient auth;
5
    private volatile Token token;
6

7
    public TokenCache(AuthClient auth) {
8
        this.auth = auth;
9
    }
10

11
    // The screening provider wants an OAuth token that expires every few minutes.
12
    // Refreshing blocks on a network call while holding the lock.
13
    // With `synchronized` (before JDK 24) that pins the carrier thread;
14
    // ReentrantLock lets the virtual thread unmount while it waits.
15
    public Token current() {
16
        Token t = token;
17
        if (t != null && !t.expired()) return t;
18
        lock.lock();
19
        try {
20
            if (token == null || token.expired()) token = auth.fetchToken();
21
            return token;
22
        } finally {
23
            lock.unlock();
24
        }
25
    }
26
}

1
spring:
2
  threads:
3
    virtual:
4
      enabled: true      # Tomcat request handling, @Async and scheduling run on virtual threads
5
  datasource:
6
    hikari:
7
      maximum-pool-size: 20   # still the real limit on concurrent queries

1
public class Main {
2

3
    public static void main(String[] args) throws Exception {
4
        var screening = new PaymentScreening(new FakeScreeningClient(Duration.ofMillis(100)),
5
                new CustomerRepository(Databases.local()));
6

7
        List<OutgoingPayment> payments = IntStream.range(0, 10_000)
8
                .mapToObj(i -> new OutgoingPayment(i, i % 500, "Beneficiary " + i, "DE"))
9
                .toList();
10

11
        long start = System.nanoTime();
12
        List<ScreeningResult> result = screening.screen(payments);
13
        System.out.printf("%d payments screened in %d ms%n", result.size(), (System.nanoTime() - start) / 1_000_000);
14
        // The time depends on the machine and the database; measure, don't assume.
15
    }
16
}