Как передать данные между потоками?

Передача данных между потоками в Python

Основные способы

В Python существует несколько механизмов для безопасного обмена данными между потоками. Рассмотрим каждый из них.

1. Queue (очередь) — самый безопасный способ

Модуль queue.Queue предоставляет потокобезопасную очередь — лучший выбор для синхронизации потоков:

import queue
import threading
import time

# Создаем потокобезопасную очередь
q = queue.Queue(maxsize=10)

def producer():
    for i in range(5):
        time.sleep(0.5)
        data = f"Данные {i}"
        q.put(data)  # Отправляем данные
        print(f"Отправлено: {data}")

def consumer():
    while True:
        try:
            data = q.get(timeout=2)  # Получаем с таймаутом
            print(f"Получено: {data}")
            q.task_done()  # Сигнализируем об окончании обработки
        except queue.Empty:
            print("Очередь пуста")
            break

t1 = threading.Thread(target=producer)
t2 = threading.Thread(target=consumer)

t1.start()
t2.start()

q.join()  # Ждем, пока все задачи обработаны
t1.join()
t2.join()

2. Lock и RLock — для защиты общих ресурсов

Используем блокировки для синхронизации доступа к общим переменным:

import threading

class SharedData:
    def __init__(self):
        self.value = 0
        self.lock = threading.Lock()
    
    def increment(self):
        with self.lock:  # Экстьм доступ к value
            temp = self.value
            temp += 1
            self.value = temp

shared = SharedData()

def worker():
    for _ in range(1000):
        shared.increment()

threads = [threading.Thread(target=worker) for _ in range(5)]
for t in threads:
    t.start()
for t in threads:
    t.join()

print(f"Финальное значение: {shared.value}")  # 5000

3. Condition — сигнализация между потоками

Используем условные переменные для уведомления потоков о событиях:

import threading
import time

condition = threading.Condition()
data = None

def producer():
    global data
    for i in range(3):
        time.sleep(1)
        with condition:
            data = f"Данные {i}"
            print(f"Произведено: {data}")
            condition.notify_all()  # Уведомляем потребителей

def consumer(name):
    global data
    while True:
        with condition:
            condition.wait()  # Ожидаем уведомления
            if data:
                print(f"Потребитель {name} получил: {data}")

t1 = threading.Thread(target=producer)
t2 = threading.Thread(target=consumer, args=("A",))
t3 = threading.Thread(target=consumer, args=("B",))

t1.start()
t2.start()
t3.start()

4. Event — простая сигнализация

Эвенты используются для простых булевых сигналов между потоками:

import threading
import time

event = threading.Event()

def waiter():
    print("Поток ждет события...")
    event.wait()  # Блокируется до set()
    print("Событие произошло!")

def setter():
    time.sleep(2)
    event.set()  # Устанавливаем флаг

t1 = threading.Thread(target=waiter)
t2 = threading.Thread(target=setter)

t1.start()
t2.start()

t1.join()
t2.join()

5. Semaphore — ограничение доступа

Семафор контролирует количество потоков, одновременно обращающихся к ресурсу:

import threading
import time

# Максимум 2 потока могут одновременно работать
sem = threading.Semaphore(2)

def worker(name):
    with sem:
        print(f"Поток {name} работает")
        time.sleep(1)
        print(f"Поток {name} закончил")

threads = [threading.Thread(target=worker, args=(i,)) for i in range(5)]
for t in threads:
    t.start()
for t in threads:
    t.join()

Рекомендации по выбору

• Queue — предпочитаемый способ для обмена данными между потоками
• Lock — для защиты критических секций кода
• Event — для простых сигналов готовности
• Condition — для сложной синхронизации с уведомлениями
• Semaphore — для контроля количества одновременных обращений

Основной принцип: избегайте гонки данных, используя синхронизацию. Queue часто безопаснее, чем прямой доступ к общим переменным с Lock.