Как передать информацию между двумя процессами?

Как передать информацию между двумя процессами

В Python есть несколько способов передачи данных между процессами. Процессы имеют изолированную память, поэтому нужны специальные механизмы.

1. Queue (очередь) — самый популярный

import multiprocessing
from multiprocessing import Queue

def producer(queue):
    """Процесс, который отправляет данные"""
    for i in range(5):
        data = f"Message {i}"
        queue.put(data)  # Положить в очередь
        print(f"Producer отправил: {data}")

def consumer(queue):
    """Процесс, который получает данные"""
    while True:
        data = queue.get()  # Получить из очереди
        if data is None:  # Сигнал выхода
            break
        print(f"Consumer получил: {data}")

if __name__ == "__main__":
    q = Queue()
    p1 = multiprocessing.Process(target=producer, args=(q,))
    p2 = multiprocessing.Process(target=consumer, args=(q,))
    
    p1.start()
    p2.start()
    
    p1.join()
    q.put(None)  # Сигнал завершения
    p2.join()

# Результат:
# Producer отправил: Message 0
# Consumer получил: Message 0
# Producer отправил: Message 1
# Consumer получил: Message 1
# ...

2. Pipe (труба) — двусторонний канал

import multiprocessing
from multiprocessing import Pipe

def parent_process(conn):
    """Родительский процесс"""
    # Отправить данные
    conn.send("Hello from parent")
    
    # Получить ответ
    message = conn.recv()
    print(f"Parent получил: {message}")
    
    conn.close()

def child_process(conn):
    """Дочерний процесс"""
    # Получить данные
    message = conn.recv()
    print(f"Child получил: {message}")
    
    # Отправить ответ
    conn.send("Hello from child")
    
    conn.close()

if __name__ == "__main__":
    parent_conn, child_conn = Pipe()
    
    p = multiprocessing.Process(target=child_process, args=(child_conn,))
    p.start()
    
    parent_process(parent_conn)
    p.join()

# Результат:
# Child получил: Hello from parent
# Parent получил: Hello from child

3. Manager — общее состояние

Можно использовать общие структуры данных (dict, list):

import multiprocessing
from multiprocessing import Manager

def writer(shared_dict):
    """Процесс, который пишет в общий словарь"""
    shared_dict["name"] = "John"
    shared_dict["age"] = 30
    shared_dict["scores"] = [10, 20, 30]
    print("Writer записал данные")

def reader(shared_dict):
    """Процесс, который читает из общего словаря"""
    # Подождать пока данные будут записаны
    import time
    time.sleep(1)
    
    print(f"Reader прочитал: {shared_dict}")
    print(f"Name: {shared_dict['name']}")
    print(f"Scores: {shared_dict['scores']}")

if __name__ == "__main__":
    with Manager() as manager:
        shared_dict = manager.dict()
        
        p1 = multiprocessing.Process(target=writer, args=(shared_dict,))
        p2 = multiprocessing.Process(target=reader, args=(shared_dict,))
        
        p1.start()
        p2.start()
        
        p1.join()
        p2.join()

# Результат:
# Writer записал данные
# Reader прочитал: {'name': 'John', 'age': 30, 'scores': [10, 20, 30]}

4. Value и Array — для простых данных

import multiprocessing
from multiprocessing import Value, Array

def increment(counter, shared_array):
    """Увеличить счётчик и обновить массив"""
    with counter.get_lock():
        counter.value += 1
    
    for i in range(len(shared_array)):
        shared_array[i] = shared_array[i] * 2

if __name__ == "__main__":
    # Общий счётчик (используем threading.Lock для синхронизации)
    counter = Value('i', 0)  # 'i' = int
    
    # Общий массив
    shared_array = Array('d', [1.0, 2.0, 3.0])  # 'd' = double
    
    p = multiprocessing.Process(target=increment, args=(counter, shared_array))
    p.start()
    p.join()
    
    print(f"Counter: {counter.value}")        # 1
    print(f"Array: {list(shared_array)}")     # [2.0, 4.0, 6.0]

5. Socket (сокет) — для удалённых процессов

Если процессы на разных машинах:

import socket
import json

def server():
    """Сервер"""
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.bind(("localhost", 5000))
    sock.listen(1)
    
    conn, addr = sock.accept()
    print(f"Подключение от {addr}")
    
    # Получить данные
    data = conn.recv(1024).decode()
    print(f"Получено: {data}")
    
    # Отправить ответ
    response = json.dumps({"status": "ok"})
    conn.send(response.encode())
    
    conn.close()
    sock.close()

def client():
    """Клиент"""
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.connect(("localhost", 5000))
    
    # Отправить данные
    message = json.dumps({"message": "Hello"})
    sock.send(message.encode())
    
    # Получить ответ
    response = sock.recv(1024).decode()
    print(f"Ответ: {response}")
    
    sock.close()

6. Pool — результаты из рабочих процессов

import multiprocessing
from multiprocessing import Pool

def square(x):
    """Вычислить квадрат"""
    return x * x

if __name__ == "__main__":
    with Pool(4) as pool:
        # Отправить задачи в рабочие процессы
        results = pool.map(square, [1, 2, 3, 4, 5])
        print(results)  # [1, 4, 9, 16, 25]
        
        # Или использовать apply_async для асинхронного выполнения
        async_result = pool.apply_async(square, (10,))
        result = async_result.get()  # Дождаться результата
        print(result)  # 100

Сравнение методов

Метод         | Двусторонний | Синхронизация | Использование
─────────────────────────────────────────────────────────────
Queue         | Нет          | Нет           | Передача данных
Pipe          | Да           | Нет           | Диалог процессов
Manager       | Да           | Внутри        | Общие структуры
Value/Array   | Нет          | Да (lock)     | Простые данные
Socket        | Да           | Нет           | Удалённые процессы
Pool          | Нет          | Нет           | Параллелизм

Синхронизация между процессами

import multiprocessing
from multiprocessing import Lock, Semaphore

def worker(lock, counter_dict):
    """Безопасно работать со счётчиком"""
    with lock:
        counter_dict["count"] += 1
        print(f"Count: {counter_dict['count']}")

if __name__ == "__main__":
    with multiprocessing.Manager() as manager:
        lock = Lock()  # Мьютекс
        counter = manager.dict({"count": 0})
        
        processes = []
        for _ in range(5):
            p = multiprocessing.Process(target=worker, args=(lock, counter))
            processes.append(p)
            p.start()
        
        for p in processes:
            p.join()
        
        print(f"Финальное значение: {counter['count']}")  # 5

Рекомендации

1. Queue — для передачи данных между процессами
2. Pipe — для двусторонней коммуникации
3. Manager.dict() — для общих структур данных
4. Pool — для параллельных задач
5. Socket — для удалённых процессов

Итог

Для передачи информации между процессами:

• Queue — простая и надёжная
• Pipe — для диалога
• Manager — для сложных структур
• Socket — для сетевого взаимодействия
• Всегда используй синхронизацию (locks) при совместном доступе к данным