Конкурентное выполнение: какой модуль выбрать для потоков

Расширенные примеры работы с многопоточностью

Синхронизация потоков с помощью Event

import threading
import time

def waiter(event):
    print("Ожидание события...")
    event.wait()
    print("Событие получено, продолжаем")

def setter(event):
    time.sleep(1)
    print("Устанавливаем событие")
    event.set()

event = threading.Event()
thread1 = threading.Thread(target=waiter, args=(event,))
thread2 = threading.Thread(target=setter, args=(event,))

thread1.start()
thread2.start()
thread1.join()
thread2.join()
print("Готово")

Ожидание события...
Устанавливаем событие
Событие получено, продолжаем
Готово

Безопасная передача данных через Queue

from queue import Queue
import threading
import time

def producer(queue):
    for i in range(5):
        item = f"Сообщение {i}"
        queue.put(item)
        print(f"Отправлено: {item}")
        time.sleep(0.2)

def consumer(queue):
    while True:
        item = queue.get()
        if item is None:  # сигнал остановки
            break
        print(f"Получено: {item}")
        queue.task_done()

q = Queue()
prod = threading.Thread(target=producer, args=(q,))
cons = threading.Thread(target=consumer, args=(q,))

prod.start()
cons.start()

prod.join()
q.put(None)  # сигнал завершения потребителя
cons.join()
print("Все задачи завершены")

Отправлено: Сообщение 0
Получено: Сообщение 0
Отправлено: Сообщение 1
Получено: Сообщение 1
Отправлено: Сообщение 2
Получено: Сообщение 2
Отправлено: Сообщение 3
Получено: Сообщение 3
Отправлено: Сообщение 4
Получено: Сообщение 4
Все задачи завершены

Пул потоков с обработкой исключений и таймаутом

from concurrent.futures import ThreadPoolExecutor, TimeoutError
import time

def risky_task(seconds):
    time.sleep(seconds)
    if seconds > 2:
        raise ValueError("Слишком долго!")
    return f"Успех за {seconds} сек"

with ThreadPoolExecutor(max_workers=3) as executor:
    futures = [executor.submit(risky_task, t) for t in [1, 3, 0.5]]
    for future in futures:
        try:
            result = future.result(timeout=2.5)
            print(result)
        except TimeoutError:
            print("Таймаут")
        except ValueError as e:
            print(f"Ошибка: {e}")

Успех за 1 сек
Ошибка: Слишком долго!
Успех за 0.5 сек

Подкласс Thread с переопределением run

import threading
import random

class MyThread(threading.Thread):
    def __init__(self, name):
        super().__init__()
        self.name = name

    def run(self):
        delay = random.uniform(0.5, 1.5)
        print(f"{self.name} засыпает на {delay:.2f} сек")
        threading.Event().wait(delay)
        print(f"{self.name} проснулся")

threads = [MyThread(f"Поток-{i}") for i in range(3)]
for t in threads:
    t.start()
for t in threads:
    t.join()
print("Все потоки завершены")

Поток-0 засыпает на 0.87 сек
Поток-1 засыпает на 1.23 сек
Поток-2 засыпает на 0.95 сек
Поток-0 проснулся
Поток-2 проснулся
Поток-1 проснулся
Все потоки завершены

Использование Condition для синхронизации производителей и потребителей

import threading
import time

class Buffer:
    def __init__(self, capacity):
        self.capacity = capacity
        self.items = []
        self.cond = threading.Condition()

    def produce(self, item):
        with self.cond:
            while len(self.items) >= self.capacity:
                self.cond.wait()
            self.items.append(item)
            self.cond.notify()

    def consume(self):
        with self.cond:
            while not self.items:
                self.cond.wait()
            item = self.items.pop(0)
            self.cond.notify()
            return item

def producer(buf):
    for i in range(5):
        buf.produce(i)
        print(f"Произведено {i}")
        time.sleep(0.1)

def consumer(buf):
    for _ in range(5):
        item = buf.consume()
        print(f"Потреблено {item}")
        time.sleep(0.15)

buf = Buffer(3)
t1 = threading.Thread(target=producer, args=(buf,))
t2 = threading.Thread(target=consumer, args=(buf,))
t1.start()
t2.start()
t1.join()
t2.join()
print("Работа завершена")

Произведено 0
Потреблено 0
Произведено 1
Произведено 2
Потреблено 1
Произведено 3
Потреблено 2
Произведено 4
Потреблено 3
Потреблено 4
Работа завершена