如何限制spring-webflux WebClient中打开套接字的数量？

Question

我有一些RESTful服务，我想用Reactor和Spring WebClient准备简单的性能基准测试。Benchmark只是创建N个用户，然后为每个创建的用户发布M个投票。

不幸的是，下面的代码超过了我的linux机器上的最大打开文件限制，即1024(ulimit-n 1024)。

    RestService restService = ...
    int N_ITERATIONS = 100;
    int M_VOTES = 100;

    Flux.range(0, N_ITERATIONS)
            .parallel()
            .runOn(Schedulers.parallel())
            .flatMap(iteration -> restService.postUserRegistration(User.builder().build()))
            .flatMap(user -> Flux.range(0, M_VOTES)
                    .flatMap(vote -> restService.postUserVote(Vote.builder().build()))
                    .collectList()
                    .map(votes -> Tuple.of(user, votes))
            ).doOnNext(userVotes -> log.info("User: {} voted: {}", userVotes._1(), userVotes._2()))
            .sequential()
            .toIterable();

RestService是用Spring WebFlux中的标准WebClient实现的。

Caused by: io.netty.channel.unix.Errors$NativeIoException: newSocketStream(..) failed: Too many open files
    at io.netty.channel.unix.Errors.newIOException(Errors.java:122) ~[netty-transport-native-unix-common-4.1.27.Final.jar:4.1.27.Final]
    ... 98 common frames omitted

Answer 1

我不认为有。但你可以采取措施防止它。

首先，为什么您的文件描述符限制这么低？Linux为每个打开的套接字打开一个文件描述符，因此如果您打算同时拥有许多打开的套接字，1024是非常低的。我会考虑把这个限制提高很多。

其次，您将并发配置留给调度程序。您应该知道FlatMap运算符的一个变体允许您控制可以并行订阅和合并多少个Publisher：

Flux<V> flatMap(
            Function<? super T,? extends Publisher<? extends V>> mapper,
            int concurrency)