Является ли ExecutorService (в частности, ThreadPoolExecutor) безопасным?

Question

Гарантирует ли ExecutorService резьбовую безопасность?

Я буду отправлять задания из разных потоков одному и тому же ThreadPoolExecutor. Нужно ли синхронизировать доступ к исполнителю перед взаимодействием / отправкой задач?

Answer 1

(В отличие от других ответов) задокументирован контракт о безопасности потоков : посмотрите в interface javadocs (в отличие от javadoc методов). Например, в нижней части ExecutorService javadoc вы найдете:

Эффекты согласованности памяти: действия в потоке до отправка задачи Runnable или Callable в ExecutorService произойдет до любые действия, выполненные этой задачей, которая, в свою очередь, произойдет до результат получен через Future.get ().

Этого достаточно, чтобы ответить на это:

«Должен ли я синхронизировать доступ к исполнителю перед взаимодействием / отправкой задач?»

Нет, нет. Можно создавать и отправлять задания любому (правильно реализованному) ExecutorService без внешней синхронизации. Это одна из главных целей проектирования.

ExecutorService - это одновременная утилита , которая означает, что она разработана для обеспечения максимальной производительности без синхронизации для производительности. (Синхронизация приводит к конфликту потоков, что может снизить эффективность многопоточности, особенно при масштабировании до большого количества потоков.)

Нет никакой гарантии относительно того, в какое время в будущем задачи будут выполняться или завершаться (некоторые могут даже выполняться сразу в том же потоке, который их отправил), однако рабочий поток гарантированно увидит все эффекты, которые выполнил отправляющий поток. до момента подачи . Поэтому (поток, который запускается) ваша задача также может безопасно читать любые данные, созданные для ее использования, без синхронизации, поточно-ориентированных классов или любых других форм «безопасной публикации». Сам процесс отправки задания достаточен для «безопасной публикации» входных данных для задания. Вам просто нужно убедиться, что входные данные не будут изменены каким-либо образом во время выполнения задачи.

Точно так же, когда вы извлекаете результат задания обратно через Future.get(), поток извлечения будет гарантированно видеть все эффекты, сделанные рабочим потоком исполнителя (в обоих возвращаемых результатах, плюс любой побочный эффект меняет работника -поточь, возможно, сделал).

Этот контракт также подразумевает, что сами задачи могут отправлять больше задач.

«Гарантирует ли ExecutorService безопасность потоков?»

Теперь эта часть вопроса гораздо более общая. Например, не удалось найти какое-либо утверждение договора о безопасности потока о методе shutdownAndAwaitTermination - хотя я отмечаю, что пример кода в Javadoc не использует синхронизацию. (Хотя, возможно, есть скрытое предположение, что завершение работы вызвано тем же потоком, который создал Исполнитель, а не, например, рабочим потоком?)

Кстати, я бы порекомендовал книгу «Параллелизм Java на практике», чтобы стать хорошим помощником в мире параллельного программирования.

Answer 2

Это правда, что рассматриваемые классы JDK, по-видимому, не дают явной гарантии поточно-ориентированной отправки задач. Однако на практике все реализации ExecutorService в библиотеке действительно являются поточно-ориентированными. Я думаю, что разумно зависеть от этого. Поскольку весь код, реализующий эти функции, был размещен в открытом доступе, абсолютно никто не может полностью переписать его другим способом.

Answer 3

Ваш вопрос довольно открытый: все, что делает интерфейс ExecutorService, это гарантирует, что какой-нибудь поток где-то обработает отправленный экземпляр Runnable или Callable.

Если отправленный Runnable / Callable ссылается на общую структуру данных, доступную из других экземпляров Runnable / Callable (потенциально обрабатываемых одновременно другими потоками), то это ваша ответственность для обеспечения безопасности потоков в этой структуре данных.

Чтобы ответить на вторую часть вашего вопроса, да, у вас будет доступ к ThreadPoolExecutor перед отправкой каких-либо задач; например,

BlockingQueue<Runnable> workQ = new LinkedBlockingQueue<Runnable>();
ExecutorService execService = new ThreadPoolExecutor(4, 4, 0L, TimeUnit.SECONDS, workQ);
...
execService.submit(new Callable(...));

EDIT

