Способы обмена данными между процессами Python

Создано: 17 февраля 2025Автор: Егор Астапов72 просмотраСложность: легкий

Существуют способы обмена данными между процессами в Python. Мне задали этот вопрос на собеседовании и я не смог правильно ответить. Я провалился. Сами способы:

Использование модуля multiprocessing

Модуль multiprocessing предоставляет разные методы для обмена данными между процессами, такие как очереди Queue, каналы Pipe и разделяемая память.

import multiprocessing

def worker(queue):
    # Отправляем данные обратно в основной процесс
    queue.put("Hello from worker!")

if __name__ == "__main__":
    queue = multiprocessing.Queue()
    
    process = multiprocessing.Process(target=worker, args=(queue,))
    process.start()
    
    # Получаем данные из очереди
    message = queue.get()
    print(message)
    
    process.join()

Использование Pipe

При помощи Pipe можно установить двустороннюю связь между двумя процессами.

import multiprocessing

def worker(conn):
    conn.send("Hello from worker!")
    conn.close()

if __name__ == "__main__":
    parent_conn, child_conn = multiprocessing.Pipe()
    
    process = multiprocessing.Process(target=worker, args=(child_conn,))
    process.start()
    
    message = parent_conn.recv()  # Получаем сообщение
    print(message)
    
    process.join()

Использование разделяемой памяти

Если данные, которые вы хотите обменять, достаточно просты (например, числа или массивы), вы можете использовать разделяемую память через Value или Array.

import multiprocessing

def worker(shared_value):
    shared_value.value = 42  # Записываем значение в разделяемую память

if __name__ == "__main__":
    shared_value = multiprocessing.Value('i', 0)  # 'i' - тип для целого числа
    
    process = multiprocessing.Process(target=worker, args=(shared_value,))
    process.start()
    
    process.join()
    print(shared_value.value)  # Выводим значение из разделяемой памяти

Эти методы позволяют осуществлять обмен данными между процессами в Python. Выбирайте подходящий метод в зависимости от ваших нужд и сложности обмена данными.

Теперь вернёмся к решению с применением очереди. Я создам 2 процесса и эти процессы будут обмениваться данными. Чтобы создать два работающих процесса, которые обмениваются данными через очередь Queue, можно использовать модуль multiprocessing в Python. В этом примере один процесс будет отправлять данные в очередь, а другой — извлекать их.

import multiprocessing
import time

def producer(queue):
    for i in range(5):
        item = f'item {i}'
        queue.put(item)
        print(f'Produced: {item}')
        time.sleep(1)  # имитируем задержку

def consumer(queue):
    for _ in range(5):
        item = queue.get()
        print(f'Consumed: {item}')
        time.sleep(2)  # имитируем задержку

if __name__ == '__main__':
    # Создаем очередь
    queue = multiprocessing.Queue()

    # Создаем процессы
    producer_process = multiprocessing.Process(target=producer, args=(queue,))
    consumer_process = multiprocessing.Process(target=consumer, args=(queue,))

    # Запускаем процессы
    producer_process.start()
    consumer_process.start()

    # Ждем завершения процессов
    producer_process.join()
    consumer_process.join()

    print('All processes completed.')
    """
    Produced: item 0
    Consumed: item 0
    Produced: item 1
    Produced: item 2
    Consumed: item 1
    Produced: item 3
    Produced: item 4
    Consumed: item 2
    Consumed: item 3
    Consumed: item 4
    All processes completed.
    """

Импортируем модули: Импортируем необходимые классы и функции из модуля multiprocessing и time.
Функция producer: Эта функция создает 5 элементов и помещает их в очередь с помощью метода put(). Между добавлением каждого элемента происходит задержка в 1 секунду.
Функция consumer: Эта функция извлекает 5 элементов из очереди с помощью метода get(). Между потреблением каждого элемента происходит задержка в 2 секунды./li>
Главная часть программы:
1. Создаем очередь Queue()
2. Создаем два процесса: один для производителя, другой для потребителя
3. Запускаем оба процесса
4. Дожидаемся завершения обоих процессов с помощью метода join()

Скопируйте и запустите этот код в вашем Python-окружении, и вы увидите, как один процесс производит элементы, а другой их потребляет.

Теперь вернёмся к решению с применением Pipe. Я снова создам 2 процесса и эти процессы будут обмениваться данными. Для передачи данных между двумя процессами в Python можно использовать модуль multiprocessing, который предоставляет класс Pipe. Ниже приведен пример, который создаёт два процесса и передаёт данные от одного процесса к другому с помощью канала Pipe

import multiprocessing
import time

