如何利用项目反应器设计/创建反应式方法

Question

我是一个新的基于项目反应器的反应式框架，比如SpringWebFlux，对此我有一些问题。

问题1：

例1：

  public Mono<String> doSome(String str){
    String key = str.substring(0, 5).toLowerCase() + str.substring(6);
    return getValueFromRedis(key);
  }

  public Mono<String> getValueFromRedis(String key){
    return reactiveRedisOperations().opsForHash().get("TestCache", key);
  }

例2：

  public Mono<String> doSome(String str){
    return Mono.fromCallable(() -> {
      String key = str.substring(0, 5).toLowerCase() + str.substring(6);
      return getValueFromRedis(key);
    }).flatMap(stringMono -> stringMono);
  }

  public Mono<String> getValueFromRedis(String key){
    return reactiveRedisOperations().opsForHash().get("TestCache", key);
  }

两个示例之间是否存在差异，或者两者都可以接受。

问题二：

例1：

  @PostMapping(value = "/greet")
  public Mono<String> greet(String name) {
    return Mono.fromCallable(() -> aMethod(name));
    // or return Mono.just(aMethod(name));
  }

  public String aMethod(String name){
    return "Hello: " + name;
  }

例2：

  @PostMapping(value = "/greet")
  public Mono<String> greet(String name) {
    return aMethod(name);
  }

  public Mono<String> aMethod(String name){
    return Mono.just("Hello: " + name);
  }

第二个问题很奇怪，我知道，但我想知道所有的方法都应该返回单声道或通量，或者我可以像问题2/示例1一样使用。

Answer 1

fromCallable每次Mono有新的订阅服务器时都执行lambda，而只是在实例化时捕获其参数并将其发送给其每个订阅服务器。

以下代码说明了区别：

private Instant getTime() {
    final Instant now = Instant.now();
    System.out.println("getTime(): " + now);
    return now;
}

@Test
public void just() throws InterruptedException {
    final Mono<Instant> mono = Mono.just(getTime())
            .doOnNext(instant -> System.out.println(instant));

    Thread.sleep(500);

    mono.subscribe();

    Thread.sleep(500);

    mono.subscribe();

    /* output is
        getTime(): 2019-08-14T22:47:06.823Z
        2019-08-14T22:47:06.823Z
        2019-08-14T22:47:06.823Z
    */
}

@Test
public void fromCallable() throws InterruptedException {
    final Mono<Instant> mono = Mono.fromCallable(() -> getTime())
            .doOnNext(instant -> System.out.println(instant));

    Thread.sleep(500);

    mono.subscribe();

    Thread.sleep(500);

    mono.subscribe();

    /* output is
        getTime(): 2019-08-14T22:47:13.947Z
        2019-08-14T22:47:13.947Z
        getTime(): 2019-08-14T22:47:14.447Z
        2019-08-14T22:47:14.447Z
    */
}

（另外，您还可以看看RxJava Single.just（）与Single。fromCallable（）？这是关于RxJava的，但是它的API几乎与Reactor相同）。

所以你的问题1的答案是：是的，有区别。在示例1中，调用doSome将立即导致从Redis获取异步值。在示例2中，只有在有人订阅了由doSome方法返回的Mono之后，才会涉及Redis。

Answer 2

问题1：是的，有区别。在示例1中，您将在Mono.fromCallable之外创建String key。这在这里没什么大不了的，但是如果是昂贵的手术，你会放慢一切。

此外，此逻辑reactiveRedisOperations（）。opsForHash（）在Mono之外执行。fromCallable。同样的事情-如果这是昂贵的，你正在放慢一切。

问题2：与问题1的观点相同<代码>单声道。只是接受常规对象，而不是稍后需要时调用的对象（如可调用的或供应商）。所以当使用Mono时。只是您正在立即为参数初始化付出代价。

在您的示例中几乎没有任何区别，但是Mono和Flux通常用于以异步方式链接昂贵的操作，因此不会阻止任何操作，比如数据库调用或对其他外部服务的调用。看看我下面的例子，其中sleep是对外部调用的模拟。注意打印语句的顺序。

public String superLongMethod() {
  System.out.println("superLongMethod");
  Thread.sleep(10000);
  return "ok";
}

System.out.println("before");
Mono.just(superLongMethod());
System.out.println("after");

// printed output is - before - superLongMethod - after

-----------------------------------------------------------------

System.out.println("before");
Mono.fromCallable(() -> superLongMethod());
System.out.println("after");

// printed output is - before - after - superLongMethod

-----------------------------------------------------------------

System.out.println("before");
String key = superLongMethod();
System.out.println("after");
return getValueFromRedis(key);

// printed output is - before - superLongMethod - after

-----------------------------------------------------------------

System.out.println("before");
Mono<String> mono = Mono.fromCallable(() -> {
  String key = superLongMethod();
  return getValueFromRedis(key);
}).flatMap(stringMono -> stringMono);
System.out.println("after");
return mono;

// printed output is - before - after - superLongMethod