如何在springframework RestClient中的每个api调用之间暂停

Question

这是一个

org.springframework.web.client.RestClient

配置

@Configuration
public class CloudConfig {
    @Value("${experimental.host}")
    String baseURI;
    @Bean
    RestClient restClient(RestClient.Builder builder){
        return builder.baseUrl(baseURI).build();
    }
}

这是调用外部 API 的服务

@Component
public class RequestService {

    public String callAPI(int userID){
        Map<String, Object> properties = Map.of("id", userID);
        return restClient.post()
                .uri("/external")
                .body(properties)
                .retrieve()
                .body(String.class);
    }
}

我从数据库获取用户列表并在循环中调用外部API

@Service
public class RabbitMQProducer {

    private UserRepository repository;
    private RequestService requestService;

    @Scheduled(fixedRate = 10000)
    public void sendUserData(){
        for(User user : repository.findAll()) {
            String data = requestService.callAPI(user.getID);
            ......
        }
    }   
}

由于外部 api 的调用时间限制为 1 秒，因此在每次调用之间暂停的正确方法是什么？ 我从 API“org.springframework.web.client.HttpClientErrorException$TooManyRequests: 1000 毫秒内请求次数过多 2 次”收到错误消息

有什么模式或解决方案可以解决此类问题吗？

我认为预期的行为：

调用 API --> 1 秒等待 --> 调用 API --> 1 秒等待 ...

解决这个问题的简单方法只是添加

Thread.sleep(1000)

但我不确定这是一个好的解决方案

Answer 1

如果使用 Thread.sleep(1000)，那么它会限制 1 次调用 1 秒，但这并不理想，因为它会阻塞当前线程。

1. Webflux 和 webClient

我用这个。

   public Mono<String> callAPI(User user) {
        return webClient.post()
                .uri("/external")
                .bodyValue(Map.of("id", user.getID()))
                .retrieve()
                .bodyToMono(String.class)
                .delayElement(Duration.ofSeconds(1)); // Delay for each subscriber
    }

2.韧性4j

这是限速库。

import io.github.resilience4j.ratelimiter.RateLimiter;
import io.github.resilience4j.ratelimiter.RateLimiterConfig;
import java.time.Duration;

@Configuration
public class RateLimiterConfig {

    @Bean
    public RateLimiter rateLimiter() {
        RateLimiterConfig config = RateLimiterConfig.custom()
            .limitRefreshPeriod(Duration.ofSeconds(1))
            .limitForPeriod(1)
            .timeoutDuration(Duration.ofMillis(500))
            .build();

        return RateLimiter.of("apiRateLimiter", config);
    }
}

.

@Service
public class RateLimitedRequestService {

    private final RequestService requestService;
    private final RateLimiter rateLimiter;

    public RateLimitedRequestService(RequestService requestService, RateLimiter rateLimiter) {
        this.requestService = requestService;
        this.rateLimiter = rateLimiter;
    }

    public String callAPI(User user) {
        return RateLimiter.decorateSupplier(rateLimiter, () -> requestService.callAPI(user.getID())).get();
    }
}

2. @异步

您应该添加@EnableAsync

@Async
public CompletableFuture<String> callAPIWithDelay(User user, long delay) {
    return CompletableFuture.supplyAsync(() -> {
        try {
            Thread.sleep(delay);
            return requestService.callAPI(user.getID());
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new IllegalStateException(e);
        }
    });
}

+a，重试

我建议您添加重试以确保正常工作。当使用具有速率限制的外部 API 时，建议制作一些重试逻辑。