Как пользоваться concurrent.futures и чем ThreadPoolExecutor отличается от ProcessPoolExecutor?
- Как распараллелить задачи в Python без ручного управления потоками?
Единый интерфейс пулов: submit возвращает Future, map применяет функцию к итерируемому, as_completed отдаёт результаты по готовности. ThreadPoolExecutor — для ввода-вывода (GIL отпускается на ожидании), ProcessPoolExecutor — для вычислений (свой интерпретатор и GIL в каждом процессе).
- Исключение задачи хранится в Future и бросается при result()
- Процессам аргументы и результаты передаются через pickle — лямбды и открытые соединения не пройдут
- with гарантирует shutdown и ожидание задач
А ещё max_workers по умолчанию: потоки — min(32, CPU + 4), процессы — число CPU
Подробно
Модель: пул рабочих, очередь задач, Future как дескриптор результата. submit(fn, *args) → Future; future.result(timeout) ждёт и возвращает значение или бросает исключение задачи.
map vs as_completed: pool.map возвращает результаты в порядке аргументов и бросает исключение при итерации на упавшем элементе; as_completed — по мере готовности, удобно для обработки ошибок по одной.
Потоки или процессы:
| ThreadPoolExecutor | ProcessPoolExecutor | |
|---|---|---|
| подходит | ввод-вывод | вычисления |
| GIL | общий | свой в каждом процессе |
| накладные расходы | малые | запуск процессов, pickle |
| общие данные | есть (нужны блокировки) | нет |
Подводные камни процессов: функции должны быть определены на уровне модуля (pickle), запуск под if __name__ == "__main__": (метод spawn на Windows и macOS), большие аргументы дорого сериализовать.
Пример
from concurrent.futures import ThreadPoolExecutor, as_completed
with ThreadPoolExecutor(max_workers=20) as pool:
futures = {pool.submit(download, url): url for url in urls}
for f in as_completed(futures):
try:
save(futures[f], f.result())
except Exception as e:
log.warning("%s: %s", futures[f], e)Что хотят услышать
- Интерфейс submit, map, as_completed, Future.
- Выбор между потоками и процессами.
- Ограничения pickle.
Проверьте себя
Один верный вариант. Подумайте, потом откройте ответ.
Показать варианты и ответ
- ✗Исключение в задаче теряется бесследно, если не обернуть функцию в try: result() вернёт None
- ✗ThreadPoolExecutor ускоряет вычисления на всех ядрах благодаря GIL
- ✓Потоки — для ввода-вывода, процессы — для вычислений; submit возвращает Future, ошибка всплывает в result()
- ✗В ProcessPoolExecutor можно передать лямбду без ограничений
Разбор темы
- Потоки, процессы и GIL
Что такое GIL и что он не защищает, когда выбирать потоки, процессы или asyncio, concurrent.futures, ограничения multiprocessing, сборка без GIL.
Связанные вопросы
Разбор собеседования с sotto
После собеседования sotto разбирает разговор: оценка, каждый вопрос — хорошо, частично или слабо, чего не хватило в ответе и над чем поработать к следующему разу. Первые 30 минут бесплатно.