На основании комментария Брайана и в случае, если я неправильно понял ваш вопрос: отправка задач из потоков нескольких производителей в ExecutorService обычно будет поточно-ориентированной (несмотря на то, что, насколько я могу, явно не упоминается в API интерфейса) сказать). Любая реализация, которая не обеспечивала бы безопасность потоков, была бы бесполезна в многопоточной среде (поскольку несколько производителей / несколько потребителей - довольно распространенная парадигма), и именно это было ExecutorService (и остальная часть java.util.concurrent) предназначен для.

Answer 4

Для ThreadPoolExecutor ответ просто да . ExecutorService не не обязывает или иным образом гарантирует, что все реализации являются поточно-ориентированными, и не могут, так как являются интерфейсом. Эти типы контрактов выходят за рамки интерфейса Java. Тем не менее, ThreadPoolExecutor и есть, и четко задокументированы как поточно-ориентированные. Более того, ThreadPoolExecutor управляет своей очередью заданий, используя java.util.concurrent.BlockingQueue, который является интерфейсом, который запрашивает все реализации поточно-ориентированными. Можно предположить, что любая реализация java.util.concurrent.* BlockingQueue является поточно-ориентированной. Любая нестандартная реализация может этого не делать, хотя это было бы глупо, если бы кто-то предоставил очередь реализации BlockingQueue, которая не была бы поточно-ориентированной.

Так что ответ на ваш заглавный вопрос однозначно да . Ответ на следующий текст вашего вопроса: , вероятно , поскольку между ними есть некоторые расхождения.

Answer 5

Для ThreadPoolExecutor, это отправить потокобезопасно. Вы можете увидеть исходный код в jdk8. При добавлении новой задачи он использует mainLock для обеспечения безопасности потока.

private boolean addWorker(Runnable firstTask, boolean core) {
            retry:
            for (;;) {
                int c = ctl.get();
                int rs = runStateOf(c);

                // Check if queue empty only if necessary.
                if (rs >= SHUTDOWN &&
                    ! (rs == SHUTDOWN &&
                       firstTask == null &&
                       ! workQueue.isEmpty()))
                    return false;

                for (;;) {
                    int wc = workerCountOf(c);
                    if (wc >= CAPACITY ||
                        wc >= (core ? corePoolSize : maximumPoolSize))
                        return false;
                    if (compareAndIncrementWorkerCount(c))
                        break retry;
                    c = ctl.get();  // Re-read ctl
                    if (runStateOf(c) != rs)
                        continue retry;
                    // else CAS failed due to workerCount change; retry inner loop
                }
            }

            boolean workerStarted = false;
            boolean workerAdded = false;
            Worker w = null;
            try {
                w = new Worker(firstTask);
                final Thread t = w.thread;
                if (t != null) {
                    final ReentrantLock mainLock = this.mainLock;
                    mainLock.lock();
                    try {
                        // Recheck while holding lock.
                        // Back out on ThreadFactory failure or if
                        // shut down before lock acquired.
                        int rs = runStateOf(ctl.get());

                        if (rs < SHUTDOWN ||
                            (rs == SHUTDOWN && firstTask == null)) {
                            if (t.isAlive()) // precheck that t is startable
                                throw new IllegalThreadStateException();
                            workers.add(w);
                            int s = workers.size();
                            if (s > largestPoolSize)
                                largestPoolSize = s;
                            workerAdded = true;
                        }
                    } finally {
                        mainLock.unlock();
                    }
                    if (workerAdded) {
                        t.start();
                        workerStarted = true;
                    }
                }
            } finally {
                if (! workerStarted)
                    addWorkerFailed(w);
            }
            return workerStarted;
        }

Answer 6

Вопреки тому, что ответ Люка Ашервуда утверждает, из документации не следует, что реализации ExecutorService гарантированно являются поточно-ориентированными. Что касается вопроса ThreadPoolExecutor, см. Другие ответы.

Да, определено отношение произойдет до , но это не подразумевает ничего о поточной безопасности самих методов, как прокомментировали Мили . В ответе Люка Ушервуда указано, что первого достаточно, чтобы доказать второе, но фактических аргументов не приводится.

«Потокобезопасность» может означать разные вещи, но вот простой контрпример к Executor (не ExecutorService, но это не имеет значения), который тривиально соответствует требуемому произойдет до отношение, но не является потокобезопасным из-за несинхронизированного доступа к полю count.

class CountingDirectExecutor implements Executor {

    private int count = 0;

    public int getExecutedTaskCount() {
        return count;
    }

    public void execute(Runnable command) {
        command.run();
    }
}

Отказ от ответственности: я не эксперт, и я нашел этот вопрос, потому что сам искал ответ.