【Reactor第八篇】WebFlux 服务编排

2023-05-01 10:00:27 浏览数 (3)

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

WebFlux服务编排的优势如下:

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

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

一个示例

代码语言:javascript复制
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 的结果

代码语言:javascript复制
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 的结果

代码语言:javascript复制
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的情况

代码语言:javascript复制
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个数比较少的情况。

代码语言:javascript复制
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 不会影响业务处理。

代码语言:javascript复制
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");
            });
   }
}
代码语言:javascript复制
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");
            });
   }
}

0 人点赞