【Reactor第八篇】WebFlux 服务编排

待我称王封你为后i 2023-10-09 19:59 3阅读 0赞

WebFlux 服务编排是指使用 WebFlux 框架来编排多个**异步服务**的执行顺序和数据流动，从而构建出一个完整的、基于事件驱动的响应式应用程序。

WebFlux服务编排的优势如下：

1.  高性能：WebFlux基于响应式编程模型，可以使用少量的线程处理大量的请求，从而提高系统的并发能力和吞吐量。
2.  异步处理：WebFlux可以异步处理请求和响应，避免线程的阻塞和等待，提高系统的并发能力和性能。
3.  高可靠性：WebFlux基于事件驱动的编程模型，可以更好地处理错误和异常，从而提高系统的可靠性和稳定性。
4.  简洁清晰：WebFlux的代码简洁清晰，可以使用函数式编程风格来编写业务逻辑，提高代码的可读性和可维护性。
5.  可扩展性：WebFlux可以轻松地集成其他的响应式组件和服务，例如Reactive Streams、Spring Cloud、RSocket等，从而提高系统的可扩展性和灵活性。

综上所述，WebFlux服务编排可以帮助我们构建高性能、高可靠性、可扩展性强的响应式应用程序，提高系统的并发能力和性能，从而更好地满足现代应用程序的需求。

一个示例

public Mono> getOrderDetails(String orderId) {
        return Mono.fromCallable(() -> {
            // 查询订单基本信息
            return "order info";
        })
        .flatMap(orderInfo -> {
            // 查询订单商品信息
            return Mono.fromCallable(() -> {
                return "order item info";
            });
        })
        .flatMap(orderItemInfo -> {
            // 查询订单配送信息
            return Mono.fromCallable(() -> {
                return "order delivery info";
            });
        })
        .flatMap(orderDeliveryInfo -> {
            // 查询订单支付信息
            return Mono.fromCallable(() -> {
                return "order payment info";
            });
        });
    }
    复制代码

为什么使用 fromCallable，就是上面说的，WebFlux 编排的是异步服务，而不是同步服务。

但是实际线上不要使用 fromCallable，会导致创建很多个线程，高并发场景下会导致资源竞争激烈，从而服务性能急剧下降。

### 1 串行 ###

#### 1.1 不需要 invoker1 的结果 ####

long start = System.currentTimeMillis();
    
    Mono<String> invoke1 = Invoker1.invoke1();
    Mono<String> result = invoke1.flatMap(p -> Invoker2.invoke2())
          .map(s -> {
             return s.toString();
          });
    // result: invoker2, 耗时：3592（串行）
    System.out.println("result: " + result.block() + ", 耗时：" + (System.currentTimeMillis() - start));
    复制代码

#### 1.2 需要返回 invoker1 的结果 ####

long start = System.currentTimeMillis();
    
    Mono<String> invoke1 = Invoker1.invoke1();
    Mono<String> result = invoke1.flatMap(p -> {
       return Invoker2.invoke2().map(s -> {
          return p + s;
       });
    });
    
    // result: invoker1invoker2, 耗时：3554（串行）
    System.out.println("result: " + result.block() + ", 耗时：" + (System.currentTimeMillis() - start));
    复制代码

### 2 并行 ###

#### 2.1 zip 方法 ####

zip() 方法可以一次组装任意个Mono，适用于有多个Mono的情况

long start = System.currentTimeMillis();
    Mono<String> invoke1 = Invoker1.invoke1();
    Mono<String> invoker2 = Invoker2.invoke2();
    Mono<String> result = Mono.zip(invoke1, invoker2)
          .map(s-> {
             String t1 = s.getT1();
             String t2 = s.getT2();
             return String.format("invoke1:%s, invoke2: %s", t1, t2);
          });
    // invoker1invoker2耗时：2650 （并行）
    System.out.println("result: " + result.block() + "，耗时：" + (System.currentTimeMillis() - start));
    复制代码

#### 2.2 zipWith 方法 ####

zipWith() 每次组装一个Mono对象，使用于组装Mono个数比较少的情况。

long start = System.currentTimeMillis();
    Mono<String> invoke1 = Invoker1.invoke1();
    Mono<String> invoker2 = Invoker2.invoke2();
    Mono<String> result = invoke1.zipWith(invoker2)
          .map(s -> {
             return String.format("invoke1:%s, invoke2: %s", s.getT1(), s.getT2());
          });
    // invoker1invoker2耗时：2469 （并行）
    System.out.println(result.block() + "，耗时：" + (System.currentTimeMillis() - start));
    复制代码

### 3 前提 ###

这里的 invoker 就是第三方系统调用。

保证 invoker 是在独立的线程中执行，这样 invoker 不会影响业务处理。

public class Invoker1 {
    
       public static Mono<String> invoke1() {
          return Mono.
                fromSupplier(() -> {
                   try {
                      Thread.sleep(1000);
                   } catch (InterruptedException e) {
                      throw new RuntimeException(e);
                   }
                   return "invoker1";
                })
                .subscribeOn(Schedulers.parallel())
                .doOnError(e -> {
                   System.out.println("error invoker1");
                });
       }
    }
    复制代码

public class Invoker2 {
    
       public static Mono<String> invoke2() {
          return Mono.fromSupplier(() -> {
                   try {
                      Thread.sleep(2000);
                   } catch (InterruptedException e) {
                      throw new RuntimeException(e);
                   }
                   return "invoker2";
                })
                .subscribeOn(Schedulers.parallel())
                .doOnError(e -> {
                   System.out.println("error invoker2");
                });
       }
    }

作者：六七十三  
链接：https://juejin.cn/post/7225902886944751677