我使用ThreadPoolTaskExecutor(spring的)来异步执行一些任务。
所需的任务将把一些对象从外部数据库加载到我的系统内存中。我使用的最大线程池大小为10,最大队列大小为100。
假设所有10个线程都在从我的DB中获取对象,并且创建了一个任务,它将进入队列。现在创建了另一个任务,该任务应该从DB中获得相同的对象(DB中的相同键),它也将进入队列(假设所有10个线程仍被占用)。
因此,我的队列可能很容易被重复的任务填满,这些任务将依次执行,我不希望发生这种情况。
我认为解决方案应该以一个唯一集合的形式出现,该集合充当线程池队列。在后台,ThreadPoolTaskExecutor使用不提供唯一性的LinkedBlockingQueue。
我想到了几个可能的解决方案,但没有一个能让我满意:
- 使用ThreadPoolExecutor而不是ThreadPoolTaskExecutor。ThreadPoolExecutor提供了一个构造函数,让我可以确定线程池队列类型,但它需要实现BlockingQueue接口。我找不到一个保持独特性的实现
这让我尝试扩展LinkedBlockingQueue并覆盖add:
public boolean add(E e)
if(!this.contains(e)) {
return super.add(e);
} else {
return false;
}
}
但据我所知,这将导致性能的大幅下降,因为contains
方法受到O(n)-坏主意的限制。
什么能解决我的问题?我的目标是获得好的性能(在内存性能权衡的情况下,我不介意为了性能而放弃内存)。
使用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件商品
该实现不处理
submit()
方法抛出的Exceptions
unqiueQueue
中的最大项目数
关于将对象从数据库加载到内存的需求,您可能需要查看Guava的缓存。
更新:
- Apache Marmotta项目的LinkedHashSet支持的BlockingQueue
如果允许您管理数据库,我建议使用数据库本身来防止重复工作:
- 向表中添加lockid列
- 向表中添加状态列(可能是"new"one_answers"done")
- 确保您的数据库隔离级别至少为READ_COMMITTED
然后在你的主线程中尝试这样的东西:
Random rand = new Random();
int lockId = rand.nextInt(Integer.MAX_VALUE - 1) + 1;
String update = "UPDATE DB.Table SET lockid=" + lockId + " WHERE lockid=0 AND status='new' " // + AND your conditions + LIMIT ##
String select = "SELECT * FROM DB.Table WHERE lockid=" + lockId;
// now execute those sql statements with QueryRunner or whatever you use in-house
从选择返回的行就是您添加到队列中的行。
然后,您有一个实现Runnable的类,通过从队列中检索这些行来处理这些行。一旦它处理了一行,就可以执行另一个SQL更新(在Runnable中),将lockId设置回零,并将状态设置为"done"。
这样做的好处是,即使您有多台机器,每台机器都有自己的队列,也能正常工作。
类似于已接受的解决方案,但基于Spring(与Guava相反)的解决方案:
创建接口RunnableWithId:
public interface RunnableWithId extends Runnable {
/**
* @return A unique id for this task
*/
String getTaskId();
}
创建另一个接口TaskWithIdExecutor:
import org.springframework.core.task.TaskExecutor;
public interface TaskWithIdExecutor extends TaskExecutor {
/**
* Executes the given task if it is not queued or already running
*
* @param task The task to execute
*/
void executeIfNotQueuedOrRunningAlready(RunnableWithId task);
}
创建您的自定义执行器UniquTaskExecutor:
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;
import java.util.Set;
/**
* In addition to all the abilities of ThreadPoolTaskExecutor adds the ability
* to execute a task only if it is not already running/queued using the
* executeIfNotQueuedOrRunningAlready method.
*
* @see ThreadPoolTaskExecutor
*/
public class UniquTaskExecutor extends ThreadPoolTaskExecutor implements TaskWithIdExecutor {
private Set<String> queuedTasks;
public UniquTaskExecutor() {
queuedTasks = Sets.newConcurrentHashSet();
}
@Override
public void execute(Runnable task) {
super.execute(task);
}
/**
* @param task The task to execute
*/
@Override
public void executeIfNotQueuedOrRunningAlready(RunnableWithId task) {
if (queuedTasks.add(task.getTaskId())) {
ListenableFuture<?> res = submitListenable(task);
res.addCallback(new ListenableFutureCallback<Object>() {
@Override
public void onFailure(Throwable throwable) {
queuedTasks.remove(task.getTaskId());
}
@Override
public void onSuccess(Object o) {
queuedTasks.remove(task.getTaskId());
}
});
}
}
}
使用UniquTaskExecutor的executeIfNotQueuedOrRunningAlready方法来实现任务执行的唯一性。