Очередь пула потоков с уникальными задачами

Использование 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

Что касается вашего требования загрузки объектов из базы данных в память, вы можете взглянуть на Тайники Гуавы.

ОБНОВЛЕНИЕ:

• BlockingQueue при поддержке LinkedHashSet из проекта Apache Marmotta

Решение, аналогичное принятому решению, но основанное на 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());
                }
            });
        }
    }
}

Используйте метод executeIfNotQueuedOrRunningAlready для UniquTaskExecutor, чтобы добиться уникальности выполнения задач.

Если вам разрешено управлять базой данных, я бы предложил использовать саму базу данных, чтобы предотвратить дублирование усилий:

• Добавьте в таблицу столбец lockid
• Добавьте столбец статуса в свою таблицу (возможно, «новый» или «готово»)
• Убедитесь, что уровень изоляции вашей БД не ниже 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 в ноль и установить статус «готово».

Это имеет то преимущество, что работает, даже если у вас есть несколько компьютеров, каждая со своей собственной очередью.