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!

这里的一个潜在问题是取消。由于调度超出了您的控制，因此可能会在您等待超时后启动任务，但 before cancel() 有机会做它的事情。结果不会被传播，但如果任务有有意义的副作用，这种方法可能会产生问题。