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";
}
}
在这里呢,是同时调用两个接口,并把两个后果作整合。