乐趣区

关于后端:Reactor第八篇WebFlux-服务编排

jiezi

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");
            });
   }
}
退出移动版