基于Java手写一个通用限流任务执行器

import com.google.common.util.concurrent.ratelimiter;
import java.util.concurrent.*;
import java.util.function.function;

/**
 * 限流任务执行器：按固定速率顺序执行任务，支持重试。
 * @param <t> 任务返回值类型
 */
public class ratelimitedexecutor<t> {
    private final blockingqueue<taskwrapper<t>> queue = new linkedblockingqueue<>();
    private final ratelimiter ratelimiter;
    private final int maxretries;
    private final long initialdelayms;    // 首次重试延迟（毫秒）
    private final double backoffmultiplier; // 退避乘数（如2.0表示每次翻倍）
    private final executorservice worker = executors.newsinglethreadexecutor(r -> {
        thread t = new thread(r, "ratelimitedexecutor-worker");
        t.setdaemon(true);  // 设为守护线程，避免阻止jvm退出
        return t;
    });
    private volatile boolean running = true;

    /**
     * 构造限流执行器
     * @param permitspersecond 每秒允许执行的任务数（如1.0表示每秒1次）
     * @param maxretries       最大重试次数（不含首次执行）
     * @param initialdelayms   首次重试延迟毫秒数
     * @param backoffmultiplier 退避乘数（1.0表示固定延迟，>1.0表示指数退避）
     */
    public ratelimitedexecutor(double permitspersecond, int maxretries,
                               long initialdelayms, double backoffmultiplier) {
        this.ratelimiter = ratelimiter.create(permitspersecond);
        this.maxretries = maxretries;
        this.initialdelayms = initialdelayms;
        this.backoffmultiplier = backoffmultiplier;
        worker.submit(this::process);
    }

    /**
     * 提交一个任务，返回completablefuture异步获取结果
     * @param task 需要执行的任务（callable）
     * @return 代表异步结果的completablefuture
     */
    public completablefuture<t> submit(callable<t> task) {
        completablefuture<t> future = new completablefuture<>();
        queue.offer(new taskwrapper<>(task, future));
        return future;
    }

    // 工作线程主循环
    private void process() {
        while (running) {
            try {
                taskwrapper<t> wrapper = queue.take(); // 阻塞直到有任务
                executewithretry(wrapper);
            } catch (interruptedexception e) {
                thread.currentthread().interrupt();
                break;
            }
        }
    }

    // 执行单个任务（带重试）
    private void executewithretry(taskwrapper<t> wrapper) {
        int retries = 0;
        long delay = initialdelayms;
        while (retries <= maxretries) {
            // 限流：获取令牌，若不足则阻塞
            ratelimiter.acquire();

            try {
                t result = wrapper.task.call();
                wrapper.future.complete(result);
                return; // 成功，结束
            } catch (exception e) {
                retries++;
                if (retries > maxretries) {
                    wrapper.future.completeexceptionally(e);
                    return;
                }
                // 重试等待（退避）
                try {
                    thread.sleep(delay);
                } catch (interruptedexception ie) {
                    thread.currentthread().interrupt();
                    wrapper.future.completeexceptionally(ie);
                    return;
                }
                // 更新下次重试延迟
                delay = (long) (delay * backoffmultiplier);
            }
        }
    }

    /**
     * 优雅关闭执行器：等待已提交任务执行完毕，不再接受新任务
     */
    public void shutdown() {
        running = false;
        worker.shutdown(); // 不再接受新任务
        try {
            if (!worker.awaittermination(5, timeunit.seconds)) {
                worker.shutdownnow();
            }
        } catch (interruptedexception e) {
            worker.shutdownnow();
            thread.currentthread().interrupt();
        }
    }

    /**
     * 立即关闭执行器，尝试中断正在执行的任务
     */
    public void shutdownnow() {
        running = false;
        worker.shutdownnow();
    }

    // 内部任务包装类
    private static class taskwrapper<t> {
        final callable<t> task;
        final completablefuture<t> future;

        taskwrapper(callable<t> task, completablefuture<t> future) {
            this.task = task;
            this.future = future;
        }
    }
}

<dependency>
    <groupid>com.google.guava</groupid>
    <artifactid>guava</artifactid>
    <version>32.1.2-jre</version>
</dependency>

public class taobaoipdemo {
    public static void main(string[] args) throws exception {
        // 创建限流执行器：每秒1次，重试3次，首次延迟1秒，指数退避2.0
        ratelimitedexecutor<string> executor = new ratelimitedexecutor<>(
            1.0,    // 每秒1次
            3,      // 重试3次
            1000,   // 首次延迟1秒
            2.0     // 指数退避
        );

        // 需要查询的ip列表
        string[] ips = {"8.8.8.8", "114.114.114.114", "223.5.5.5"};

        // 提交所有任务
        for (string ip : ips) {
            completablefuture<string> future = executor.submit(() -> queryip(ip));

            // 异步处理结果
            future.thenaccept(result -> {
                system.out.println("ip: " + ip + ", 结果: " + result);
            }).exceptionally(ex -> {
                system.err.println("ip: " + ip + ", 查询失败: " + ex.getmessage());
                return null;
            });
        }

        // 等待所有任务完成（实际应用中不需要，这里仅演示）
        thread.sleep(10000);

        // 优雅关闭
        executor.shutdown();
    }

    private static string queryip(string ip) {
        string url = "https://ip.taobao.com/outgetipinfo?accesskey=alibaba-inc&ip=" + ip;
        return restclient.create()
                .get()
                .uri(url)
                .retrieve()
                .body(string.class);
    }
}