关于java:reactor编程如何处理并发远程调用

1、搭建一个简略的 webflux 工程

pom 依赖很简略，就一个，让它去传递依赖去，底层依赖的是 4.1.72 版本的 netty，目前最新是 4.1.73 了：

<dependencies>
    <dependency>
      <groupId>org.springframework.boot</groupId>
      <artifactId>spring-boot-starter-webflux</artifactId>
      <version>2.6.2</version>
    </dependency>
</dependencies>

默认会有很多的 debug 日志，影响调试察看的话，能够设置日志 logback.xml：

<?xml version="1.0" encoding="UTF-8"?>
<configuration scan="true" scanPeriod="60 seconds" debug="false">
    <appender name="stdout" class="ch.qos.logback.core.ConsoleAppender">
        <Target>System.out</Target>
        <encoder>
            <pattern>[%d] [%thread] %-5level %logger{36} - %msg%n</pattern>
        </encoder>
    </appender>
    <root level="INFO">
        <appender-ref ref="stdout" />
    </root>
</configuration>

启动类，外面能够依据配置来指定 io worker 数量，默认会应用 runtime 去取核数

@SpringBootApplication
public class WebfluxApplication {public static void main(String[] args) {System.setProperty("reactor.netty.ioWorkerCount", "4");
        SpringApplication.run(WebfluxApplication.class, args);
    }
}

2、编写并行调用的 controller

    @GetMapping("/zip")
    public Mono<String> zip() {Mono<String> mono1 = client.get().uri("abc1").retrieve().bodyToMono(String.class);
        Mono<String> mono2 = client.get().uri("abc2").retrieve().bodyToMono(String.class);
        return Mono.zip(mono1, mono2, User::combine);
    }

public class User {static ObjectMapper objectMapper = new ObjectMapper();

    public static String combine(String s1, String s2) {
        try {Map<String, String> map1 = objectMapper.readValue(s1, new TypeReference<Map<String, String>>() {});
            Map<String, String> map2 = objectMapper.readValue(s2, Map.class);
            map1.putAll(map2);
            return objectMapper.writeValueAsString(map1);
        } catch (JsonMappingException e) {e.printStackTrace();
        } catch (JsonProcessingException e) {e.printStackTrace();
        }
        return "test";
    }
    
}

在这里呢，是同时调用两个接口，并把两个后果作整合。