知識庫 / Spring WebFlux RSS 訂閱

讓我們嘗試識別我們使用案例所需的轉換：

• 我們從一個原始 productId 開始
• 我們將其轉換為 Product
• 我們使用 Product 來計算 Price
• 我們使用 Price 來創建 event
• 最後，我們將 event 發佈到消息隊列

現在，讓我們重構代碼以反映此函數鏈：

void addProductToCart(String productId, String cartId) {
    Mono<Product> productMono = repository.findById(productId)
      .switchIfEmpty(Mono.error(() -> new IllegalArgumentException("not found!")));

    Mono<Price> priceMono = productMono.flatMap(product -> {
        if (product.category().isEligibleForDiscount()) {
            return discountService.discountForProduct(productId)
              .map(product.basePrice()::applyDiscount);
        }
        return Mono.just(product.basePrice());
    });

    Mono<ProductAddedToCartEvent> eventMono = priceMono.map(
      price -> new ProductAddedToCartEvent(productId, price.value(), price.currency(), cartId));

    eventMono.subscribe(event -> kafkaTemplate.send(PRODUCT_ADDED_TO_CART_TOPIC, cartId, event));
}

現在，我們內聯本地變量以保持代碼緊湊。此外，我們提取一個計算價格的函數，並在 <em>flatMap()</em> 中使用它。

void addProductToCart(String productId, String cartId) {
    repository.findById(productId)
      .switchIfEmpty(Mono.error(() -> new IllegalArgumentException("not found!")))
      .flatMap(this::computePrice)
      .map(price -> new ProductAddedToCartEvent(productId, price.value(), price.currency(), cartId))
      .subscribe(event -> kafkaTemplate.send(PRODUCT_ADDED_TO_CART_TOPIC, cartId, event));
}

Mono<Price> computePrice(Product product) {
    if (product.category().isEligibleForDiscount()) {
        return discountService.discountForProduct(product.id())
          .map(product.basePrice()::applyDiscount);
    }
    return Mono.just(product.basePrice());
}

4. 虛擬線程

虛擬線程是通過 Java Project Loom 在 Java 中引入的一種替代方案，用於並行處理。 它們是輕量級的，在用户模式下由 Java 虛擬機 (JVM) 管理的線程。 由於如此，它們特別適合用於 I/O 操作，其中傳統線程可能會花費大量時間等待外部資源。

與異步或響應式解決方案相比，虛擬線程使我們能夠繼續使用線程池單請求處理模型。 換句話説，我們可以繼續編寫順序代碼，而無需混合業務邏輯和響應式 API。

4.1. 虛擬線程

有幾種方法可以利用虛擬線程來執行我們的代碼。對於單個方法，例如前一個示例中演示的方法，我們可以使用 <em >startVirtualThread()</em>。這個靜態方法最近被添加到 <em >Thread</em> API 中，它會在一個新的虛擬線程上執行一個 <em >Runnable</em>：

public void addProductToCart(String productId, String cartId) {
    Thread.startVirtualThread(() -> computePriceAndPublishMessage(productId, cartId));
}

private void computePriceAndPublishMessage(String productId, String cartId) {
    // ...
}

或者，我們還可以創建一個ExecutorService，該ExecutorService依賴於新的靜態工廠方法Executors.newVirtualThreadPerTaskExecutor()。Furthermore, for applications using Spring Framework 6 and Spring Boot 3, we can leverage the new Executor and configure Spring to favor virtual threads over platform threads.

4.2. 兼容性

虛擬線程通過使用更傳統的同步編程模型，簡化了代碼。因此，我們可以像進行阻塞 I/O 操作一樣，以順序的方式編寫代碼，而無需擔心顯式的反應式構造。

此外，我們可以無縫地從常規單線程代碼切換到虛擬線程，只需少量或無需修改。 例如，在我們之前的示例中，我們只需要使用靜態工廠方法 startVirtualThread() 創建虛擬線程並執行其中的邏輯：

void addProductToCart(String productId, String cartId) {
    Thread.startVirtualThread(() -> computePriceAndPublishMessage(productId, cartId));
}

void computePriceAndPublishMessage(String productId, String cartId) {
    Product product = repository.findById(productId)
      .orElseThrow(() -> new IllegalArgumentException("not found!"));

    Price price = computePrice(productId, product);

    var event = new ProductAddedToCartEvent(productId, price.value(), price.currency(), cartId);
    kafkaTemplate.send(PRODUCT_ADDED_TO_CART_TOPIC, cartId, event);
}

Price computePrice(String productId, Product product) {
    if (product.category().isEligibleForDiscount()) {
        BigDecimal discount = discountService.discountForProduct(productId);
        return product.basePrice().applyDiscount(discount);
    }
    return product.basePrice();
}

4.3. 可讀性

採用線程池單請求處理模型，可以更容易地理解和推理業務邏輯。這有助於降低與響應式編程範式相關的認知負擔。

換句話説，虛擬線程允許我們清晰地將技術關注點與業務邏輯分離。從而消除了在實現業務用例時需要外部 API 的需求。

5. 結論

在本文中，我們比較了兩種不同的併發和異步處理方法。我們首先分析了項目 Reactor 的 WebFlux 以及反應式編程範式。我們發現這種方法傾向於使用不可變對象和函數式流水線。

隨後，我們討論了虛擬線程及其與遺留代碼庫的卓越兼容性，這使得平滑過渡到非阻塞代碼成為可能。此外，它們還具有將業務邏輯與基礎設施代碼和其他技術問題明確分離的額外優勢。