java - 具有独特任务的线程池队列

Question

我正在使用 ThreadPoolTaskExecutor (spring 的)为了异步执行一些任务。

所需的任务将从外部数据库加载一些对象到我的系统内存中。 我使用的最大线程池大小为 10，最大队列大小为 100。

假设所有 10 个线程都被占用从我的数据库中获取对象并创建一个任务，它将进入队列。现在创建了另一个任务，它应该从数据库中获取相同的对象(数据库中的相同键)，它也会进入队列(假设所有 10 个线程仍然被占用)。

所以我的队列可能很容易被重复的任务填满，这些任务将依次执行，我不希望这种情况发生。

我认为解决方案应该以充当线程池队列的唯一集合的形式出现。 在底层，ThreadPoolTask​​Executor 使用不提供唯一性的 LinkedBlockingQueue。

我想到了一些可能的解决方案，但没有一个让我满意:

• 使用ThreadPoolExecutor而不是 ThreadPoolTask​​Executor。 ThreadPoolExecutor 提供了一个构造函数，可以让我确定线程池队列类型，但它需要实现 BlockingQueue 接口(interface)。我找不到保留唯一性的实现。

这让我尝试扩展 LinkedBlockingQueue并覆盖添加:

public boolean add(E e)
    if(!this.contains(e)) {
        return super.add(e);
    } else {
        return false;
    }
}

但据我所知，这将导致性能大幅下降，因为 contains 方法受 O(n) 的限制——这是个坏主意。

什么可以解决我的问题？我的目标是获得良好的性能(在内存性能折衷的情况下，我不介意为了性能而放弃内存)。

Answer 1

使用 Guava和 ListenableFuture你可以做类似的事情(还没有测试过)

Set<String> uniqueQueue = Sets.newConcurrentHashSet();
ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(10, 10, 0, TimeUnit.SECONDS, Queues.newLinkedBlockingQueue(100));
ListeningExecutorService executorService = MoreExecutors.listeningDecorator(threadPoolExecutor);

String t1 = "abc";
if(uniqueQueue.add(t1)) {
    ListenableFuture<String> future = executorService.submit(() -> "do something with " + t1);
    Futures.addCallback(future, new FutureCallback<String>() {
        @Override
        public void onSuccess(String result) {
            uniqueQueue.remove(t1);
        }

        @Override
        public void onFailure(Throwable t) {
            uniqueQueue.remove(t1);
        }
    });
}

导致

• 只有当前没有被处理或在队列中的项目才会被添加到队列中(uniqueQueue)
• 已处理的项目将从 uniqueQueue
中移除
• 队列中最多只能有 100 个项目

这个实现不处理

• Exceptions 由 submit() 方法抛出
• unqiueQueue 中的最大项目数

关于将对象从数据库加载到内存中的需求，您可能需要查看 Guava's Caches .

更新:

• BlockingQueue backed by LinkedHashSet from Apache Marmotta project