Java并发任务实现

Question

我的问题是这样的：我最多可以运行三个并发任务。这些任务可以同时处理1到100个作业。我有很多线程不断地提交单个作业，我想尽可能快地响应它们。在一个任务中处理100个作业所花费的时间与在一个任务中处理1个作业所花费的时间相同。作业随机间隔出现。提交作业的线程需要阻塞，直到作业完成，或者遇到超时。快速响应提交作业的线程是这里的驱动程序。

所以我目前的逻辑是：如果有

我只是不太确定在Java中设置它的最佳方式。我创建了一个简单的信号量版本，它运行良好，但没有利用同时提交作业的能力。我应该如何最好地扩展它以完全满足我的要求？（没有使用信号量的要求，这只是我到目前为止拥有的）。

private static final Semaphore semaphore = new Semaphore(3);

public static Response doJob(Job job) throws Exception
{ 
    final boolean tryAcquire = this.semaphore.tryAcquire(this.maxWaitTime, TimeUnit.MILLISECONDS);

    if (tryAcquire)
    {
        try
        {
            return doJobInNewTask(job); // we'd actually like to do all the jobs which are queued up waiting for the semaphore (if there are any)
        }
        finally
        {
            this.semaphore.release()
        }       
    }
}

Answer 1

您可以将Executor服务与固定大小的线程池一起使用：

class ExecutorExample {
    private final static ExecutorService executorService;
    private final static long maxWaitTime = 5000;

    static {
        executorService = Executors.newFixedThreadPool(3);
    }

    private static class Response {}
    private static class Job {}

    public static Response doJob(final Job job) throws Exception {
        final Future<Response> future = executorService.submit(
            new Callable<Response>() {
                @Override
                public Response call() throws Exception {
                    return doJobInNewTask(job);
                }
            }
        );
        try {
            // get() blocks until the task finishes.
            return future.get(maxWaitTime, TimeUnit.MILLISECONDS);
        }
        catch (final TimeoutException e) {
            // we timed out, so *try* to cancel the task (may be too late)
            future.cancel(/*mayInterruptIfRunning:*/false);
            throw e;
        }
    }

    private static Response doJobInNewTask(final Job job) {
        try { Thread.sleep(maxWaitTime / 2); }
        catch (final InterruptedException ignored) {}
        return new Response();
    }

    public static void main(final String[] args) {
        final List<Thread> threads = new ArrayList<>();

        for (int i = 0; i < 10; i++) {
            final Thread t = new Thread() {
                @Override
                public void run() {
                    try {
                        System.out.println(doJob(new Job()));
                    }
                    catch (final Exception e) {
                        System.out.println(e.getClass().getSimpleName());
                    }
                }
            };
            threads.add(t);
            t.start();
        }

        for (final Thread thread : threads) {
            try { thread.join(); }
            catch (final InterruptedException ignored) {}
        }

        System.out.println("Done!");
    }
}

输出：

ExecutorExample$Response@1fe4169
ExecutorExample$Response@9fdee
ExecutorExample$Response@15b123b
ExecutorExample$Response@bbfa5c
ExecutorExample$Response@10d95cd
ExecutorExample$Response@131de9b
TimeoutException
TimeoutException
TimeoutException
TimeoutException
Done!

这里的一个潜在问题是取消。由于日程安排已超出您的控制范围，因此有可能在您等待任务超时后，但在cancel（）有机会完成任务之前，启动任务。结果不会传播，但如果任务有有意义的副作用，这种方法可能会产生问题。