Spring WebFlux&WebClient进行许多API调用

Question

list urls=arrays.aslist（“www.example-rest.com/url1”、“www.example-rest.com/url2”、...、“www.example-rest.com/url600”）；

• 我希望以并行方式并使用无阻塞线程（如Python中的eventlet)
• 我想用这样的线程创建单独的共享工作线程池，以便不为每个传入请求创建一个共享工作线程池。

以下是有关调度器的文档http://projectreactor.io/docs/core/release/reference/#schedulers

但是我不能为每个请求创建新的工作池，池中的线程仍然以阻塞的方式工作。

如果有人用Spring WebClient做类似的任务，请提供一个例子，解释什么是正确的方法。

Answer 1

我也做过类似的事...我的目标是创建一个带有如下签名的方法：

Flux<BasicIssue> getIssues(WebClient webClient);

因为我所调用的网站只提供了一个分页接口，所以我需要将多个REST调用的结果输入到一个流量中。下面是我的实现。请注意我对CachedThreadPool的使用。

Flux<BasicIssue> getIssues(WebClient webClient) {
    return Flux.generate(
        () -> new IssuesState(webClient),
        (state, sink) -> {
            BasicIssue ret = state.getNext();
            if (ret == null) {
                sink.complete();
            } else {
                sink.next(ret);
            }
            return state;
        }
}


class IssuesState {

    private final AtomicBoolean isDone = new AtomicBoolean(false);
    private final AtomicInteger threadCount = new AtomicInteger(1);
    private final Executor executor = Executors.newCachedThreadPool();
    private final LinkedBlockingQueue<BasicIssue> issueQueue = new LinkedBlockingQueue();

    public IssuesState(WebClient webClient) {
        executor.execute(() -> getNextBlock(webClient, 0));
    }

    private void getNextBlock(final WebClient webClient, final int startAt) {
        webClient
            .get()
            .uri(...)
            .header("Authorization", "Basic " + Base64Utils.encodeToString(("username:password".getBytes(UTF_8))))
            .accept(MediaType.APPLICATION_JSON)
            .retrieve()
            .bodyToMono(PageableIssue.class)
            .subscribe(pageableIssue -> {
                int maxResults = pageableIssue.getMaxResults();
                int total = pageableIssue.getTotal();
                if (startAt == 0) {
                    for (int i = startAt + maxResults; i < total; i += maxResults) {
                        threadCount.incrementAndGet();
                        final int x = i;
                        executor.execute(() -> getNextBlock(webClient, x));
                    }
                }
                synchronized (issueQueue) {
                    for (BasicIssue issue : pageableIssue.getIssues()) {
                        issueQueue.add(issue);
                    }

                    if (threadCount.decrementAndGet() == 0) {
                        isDone.set(true);
                    }
                }
            });
    }

    public BasicIssue getNext() {
        synchronized (issueQueue) {
            if (isDone.get() && issueQueue.isEmpty()) {
                return null;
            }
        }
        try {
            return issueQueue.take();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return null;
    }
}

使用上面的方法…

getIssues(webClient)
    .subscribe(basicIssue -> System.out.println(basicIssue.getName());