spring - 如何使用 WebClient 限制请求/秒？

Question

我正在使用 WebClient对象将 Http Post 请求发送到服务器。
它非常迅速地发送大量请求(在 QueueChannel 中有大约 4000 条消息)。问题是...似乎服务器的响应速度不够快...所以我收到了很多服务器错误 500 并且连接过早关闭。

有没有办法限制每秒请求的数量？或者限制它使用的线程数？

编辑:

Message 端点处理 QueueChannel 中的消息:

@MessageEndpoint
public class CustomServiceActivator {

    private static final Logger logger = LogManager.getLogger();

    @Autowired
    IHttpService httpService;

    @ServiceActivator(
            inputChannel = "outputFilterChannel",
            outputChannel = "outputHttpServiceChannel",
            poller = @Poller( fixedDelay = "1000" )
    )
    public void processMessage(Data data) {
        httpService.push(data);
        try {
            Thread.sleep(20);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

}

WebClient 服务类:

@Service
public class HttpService implements IHttpService {

    private static final String URL = "http://www.blabla.com/log";

    private static final Logger logger = LogManager.getLogger();

    @Autowired
    WebClient webClient;

    @Override
    public void push(Data data) {
        String body = constructString(data);
        Mono<ResponseEntity<Response>> res = webClient.post()
                .uri(URL + getLogType(data))
                .contentLength(body.length())
                .contentType(MediaType.APPLICATION_JSON)
                .syncBody(body)
                .exchange()
                .flatMap(response -> response.toEntity(Response.class));

        res.subscribe(new Consumer<ResponseEntity<Response>>() { ... });
    }
}

Answer 1

问题 Limiting rate of requests with Reactor提供两个答案(评论中的一个)

zip 使用另一种作为速率限制器的通量

.zipWith(Flux.interval(Duration.of(1, ChronoUnit.SECONDS)))

只是延迟每个 Web 请求

使用延迟元素函数

编辑:下面的答案对于阻止 RestTemplate 有效，但不太适合 react 模式。

WebClient 无法限制请求，但您可以使用组合轻松添加此功能。

您可以使用来自 Guava/的 RateLimiter 从外部限制您的客户端
( https://google.github.io/guava/releases/19.0/api/docs/index.html?com/google/common/util/concurrent/RateLimiter.html )

在本教程中 http://www.baeldung.com/guava-rate-limiter您将找到如何以阻塞方式或超时方式使用速率限制器。

我会在单独的类中装饰所有需要限制的调用

限制每秒调用次数

使用 WebClient 执行实际的网络调用