Java节流

Question

如何组合使用ScheduledThreadPoolExecutor、ScheduledFuture和ExecutorCompletionService来限制接受变量参数的Callable命令？在收到来自Callable命令的响应时，我需要根据前面提到的Callable命令的输出创建一个新的Callable命令。我还需要遵守每秒100个呼叫的阈值。

Answer 1

您应该实现the Leaky Bucket algorithm。在进行呼叫之前，请阻塞，直到您有令牌为止。您可以在几十行Java代码中实现此算法。

Answer 2

我建议使用代理，例如RabbitMQ。您可以将最大消费者数量设置为100，并拥有单个生产者实例，该实例以每秒100条消息的速度发布。

Here您可以找到在分布式系统中实现节流机制的三种方法的说明。您感兴趣的是使用RabbitMQ的分布式应用程序。此方法旨在限制任何给定时间的并发消息数，假设在任何给定时间最多不超过100条。您需要对其进行修改，以便发布者每秒发布的消息不超过100条。在底部，你可以找到一个包含源代码的git资源库的url，但无论如何，我也粘贴了here。

根据注释进行编辑：

首先使用java.util.Semaphore，它配置了它将处理的许可数量。每个线程都会尝试获取一个许可证，如果没有许可证，则会被阻塞，直到释放一个许可证为止。第二个使用固定大小的ThreadPoolExecutor。executor在任何给定时间最多具有指定数量的工作线程。第三个使用RabbitMQ。并发使用者的最大数量将是工作线程的最大数量。git repo有更详细的英文解释。希望这能有所帮助

Answer 3

您可以使用信号量进行节流。你需要区分你是A)“每秒节流”(同一时间内作业的上限)还是B)“每间隔节流”(一个区间内作业的上限)

A)对信号量进行上下计数就足以达到“每一瞬间的节流”。例如。

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
import java.util.concurrent.atomic.AtomicInteger;

public class ThrottlePerPerInstantSample {
    private static final int JOBS_COUNT = 100;
    private static final int JOBS_THROTTLE_PER_INSTANT = 10;
    private static final Semaphore THROTTLE_PER_INSTANT_SEMAPHORE = new Semaphore(
            JOBS_THROTTLE_PER_INSTANT);
    private static final ExecutorService executorService = Executors
            .newFixedThreadPool(JOBS_THROTTLE_PER_INSTANT);

    private final static AtomicInteger jobsAtTheSameTimeCounter = new AtomicInteger(
            0);

    public static void main(String[] args) throws InterruptedException {
        for (int i = 1; i <= JOBS_COUNT; i++) {
            THROTTLE_PER_INSTANT_SEMAPHORE.acquire();
            final PrintJob printJob = new PrintJob(i, jobsAtTheSameTimeCounter);
            final ThrottledJob throttledJob = new ThrottledJob(printJob,
                    THROTTLE_PER_INSTANT_SEMAPHORE);
            executorService.execute(throttledJob);
        }
        executorService.shutdown();
    }

    static class ThrottledJob implements Runnable {
        private final Runnable delegate;
        private final Semaphore throttlePerInstantSemaphore;

        public ThrottledJob(Runnable delegate,
                Semaphore throttlePerInstantSemaphore) {
            super();
            this.delegate = delegate;
            this.throttlePerInstantSemaphore = throttlePerInstantSemaphore;
        }

        @Override
        public void run() {
            try {
                delegate.run();
            } finally {
                throttlePerInstantSemaphore.release();
            }
        }

    }

    static class PrintJob implements Runnable {
        final int jobNumber;
        final AtomicInteger jobsAtTheSameTimeCounter;

        public PrintJob(int jobNumber, AtomicInteger jobsAtTheSameTimeCounter) {
            super();
            this.jobNumber = jobNumber;
            this.jobsAtTheSameTimeCounter = jobsAtTheSameTimeCounter;
        }