def sender(conn):
    for i in range(5):
        time.sleep(1)  # Завершаем выполнение на 1 секунду
        conn.send(f"Сообщение {i}")  # Отправляем сообщение
        print(f"[Отправитель] Отправлено: Сообщение {i}")
    conn.close()  # Закрываем соединение после отправки сообщений

def receiver(conn):
    while True:
        try:
            msg = conn.recv()  # Получаем сообщение
            print(f"[Получатель] Получено: {msg}")  # Печатаем полученное сообщение
            if msg == 'Сообщение 4':
                print("stop")
                break
        except EOFError:  # Выходим, если соединение закрыто
            break

if __name__ == "__main__":
    # Создаём Pipe
    parent_conn, child_conn = multiprocessing.Pipe()

    # Создаём процессы
    process_sender = multiprocessing.Process(target=sender, args=(child_conn,))
    process_receiver = multiprocessing.Process(target=receiver, args=(parent_conn,))

    # Запускаем процессы
    process_sender.start()
    process_receiver.start()

    # Ожидаем завершения процессов
    process_sender.join()
    process_receiver.join()
    
    print('All processes completed.')
    """
    [Отправитель] Отправлено: Сообщение 0
    [Получатель] Получено: Сообщение 0
    [Отправитель] Отправлено: Сообщение 1
    [Получатель] Получено: Сообщение 1
    [Отправитель] Отправлено: Сообщение 2
    [Получатель] Получено: Сообщение 2
    [Отправитель] Отправлено: Сообщение 3
    [Получатель] Получено: Сообщение 3
    [Отправитель] Отправлено: Сообщение 4
    [Получатель] Получено: Сообщение 4
    stop
    All processes completed.
    """

Мы создаём Pipe, который предоставляет две точки подключения (parent_conn и child_conn)
Функция sender отправляет сообщения через child_conn
Функция receiver получает сообщения через parent_conn и печатает их
Мы запускаем процессы и ждём их завершения с помощью join()

Запустив этот код, вы увидите, как сообщения передаются от одного процесса к другому

Теперь вернёмся к решению с применением разделяемой памяти. Я снова создам 2 процесса и эти процессы будут обмениваться данными. Для передачи данных между двумя процессами с использованием разделяемой памяти в Python можно использовать модуль `multiprocessing`. Здесь я приведу пример, который показывает, как создать два процесса и передать данные от одного к другому, используя Value или Array из модуля multiprocessing.

import multiprocessing
import time

def sender(shared_value):
    """Функция отправителя, которая записывает значение в разделяемую память"""
    for i in range(5):
        print(f"Sender: Sending value {i}")
        shared_value.value = i  # Записываем значение в разделяемую память
        time.sleep(1)  # Задержка для имитации работы

def receiver(shared_value):
    """Функция получателя, которая читает значение из разделяемой памяти"""
    for _ in range(5):
        time.sleep(1.5)  # Задержка, чтобы дать отправителю время отправить значение
        print(f"Receiver: Received value {shared_value.value}")

if __name__ == "__main__":
    # Создаем разделяемую переменную
    shared_value = multiprocessing.Value('i', 0)  # 'i' означает тип integer

    # Создаем процессы
    p1 = multiprocessing.Process(target=sender, args=(shared_value,))
    p2 = multiprocessing.Process(target=receiver, args=(shared_value,))

    # Запускаем процессы
    p1.start()
    p2.start()

    # Ждем, пока оба процесса завершатся
    p1.join()
    p2.join()

    print("Both processes have finished.")
    """
    Sender: Sending value 0
    Sender: Sending value 1
    Receiver: Received value 1
    Sender: Sending value 2
    Sender: Sending value 3
    Receiver: Received value 3
    Sender: Sending value 4
    Receiver: Received value 4
    Receiver: Received value 4
    Receiver: Received value 4
    Both processes have finished.
    """

Импортируем модули: Импортируем multiprocessing для работы с процессами и time для задержек.
Создаем функцию sender, которая будет отправлять данные: она записывает значения в разделяемую переменную shared_value.
Создаем функцию receiver, которая будет получать данные: она читает значения из shared_value.
Создаем разделяемую переменную: Используем multiprocessing.Value для создания разделяемого целочисленного значения.
Создаем и запускаем процессы: Создаем два процесса p1 и p2, передав нужные целевые функции и аргументы. Затем запускаем процессы с помощью метода start().
Ожидаем завершения процессов: С помощью метода join() мы ждем, пока оба процесса завершатся.

Запустив этот код, вы увидите, как данные передаются от одного процесса к другому через разделяемую память.

Способы обмена данными между процессами Python

Использование модуля multiprocessing

Использование Pipe

Использование разделяемой памяти

Комментарии