ExecutorService, который выполняет задачи последовательно, но принимает потоки из пула

Ответ 1

Если вы хотите выполнить свою задачу последовательно, просто создайте ExecutorService с помощью только одного потока благодаря Executors.newSingleThreadExecutor().

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

Итак, скажем, что у вас есть 1 000 разные типы задач, вы можете использовать 200 single threaded ExecutorService, единственное, что вам нужно реализовать самостоятельно, это тот факт, что вам всегда нужно использовать один и тот же сингл threaded ExecutorService для заданного типа задачи.

Ответ 2

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

ExecutorService[] es = // many single threaded executors

public <T> Future<T> submit(String key, Callable<T> calls) {
    int h = Math.abs(key.hashCode() % es.length);
    return es[h].submit(calls);
}

В общем случае вам нужны только 2 * N потоки, чтобы поддерживать N ядер в сети, если ваша задача связана с ЦП, больше, чем просто добавляет служебные данные.

Ответ 3

@Николайский ответ, вероятно, лучший выбор, поскольку он прост, хорошо протестирован и эффективен.

Если, однако, это не соответствует вашему требованию, я бы сделал это так:

• Не выполняйте сервис "SequentialPooledExecutor" исполнителя, сделайте его фасад для "пула" служб-исполнителей одного потока.
• Сделайте ваш "SequentialPooledExecutor" реализующим метод отправки (принимает Runnable/Callable и String, представляющий "имя очереди" ), возвращает Будущее, например, службу-исполнитель
• При вызове этого метода сделайте свой "SequentialPooledExecutor" отправкой в ​​один из своих внутренних, одиночных потоков, службой-исполнителем, взяв хэш имени очереди и отправив его соответствующему внутреннему исполнителю.

Хеширующая часть, которая имеет место на шаге 3, позволяет вам выполнять задачи для каждого "имени очереди", всегда переходят к тому же (одному потоку) службе исполнителя внутри вашего "SequentialPooledExecutor".

Другим возможным маршрутом является использование CompletionStage и CompletableFuture s. Это, по сути, прослушиваемые фьючерсы (у которых есть обработчик завершения). При этом при первом сеансе вы создаете CompletableFuture с первой задачей и держитесь за нее. При каждой новой задаче вы объединяете предыдущее будущее с новой задачей, вызывая thenAcceptAsync (или любой другой). Вы получаете линейную цепочку выполнения задач.

Ответ 4

private Map<Integer, CompletableFuture<Void>> sessionTasks = new HashMap<>();
private ExecutorService pool = Executors.newFixedThreadPool(200);

public void submit(int sessionId, Runnable task) {  
    if (sessionTasks.containsKey(sessionId)) {
        sessionTasks.compute(sessionId, (i, c) -> c.thenRunAsync(task, pool));
    } else {
        sessionTasks.put(sessionId, CompletableFuture.runAsync(task, pool));
    }
}

Если сеанс не имеет задачи, в созданном пуле создается и запускается новая задача. Если в сеансе уже есть задачи при добавлении новой задачи, последний привязывается (с thenRun) к предыдущему, обеспечивая порядок.

Ответ 5

Если вы хотите настроить ограниченную очередь, используйте ThreadPoolExecutor

ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, 
TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory,
RejectedExecutionHandler handler)

В вашем случае использования используйте ThreadPoolExecutor как

ThreadPoolExecutor executor =    
ThreadPoolExecutor(1,1,60,TimeUnit.SECONDS,new ArrayBlockingQueue<Runnable>(1000));

Размер кэша превышает размер очереди ThreadPoolExecutor как 1000. Если вы хотите использовать собственный обработчик отклоненных обработок, вы можете настроить RejectedExeutionHandler.

Связанный вопрос SE:

Как правильно использовать Java Executor?

Ответ 6

Недавно я столкнулся с той же проблемой. Для этого нет встроенного класса, но очередь достаточно близка. Моя простая реализация выглядит так (возможно, это полезно для других, которые ищут примеры по одной и той же проблеме)

public class SerializedAsyncRunnerSimple implements Runnable {
private final ExecutorService pool;
protected final LinkedBlockingQueue<Runnable> queue = new LinkedBlockingQueue<>(); //thread safe queue
protected final AtomicBoolean active = new AtomicBoolean(false);


public SerializedAsyncRunnerSimple(ExecutorService threadPool) {this.pool = threadPool;}


public void addWork(Runnable r){        
    queue.add(r);
    startExecutionIfInactive();
}

private void startExecutionIfInactive() {
    if(active.compareAndSet(false, true)) {
        pool.execute(this);
    }
}

@Override
public synchronized void run() {
    while(!queue.isEmpty()){
        queue.poll().run();
    }
    active.set(false); //all future adds will not be executed on this thread anymore
    if(!queue.isEmpty()) { //if some items were added to the queue after the last queue.poll
        startExecutionIfInactive();// trigger an execution on next thread
    }
}