Многопроцессорная обработка в Python: варианты и рекомендации

Расширенные примеры работы с процессами

Применение Lock для синхронизации конкурентного доступа к общему ресурсу

Использование блокировок предотвращает одновременное изменение данных. Пример с многопроцессным инкрементом счётчика.

import multiprocessing as mp

def increment(counter, lock):
    for _ in range(1000):
        with lock:
            counter.value += 1

if __name__ == '__main__':
    counter = mp.Value('i', 0)
    lock = mp.Lock()
    processes = [mp.Process(target=increment, args=(counter, lock)) for _ in range(4)]
    for p in processes:
        p.start()
    for p in processes:
        p.join()
    print(f'Итоговое значение: {counter.value}')

Итоговое значение: 4000

Пояснение: mp.Value создаёт разделяемую переменную. Lock гарантирует атомарность операции прибаления. Без блокировки значение было бы меньше 4000 из-за гонок.

Пул процессов с разными функциями и ошибкой сериализации

Лямбда-функции не сериализуются pickle, что приводит к AttributeError. Пример исправления через глобальную функцию.

def bad_lambda():
    return 1

if __name__ == '__main__':
    import multiprocessing as mp
    with mp.Pool(2) as pool:
        try:
            result = pool.apply_async(lambda: 42).get()
        except AttributeError as e:
            print(f'Ошибка: {e}')
    # Исправление
    with mp.Pool(2) as pool:
        result = pool.apply_async(bad_lambda).get()
        print(f'Результат: {result}')

Ошибка: Can't pickle  at 0x...>
Результат: 1

Пояснение: лямбды не могут быть переданы в другой процесс, так как они не имеют глобального имени. Всегда определяйте функции на верхнем уровне модуля.

Запуск демонического процесса, завершающегося вместе с родителем

Установка флага daemon=True заставляет процесс завершаться при завершении родительского. Полезно для фоновых служб.

import multiprocessing as mp
import time

def background_task():
    while True:
        print('Фоновый процесс жив')
        time.sleep(1)

if __name__ == '__main__':
    p = mp.Process(target=background_task, daemon=True)
    p.start()
    time.sleep(2)
    print('Родитель завершается. Демон будет убит.')

Фоновый процесс жив
Фоновый процесс жив
Родитель завершается. Демон будет убит.

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

Обмен большими объёмами данных через Queue с Batch-обработкой

Использование очереди для передачи чанков данных и сигналов завершения.

import multiprocessing as mp
import numpy as np

def processor(in_q, out_q):
    while True:
        chunk = in_q.get()
        if chunk is None:
            break
        result = np.mean(chunk)
        out_q.put(result)

if __name__ == '__main__':
    data = [np.random.rand(1000) for _ in range(5)]
    in_q = mp.Queue()
    out_q = mp.Queue()
    p = mp.Process(target=processor, args=(in_q, out_q))
    p.start()
    for chunk in data:
        in_q.put(chunk)
    in_q.put(None)
    p.join()
    results = []
    while not out_q.empty():
        results.append(out_q.get())
    print(f'Средние: {results}')

Средние: [0.493..., 0.512..., ...]

Пояснение: numpy массивы не сериализуются напрямую, но они поддерживают буферный протокол. Queue автоматически упаковывает их в bytes. Для очень больших массивов лучше использовать shared_memory.

Использование Semaphore для ограничения одновременного доступа к ресурсу

Семафор разрешает фиксированное число процессов одновременно выполнять определённый код.

import multiprocessing as mp
import time

def limited_access(sem, proc_id):
    with sem:
        print(f'Процесс {proc_id} начал работу')
        time.sleep(1)
        print(f'Процесс {proc_id} завершил')

if __name__ == '__main__':
    sem = mp.Semaphore(2)
    procs = [mp.Process(target=limited_access, args=(sem, i)) for i in range(4)]
    for p in procs:
        p.start()
    for p in procs:
        p.join()

Процесс 0 начал работу
Процесс 1 начал работу
Процесс 0 завершил
Процесс 1 завершил
Процесс 2 начал работу
Процесс 3 начал работу
Процесс 2 завершил
Процесс 3 завершил

Пояснение: семафор счётчиком 2 позволяет только двум процессам одновременно войти в критическую секцию. Остальные ждут освобождения.

Комбинация asyncio и multiprocessing для асинхронного ввода-вывода и параллельных вычислений

С помощью loop.run_in_executor с ProcessPoolExecutor можно выполнять CPU-интенсивные задачи, не блокируя цикл событий.

import asyncio
from concurrent.futures import ProcessPoolExecutor
import time

def heavy_calc(n):
    return sum(i**2 for i in range(n))

async def main():
    loop = asyncio.get_event_loop()
    with ProcessPoolExecutor() as pool:
        tasks = [loop.run_in_executor(pool, heavy_calc, 10_000_000) for _ in range(3)]
        results = await asyncio.gather(*tasks)
    print(results)

if __name__ == '__main__':
    asyncio.run(main())

[333283335000000, 333283335000000, 333283335000000]

Пояснение: asyncio управляет I/O, а процессы обрабатывают вычисления. Не забывайте, что количество процессов не должно превышать число ядер для оптимальной производительности.