import com.google.common.util.concurrent.RateLimiter;
import java.util.concurrent.*;
import java.util.function.Function;

/**
 * 限流任務(wù)執(zhí)行器：按固定速率順序執(zhí)行任務(wù)，支持重試。
 * @param <T> 任務(wù)返回值類型
 */
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; // 退避乘數(shù)（如2.0表示每次翻倍）
    private final ExecutorService worker = Executors.newSingleThreadExecutor(r -> {
        Thread t = new Thread(r, "RateLimitedExecutor-Worker");
        t.setDaemon(true);  // 設(shè)為守護線程，避免阻止JVM退出
        return t;
    });
    private volatile boolean running = true;

    /**
     * 構(gòu)造限流執(zhí)行器
     * @param permitsPerSecond 每秒允許執(zhí)行的任務(wù)數(shù)（如1.0表示每秒1次）
     * @param maxRetries       最大重試次數(shù)（不含首次執(zhí)行）
     * @param initialDelayMs   首次重試延遲毫秒數(shù)
     * @param backoffMultiplier 退避乘數(shù)（1.0表示固定延遲，>1.0表示指數(shù)退避）
     */
    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);
    }

    /**
     * 提交一個任務(wù)，返回CompletableFuture異步獲取結(jié)果
     * @param task 需要執(zhí)行的任務(wù)（Callable）
     * @return 代表異步結(jié)果的CompletableFuture
     */
    public CompletableFuture<T> submit(Callable<T> task) {
        CompletableFuture<T> future = new CompletableFuture<>();
        queue.offer(new TaskWrapper<>(task, future));
        return future;
    }

    // 工作線程主循環(huán)
    private void process() {
        while (running) {
            try {
                TaskWrapper<T> wrapper = queue.take(); // 阻塞直到有任務(wù)
                executeWithRetry(wrapper);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                break;
            }
        }
    }

    // 執(zhí)行單個任務(wù)（帶重試）
    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; // 成功，結(jié)束
            } 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);
            }
        }
    }

    /**
     * 優(yōu)雅關(guān)閉執(zhí)行器：等待已提交任務(wù)執(zhí)行完畢，不再接受新任務(wù)
     */
    public void shutdown() {
        running = false;
        worker.shutdown(); // 不再接受新任務(wù)
        try {
            if (!worker.awaitTermination(5, TimeUnit.SECONDS)) {
                worker.shutdownNow();
            }
        } catch (InterruptedException e) {
            worker.shutdownNow();
            Thread.currentThread().interrupt();
        }
    }

    /**
     * 立即關(guān)閉執(zhí)行器，嘗試中斷正在執(zhí)行的任務(wù)
     */
    public void shutdownNow() {
        running = false;
        worker.shutdownNow();
    }

    // 內(nèi)部任務(wù)包裝類
    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 {
        // 創(chuàng)建限流執(zhí)行器：每秒1次，重試3次，首次延遲1秒，指數(shù)退避2.0
        RateLimitedExecutor<String> executor = new RateLimitedExecutor<>(
            1.0,    // 每秒1次
            3,      // 重試3次
            1000,   // 首次延遲1秒
            2.0     // 指數(shù)退避
        );

        // 需要查詢的IP列表
        String[] ips = {"8.8.8.8", "114.114.114.114", "223.5.5.5"};

        // 提交所有任務(wù)
        for (String ip : ips) {
            CompletableFuture<String> future = executor.submit(() -> queryIp(ip));

            // 異步處理結(jié)果
            future.thenAccept(result -> {
                System.out.println("IP: " + ip + ", 結(jié)果: " + result);
            }).exceptionally(ex -> {
                System.err.println("IP: " + ip + ", 查詢失敗: " + ex.getMessage());
                return null;
            });
        }

        // 等待所有任務(wù)完成（實際應(yīng)用中不需要，這里僅演示）
        Thread.sleep(10000);

        // 優(yōu)雅關(guān)閉
        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);
    }
}