快捷導(dǎo)航

Java線程池如何實(shí)現(xiàn)精準(zhǔn)控制每秒API請(qǐng)求

更新時(shí)間：2024年08月22日 15:05:21 作者：promise524

這篇文章主要介紹了Java線程池如何實(shí)現(xiàn)精準(zhǔn)控制每秒API請(qǐng)求問(wèn)題,具有很好的參考價(jià)值,希望對(duì)大家有所幫助,如有錯(cuò)誤或未考慮完全的地方,望不吝賜教

Java中基于線程池實(shí)現(xiàn)指定每秒發(fā)送一定數(shù)量的API請(qǐng)求，可以使用ScheduledExecutorService來(lái)調(diào)度任務(wù)，同時(shí)使用ThreadPoolExecutor來(lái)處理并發(fā)請(qǐng)求，可以根據(jù)實(shí)際需求調(diào)整每秒請(qǐng)求數(shù)量、執(zhí)行時(shí)間、以及線程池大小。

實(shí)現(xiàn)思路

1.創(chuàng)建線程池

使用Executors.newScheduledThreadPool()來(lái)創(chuàng)建一個(gè)調(diào)度線程池
并使用Executors.newFixedThreadPool()來(lái)創(chuàng)建一個(gè)用于發(fā)送API請(qǐng)求的線程池

2.調(diào)度任務(wù)

使用ScheduledExecutorService來(lái)按固定速率調(diào)度任務(wù)。
通過(guò)控制任務(wù)的頻率，可以確保每秒發(fā)送指定數(shù)量的請(qǐng)求。

3.定義API請(qǐng)求任務(wù)

定義一個(gè)實(shí)現(xiàn)Runnable接口的類(lèi)
負(fù)責(zé)執(zhí)行具體的API請(qǐng)求

4.控制請(qǐng)求速率

使用調(diào)度器每秒提交指定數(shù)量的任務(wù)到線程池中執(zhí)行。

引入依賴(lài)

    <!-- Apache HttpClient -->
    <dependency>
        <groupId>org.apache.httpcomponents.client5</groupId>
        <artifactId>httpclient5</artifactId>
        <version>5.2</version>
    </dependency>

實(shí)現(xiàn)代碼

import org.apache.hc.client5.http.classic.methods.HttpGet;
import org.apache.hc.client5.http.classic.methods.HttpUriRequestBase;
import org.apache.hc.client5.http.impl.classic.CloseableHttpClient;
import org.apache.hc.client5.http.impl.classic.CloseableHttpResponse;
import org.apache.hc.client5.http.impl.classic.HttpClients;
import org.apache.hc.core5.http.ParseException;
import org.apache.hc.core5.http.io.entity.EntityUtils;

import java.io.IOException;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

public class ApiRequestScheduler {

    // 定義線程池，用于并發(fā)發(fā)送API請(qǐng)求
    private final ExecutorService requestExecutor;

    // 定義調(diào)度線程池，用于定時(shí)調(diào)度請(qǐng)求任務(wù)
    private final ScheduledExecutorService scheduler;

    // 記錄已發(fā)送的請(qǐng)求數(shù)量
    private final AtomicInteger requestCounter;

    // 每秒發(fā)送的請(qǐng)求數(shù)量
    private final int requestsPerSecond;

    // Apache HttpClient 實(shí)例
    private final CloseableHttpClient httpClient;

    // API 請(qǐng)求的目標(biāo)URL
    private final String apiUrl;

    // 構(gòu)造函數(shù)，初始化線程池和調(diào)度器
    public ApiRequestScheduler(int requestsPerSecond, String apiUrl) {
        this.requestsPerSecond = requestsPerSecond;
        this.requestExecutor = Executors.newFixedThreadPool(requestsPerSecond);
        this.scheduler = Executors.newScheduledThreadPool(1);
        this.requestCounter = new AtomicInteger(0);
        this.httpClient = HttpClients.createDefault();
        this.apiUrl = apiUrl;
    }

    // 開(kāi)始調(diào)度API請(qǐng)求任務(wù)
    public void start() {
        // 每秒調(diào)度任務(wù)，按照每秒發(fā)送的請(qǐng)求數(shù)量來(lái)執(zhí)行
        scheduler.scheduleAtFixedRate(() -> {
            for (int i = 0; i < requestsPerSecond; i++) {
                requestExecutor.submit(this::sendApiRequest);
            }
        }, 0, 1, TimeUnit.SECONDS);
    }

    // 停止調(diào)度和關(guān)閉線程池及HttpClient
    public void stop() {
        scheduler.shutdown();
        requestExecutor.shutdown();
        try {
            if (!scheduler.awaitTermination(1, TimeUnit.SECONDS)) {
                scheduler.shutdownNow();
            }
            if (!requestExecutor.awaitTermination(1, TimeUnit.SECONDS)) {
                requestExecutor.shutdownNow();
            }
            httpClient.close();
        } catch (InterruptedException | IOException e) {
            scheduler.shutdownNow();
            requestExecutor.shutdownNow();
            try {
                httpClient.close();
            } catch (IOException ioException) {
                ioException.printStackTrace();
            }
        }
    }

    // 使用Apache HttpClient發(fā)送API請(qǐng)求
    private void sendApiRequest() {
        int requestId = requestCounter.incrementAndGet();
        HttpUriRequestBase request = new HttpGet(apiUrl);
        System.out.println("Sending API request #" + requestId);

        try (CloseableHttpResponse response = httpClient.execute(request)) {
            String responseBody = EntityUtils.toString(response.getEntity());
            System.out.println("Request #" + requestId + " completed with status: " + response.getCode() +
                    ", response: " + responseBody);
        } catch (IOException | ParseException e) {
            System.err.println("Request #" + requestId + " failed: " + e.getMessage());
        }
    }

    public static void main(String[] args) {
        // 每秒發(fā)送5個(gè)API請(qǐng)求，目標(biāo)URL為http://example.com/api
        ApiRequestScheduler scheduler = new ApiRequestScheduler(5, "http://www.dzy.com/api");

        // 啟動(dòng)調(diào)度器
        scheduler.start();

        // 運(yùn)行10秒后停止調(diào)度器
        try {
            Thread.sleep(10000);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }

        // 停止調(diào)度器
        scheduler.stop();
    }
}