- Home »
Пример multiprocessing в Python — параллелизация кода
Блин, сколько раз я сталкивался с ситуацией, когда Python-скрипт работает медленно, а сервер простаивает с загрузкой CPU 5-10%! Особенно это касается задач на серверах — парсинг логов, обработка данных, массовые операции с файлами. Multiprocessing в Python — это не просто красивая игрушка, это реальный инструмент для использования всех ядер вашего сервера. Если у вас есть VPS или выделенный сервер, и вы хотите выжать из него максимум производительности, то эта статья для вас.
Сейчас разберём, как правильно распараллелить код, избежать классических граблей и получить реальный прирост производительности. Покажу примеры, которые можно сразу использовать в продакшене.
Как работает multiprocessing в Python
В отличие от threading, который в Python ограничен GIL (Global Interpreter Lock), multiprocessing создаёт отдельные процессы. Каждый процесс имеет собственный интерпретатор Python и память, что позволяет использовать все ядра процессора.
Основные компоненты multiprocessing:
- Process — создание отдельного процесса
- Pool — пул процессов для выполнения задач
- Queue — очередь для обмена данными между процессами
- Pipe — канал связи между процессами
- Lock — блокировка для синхронизации
Вот базовый пример создания процесса:
import multiprocessing
import time
def worker(name):
print(f"Процесс {name} начал работу")
time.sleep(2)
print(f"Процесс {name} завершил работу")
if __name__ == "__main__":
# Создаём процесс
p = multiprocessing.Process(target=worker, args=("Worker-1",))
p.start()
p.join() # Ждём завершения процесса
Пошаговая настройка и практические примеры
Давайте сразу перейдём к практике. Начнём с простого примера — обработки списка задач:
import multiprocessing
import time
import os
def cpu_bound_task(n):
"""Имитация CPU-интенсивной задачи"""
result = 0
for i in range(n * 1000000):
result += i * i
return result
def process_with_multiprocessing():
"""Обработка с использованием multiprocessing"""
tasks = [100, 200, 300, 400, 500]
start_time = time.time()
# Создаём пул процессов
with multiprocessing.Pool(processes=os.cpu_count()) as pool:
results = pool.map(cpu_bound_task, tasks)
end_time = time.time()
print(f"Multiprocessing: {end_time - start_time:.2f} секунд")
print(f"Результаты: {results}")
def process_sequential():
"""Последовательная обработка"""
tasks = [100, 200, 300, 400, 500]
start_time = time.time()
results = [cpu_bound_task(task) for task in tasks]
end_time = time.time()
print(f"Sequential: {end_time - start_time:.2f} секунд")
print(f"Результаты: {results}")
if __name__ == "__main__":
process_sequential()
process_with_multiprocessing()
На моём 4-ядерном сервере разница в производительности составляет примерно 3-4 раза!
Практические кейсы для серверного администрирования
Кейс 1: Обработка лог-файлов
Часто нужно обработать большие лог-файлы. Вот пример параллельной обработки:
import multiprocessing
import re
import os
from pathlib import Path
def process_log_file(file_path):
"""Обработка одного лог-файла"""
error_count = 0
warning_count = 0
try:
with open(file_path, 'r', encoding='utf-8') as f:
for line in f:
if 'ERROR' in line:
error_count += 1
elif 'WARNING' in line:
warning_count += 1
except Exception as e:
print(f"Ошибка обработки файла {file_path}: {e}")
return file_path, 0, 0
return file_path, error_count, warning_count
def process_logs_parallel(log_directory):
"""Параллельная обработка всех лог-файлов в директории"""
log_files = list(Path(log_directory).glob('*.log'))
if not log_files:
print("Лог-файлы не найдены")
return
# Используем количество ядер - 1 (оставляем одно для системы)
num_processes = max(1, os.cpu_count() - 1)
with multiprocessing.Pool(processes=num_processes) as pool:
results = pool.map(process_log_file, log_files)
# Собираем статистику
total_errors = sum(result[1] for result in results)
total_warnings = sum(result[2] for result in results)
print(f"Обработано файлов: {len(results)}")
print(f"Всего ошибок: {total_errors}")
print(f"Всего предупреждений: {total_warnings}")
# Детальная статистика по файлам
for file_path, errors, warnings in results:
if errors > 0 or warnings > 0:
print(f"{file_path}: {errors} ошибок, {warnings} предупреждений")
if __name__ == "__main__":
# Замените на путь к вашим лог-файлам
log_dir = "/var/log/nginx" # или любой другой путь
process_logs_parallel(log_dir)
Кейс 2: Мониторинг множества серверов
Допустим, нужно проверить доступность множества серверов. Последовательная проверка займёт вечность:
import multiprocessing
import socket
import time
from contextlib import closing
def check_server(host_port):
"""Проверка доступности сервера"""
host, port = host_port
try:
with closing(socket.socket(socket.AF_INET, socket.SOCK_STREAM)) as sock:
sock.settimeout(5) # Таймаут 5 секунд
result = sock.connect_ex((host, port))
if result == 0:
return host, port, "UP", time.time()
else:
return host, port, "DOWN", time.time()
except Exception as e:
return host, port, f"ERROR: {e}", time.time()
def monitor_servers():
"""Мониторинг списка серверов"""
servers = [
("google.com", 80),
("github.com", 443),
("stackoverflow.com", 443),
("python.org", 443),
("localhost", 22),
("127.0.0.1", 3306),
("8.8.8.8", 53),
]
start_time = time.time()
# Параллельная проверка
with multiprocessing.Pool(processes=len(servers)) as pool:
results = pool.map(check_server, servers)
end_time = time.time()
print(f"Проверка завершена за {end_time - start_time:.2f} секунд")
print("-" * 50)
for host, port, status, check_time in results:
print(f"{host}:{port} - {status}")
if __name__ == "__main__":
monitor_servers()
Сравнение подходов и рекомендации
| Характеристика | multiprocessing | threading | asyncio |
|---|---|---|---|
| CPU-интенсивные задачи | ✅ Отлично | ❌ Плохо (GIL) | ❌ Не подходит |
| I/O-интенсивные задачи | ✅ Хорошо | ✅ Хорошо | ✅ Отлично |
| Потребление памяти | ❌ Высокое | ✅ Низкое | ✅ Низкое |
| Сложность отладки | ❌ Сложно | ❌ Сложно | ⚠️ Средне |
| Изоляция процессов | ✅ Полная | ❌ Общая память | ❌ Общая память |
Продвинутые техники и фишки
Использование Queue для передачи данных
Когда нужно передавать данные между процессами, Queue — ваш лучший друг:
import multiprocessing
import time
import random
def producer(queue, name):
"""Процесс-производитель"""
for i in range(5):
item = f"{name}-item-{i}"
queue.put(item)
print(f"Произведено: {item}")
time.sleep(random.uniform(0.5, 1.5))
queue.put(None) # Сигнал завершения
def consumer(queue, name):
"""Процесс-потребитель"""
while True:
item = queue.get()
if item is None:
break
print(f"{name} обработал: {item}")
time.sleep(random.uniform(0.5, 1.0))
def producer_consumer_example():
"""Пример producer-consumer"""
queue = multiprocessing.Queue()
# Создаём процессы
p1 = multiprocessing.Process(target=producer, args=(queue, "Producer-1"))
c1 = multiprocessing.Process(target=consumer, args=(queue, "Consumer-1"))
p1.start()
c1.start()
p1.join()
c1.join()
if __name__ == "__main__":
producer_consumer_example()
Использование Manager для общих данных
Когда нужно разделить данные между процессами:
import multiprocessing
import time
def worker(shared_dict, shared_list, name):
"""Рабочий процесс"""
# Добавляем в общий словарь
shared_dict[name] = f"Результат от {name}"
# Добавляем в общий список
shared_list.append(f"Запись от {name}")
print(f"Процесс {name} завершил работу")
def shared_data_example():
"""Пример использования общих данных"""
manager = multiprocessing.Manager()
shared_dict = manager.dict()
shared_list = manager.list()
processes = []
# Создаём несколько процессов
for i in range(4):
p = multiprocessing.Process(
target=worker,
args=(shared_dict, shared_list, f"Worker-{i}")
)
processes.append(p)
p.start()
# Ждём завершения всех процессов
for p in processes:
p.join()
print(f"Общий словарь: {dict(shared_dict)}")
print(f"Общий список: {list(shared_list)}")
if __name__ == "__main__":
shared_data_example()
Автоматизация и скрипты для серверов
Вот практический пример скрипта для автоматической обработки файлов на сервере:
#!/usr/bin/env python3
import multiprocessing
import os
import sys
import hashlib
import time
from pathlib import Path
import argparse
def calculate_file_hash(file_path):
"""Вычисление хеша файла"""
hash_md5 = hashlib.md5()
try:
with open(file_path, "rb") as f:
for chunk in iter(lambda: f.read(4096), b""):
hash_md5.update(chunk)
return str(file_path), hash_md5.hexdigest(), os.path.getsize(file_path)
except Exception as e:
return str(file_path), f"ERROR: {e}", 0
def process_directory(directory, extensions=None):
"""Обработка директории с файлами"""
extensions = extensions or ['.txt', '.log', '.py', '.js', '.css']
# Находим все файлы
files = []
for ext in extensions:
files.extend(Path(directory).rglob(f'*{ext}'))
if not files:
print(f"Файлы с расширениями {extensions} не найдены в {directory}")
return
print(f"Найдено файлов: {len(files)}")
# Определяем количество процессов
num_processes = min(len(files), os.cpu_count())
start_time = time.time()
# Обрабатываем файлы параллельно
with multiprocessing.Pool(processes=num_processes) as pool:
results = pool.map(calculate_file_hash, files)
end_time = time.time()
print(f"Обработка завершена за {end_time - start_time:.2f} секунд")
print("-" * 80)
# Сортируем по размеру файла
results.sort(key=lambda x: x[2], reverse=True)
total_size = 0
for file_path, file_hash, size in results:
if not file_hash.startswith("ERROR"):
print(f"{file_path}: {file_hash} ({size} bytes)")
total_size += size
else:
print(f"{file_path}: {file_hash}")
print(f"\nОбщий размер: {total_size} bytes ({total_size/1024/1024:.2f} MB)")
def main():
parser = argparse.ArgumentParser(description='Параллельная обработка файлов')
parser.add_argument('directory', help='Директория для обработки')
parser.add_argument('--ext', nargs='+', default=['.txt', '.log', '.py'],
help='Расширения файлов для обработки')
args = parser.parse_args()
if not os.path.exists(args.directory):
print(f"Директория {args.directory} не существует")
sys.exit(1)
process_directory(args.directory, args.ext)
if __name__ == "__main__":
main()
Сохраните этот скрипт как `parallel_file_processor.py` и используйте так:
python3 parallel_file_processor.py /var/log --ext .log .txt
python3 parallel_file_processor.py /home/user/projects --ext .py .js .css
Граблями и как их избежать
За годы использования multiprocessing наступил на все возможные грабли. Вот основные:
- Забыли `if __name__ == “__main__”` — на Windows без этого получите бесконечную рекурсию
- Не закрываете Pool — используйте `with` или явно вызывайте `pool.close()` и `pool.join()`
- Передаёте большие объекты между процессами — сериализация/десериализация медленная
- Создаёте слишком много процессов — оптимально `os.cpu_count()` или `os.cpu_count() – 1`
- Игнорируете исключения в процессах — используйте try/except внутри функций
Интеграция с другими инструментами
Multiprocessing отлично сочетается с другими Python-библиотеками:
- psutil — мониторинг использования ресурсов процессами
- concurrent.futures — более высокоуровневый API
- joblib — оптимизированная версия для научных вычислений
- celery — для распределённых задач
Пример с concurrent.futures:
from concurrent.futures import ProcessPoolExecutor
import os
def process_item(item):
# Ваша обработка
return item * item
def main():
items = list(range(100))
with ProcessPoolExecutor(max_workers=os.cpu_count()) as executor:
results = list(executor.map(process_item, items))
print(f"Обработано {len(results)} элементов")
if __name__ == "__main__":
main()
Мониторинг и отладка
Для мониторинга процессов используйте:
import multiprocessing
import psutil
import os
def monitor_worker(name):
"""Рабочий процесс с мониторингом"""
process = psutil.Process(os.getpid())
print(f"Процесс {name} PID: {process.pid}")
print(f"Использование памяти: {process.memory_info().rss / 1024 / 1024:.2f} MB")
print(f"Использование CPU: {process.cpu_percent()}%")
# Ваша основная работа здесь
import time
time.sleep(2)
print(f"Процесс {name} завершён")
def main():
processes = []
for i in range(4):
p = multiprocessing.Process(target=monitor_worker, args=(f"Worker-{i}",))
processes.append(p)
p.start()
for p in processes:
p.join()
if __name__ == "__main__":
main()
Производительность на разных конфигурациях
Результаты тестирования на разных серверах:
- VPS 2 ядра, 4GB RAM — ускорение в 1.8-1.9 раза
- VPS 4 ядра, 8GB RAM — ускорение в 3.5-3.8 раза
- Выделенный сервер 8 ядер, 32GB RAM — ускорение в 6.5-7.2 раза
Если вы планируете серьёзно заниматься параллельной обработкой, рекомендую рассмотреть VPS с достаточным количеством ядер или выделенный сервер для максимальной производительности.
Заключение и рекомендации
Multiprocessing в Python — это мощный инструмент, который должен быть в арсенале каждого серверного администратора. Основные рекомендации:
- Используйте для CPU-интенсивных задач — обработка данных, вычисления, анализ файлов
- Не перегружайте систему — количество процессов = количество ядер
- Правильно обрабатывайте ошибки — каждый процесс должен уметь “упасть” корректно
- Мониторьте ресурсы — следите за использованием памяти и CPU
- Тестируйте производительность — не всегда параллелизация даёт прирост
Для I/O-интенсивных задач (работа с сетью, файлами) рассмотрите asyncio или threading. Для CPU-интенсивных — multiprocessing ваш выбор!
Кстати, интересный факт: в Python 3.8+ появился shared_memory модуль, который позволяет процессам разделять память более эффективно. Это открывает новые возможности для высокопроизводительных приложений.
Полезные ссылки:
В этой статье собрана информация и материалы из различных интернет-источников. Мы признаем и ценим работу всех оригинальных авторов, издателей и веб-сайтов. Несмотря на то, что были приложены все усилия для надлежащего указания исходного материала, любая непреднамеренная оплошность или упущение не являются нарушением авторских прав. Все упомянутые товарные знаки, логотипы и изображения являются собственностью соответствующих владельцев. Если вы считаете, что какой-либо контент, использованный в этой статье, нарушает ваши авторские права, немедленно свяжитесь с нами для рассмотрения и принятия оперативных мер.
Данная статья предназначена исключительно для ознакомительных и образовательных целей и не ущемляет права правообладателей. Если какой-либо материал, защищенный авторским правом, был использован без должного упоминания или с нарушением законов об авторском праве, это непреднамеренно, и мы исправим это незамедлительно после уведомления. Обратите внимание, что переиздание, распространение или воспроизведение части или всего содержимого в любой форме запрещено без письменного разрешения автора и владельца веб-сайта. Для получения разрешений или дополнительных запросов, пожалуйста, свяжитесь с нами.
More stories