        public void run() {
            jobsAtTheSameTimeCounter.incrementAndGet();

            try {
                Thread.sleep(50); // wait some time
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

            synchronized (System.out) {
                System.out.println(jobsAtTheSameTimeCounter.getAndDecrement()
                        + " : Job " + jobNumber);
            }
        }
    }
}

输出可以是：

10 : Job 1
9 : Job 2
8 : Job 3
7 : Job 5
7 : Job 4
...
9 : Job 87
10 : Job 90
9 : Job 89
10 : Job 91
9 : Job 92
8 : Job 93
7 : Job 94
6 : Job 95
5 : Job 98
4 : Job 97
3 : Job 96
2 : Job 100
1 : Job 99

B)对信号量进行倒计时，并周期性地将信号量重置为其初始值，这对于“每次间隔的节流”是足够的。

例如。

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;

public class ThrottlePerIntervallSample {
    private static final int JOBS_COUNT = 100;
    private static final int JOBS_THROTTLE_PER_INTERVALL = 10;
    private static final long INTERVALL_IN_UNITS = 1;
    private static final TimeUnit UNIT_OF_INTERVALL = TimeUnit.SECONDS;

    private static final Semaphore THROTTLE_PER_INTERVALL_SEMAPHORE = new Semaphore(
            JOBS_THROTTLE_PER_INTERVALL);

    private static final ScheduledExecutorService SCHEDULED_EXECUTOR_SERVICE = Executors
            .newScheduledThreadPool(JOBS_THROTTLE_PER_INTERVALL + 1); 
    // plus one because the resetting of the semaphore must be possible! 

    public static void main(String[] args) throws InterruptedException {

        SCHEDULED_EXECUTOR_SERVICE.scheduleAtFixedRate(()-> {
            THROTTLE_PER_INTERVALL_SEMAPHORE.drainPermits(); // remove permits from previous intervall
        THROTTLE_PER_INTERVALL_SEMAPHORE.release(JOBS_THROTTLE_PER_INTERVALL); // set permits for the next intervall
        }, INTERVALL_IN_UNITS, INTERVALL_IN_UNITS, UNIT_OF_INTERVALL);

        for (int i = 1; i <= JOBS_COUNT; i++) {
            THROTTLE_PER_INTERVALL_SEMAPHORE.acquire();
            final PrintJob printJob = new PrintJob(i);
            SCHEDULED_EXECUTOR_SERVICE.execute(printJob);
        }

        SCHEDULED_EXECUTOR_SERVICE.shutdown();
    }

    static class PrintJob implements Runnable {
        final int jobNumber;

        public PrintJob(int jobNumber) {
            super();
            this.jobNumber = jobNumber;
        }

        public void run() {

            try {
                Thread.sleep(50); // wait some time
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

            SimpleDateFormat simpleDateFormat = new SimpleDateFormat("HH:mm:ss:SSS");

            synchronized (System.out) {
                System.out.println(simpleDateFormat.format(new Date())
                        + " : Job " + jobNumber);
            }
        }
    }

}

输出可以是：

00:42:29:253 : Job 9
00:42:29:255 : Job 2
00:42:29:255 : Job 6
00:42:29:255 : Job 5
00:42:29:255 : Job 10
00:42:29:256 : Job 7
00:42:29:256 : Job 3
00:42:29:256 : Job 1
00:42:29:256 : Job 8
00:42:29:257 : Job 4
00:42:30:140 : Job 11
...
00:42:37:142 : Job 90
00:42:38:140 : Job 91
00:42:38:140 : Job 92
00:42:38:141 : Job 99
00:42:38:141 : Job 93
00:42:38:141 : Job 94
00:42:38:142 : Job 98
00:42:38:142 : Job 96
00:42:38:142 : Job 95
00:42:38:143 : Job 100
00:42:38:143 : Job 97

以下是一些备注：

1)在生产系统中，最好使用带超时的tryAcquire方法，而不是acquire，以避免实际的死锁！

2)如果您要处理多个作业，在将作业提交给(定时) executor服务之前，请先调用aquire/tryAquire。否则，您可能会在此时使用太多作业污染线程池的队列。