WebFlux ve Event Loop — Bloklayan Tek Satır Neyi Durdurur?

1
@RestController
2
@RequestMapping("/accounts")
3
class AccountSummaryController {
4

5
    private final AccountRepository accounts;
6
    private final CustomerClient customers;
7
    private final LegacyCardAdapter cards;
8

9
    AccountSummaryController(AccountRepository accounts, CustomerClient customers, LegacyCardAdapter cards) {
10
        this.accounts = accounts;
11
        this.customers = customers;
12
        this.cards = cards;
13
    }
14

15
    // The app's home screen: balance, greeting and card limit. Opened by every
16
    // customer every morning, so thousands of these wait on I/O at the same time.
17
    @GetMapping("/{id}/summary")
18
    Mono<AccountSummary> summary(@PathVariable long id) {
19
        return accounts.findById(id)
20
                .switchIfEmpty(Mono.error(new AccountNotFoundException(id)))
21
                // The three lookups run concurrently; no thread waits for any of them.
22
                .flatMap(account -> Mono.zip(
23
                        Mono.just(account),
24
                        customers.fetch(account.customerId()),
25
                        cards.availableLimit(account.customerId())))
26
                .map(tuple -> AccountSummary.of(tuple.getT1(), tuple.getT2(), tuple.getT3()));
27
    }
28
}

1
@Component
2
class CustomerClient {
3

4
    private final WebClient http;
5

6
    CustomerClient(WebClient.Builder builder) {
7
        this.http = builder.baseUrl("http://customer-service").build();
8
    }
9

10
    Mono<Customer> fetch(long customerId) {
11
        return http.get().uri("/customers/{id}", customerId)
12
                .retrieve()
13
                .bodyToMono(Customer.class)
14
                .timeout(Duration.ofSeconds(2));   // a timeout is part of the chain, not a thread setting
15
    }
16
}

1
@Component
2
class LegacyCardAdapter {
3

4
    private final CardSystemSoapClient soap;   // the card platform only speaks SOAP: blocking
5

6
    LegacyCardAdapter(CardSystemSoapClient soap) {
7
        this.soap = soap;
8
    }
9

10
    Mono<CardLimit> availableLimit(long customerId) {
11
        // fromCallable: nothing runs until subscription (Mono.just would call SOAP right here).
12
        // subscribeOn(boundedElastic): the blocking call runs off the event loop.
13
        return Mono.fromCallable(() -> soap.availableLimit(customerId))
14
                .subscribeOn(Schedulers.boundedElastic())
15
                .timeout(Duration.ofSeconds(3))
16
                // The balance still shows if the card system is slow; the app says "limit unavailable".
17
                .onErrorReturn(CardLimit.UNAVAILABLE);
18
    }
19
}

1
interface AccountRepository extends ReactiveCrudRepository<AccountRow, Long> {
2

3
    // R2DBC: the driver itself is non-blocking. A JPA repository wrapped in Mono would not be.
4
    Flux<AccountRow> findByCustomerIdOrderByOpenedAtAsc(long customerId);
5
}
6

7
@Table("accounts")
8
record AccountRow(@Id Long id, long customerId, String iban, BigDecimal balance, Instant openedAt) {
9
}

1
spring:
2
  r2dbc:
3
    url: r2dbc:postgresql://db:5432/mobile
4
    username: mobile_api
5
    password: ${DB_PASSWORD}
6
    pool:
7
      max-size: 20
8
  reactor:
9
    context-propagation: auto   # MDC and tracing follow the chain across threads

1
@WebFluxTest(AccountSummaryController.class)
2
class AccountSummaryBlockingTest {
3

4
    @BeforeAll
5
    static void installBlockHound() {
6
        // Any blocking call on a non-blocking thread now throws.
7
        // On JDK 13+ the test JVM needs -XX:+AllowRedefinitionToAddDeleteMethods.
8
        BlockHound.install();
9
    }
10

11
    @Autowired WebTestClient client;
12
    @MockitoBean AccountRepository accounts;
13
    @MockitoBean CustomerClient customers;
14
    @MockitoBean LegacyCardAdapter cards;
15

16
    @Test
17
    void servesTheSummaryWithoutBlockingTheEventLoop() {
18
        given(accounts.findById(7L)).willReturn(Mono.just(
19
                new AccountRow(7L, 3L, "TR330006100519786457841326", new BigDecimal("1250.40"), Instant.EPOCH)));
20
        given(customers.fetch(3L)).willReturn(Mono.just(new Customer(3L, "Ada")));
21
        given(cards.availableLimit(3L)).willReturn(Mono.just(new CardLimit(new BigDecimal("8000.00"))));
22

23
        client.get().uri("/accounts/7/summary").exchange()
24
                .expectStatus().isOk()
25
                .expectBody()
26
                .jsonPath("$.customerName").isEqualTo("Ada")
27
                .jsonPath("$.balance").isEqualTo(1250.40);
28
    }
29
}