Как можно контролировать количество одновременно выполняющихся потоков?

Для контроля количества одновременно выполняющихся потоков в Python существует несколько подходов, каждый из которых имеет свои преимущества и недостатки.

1. Использование threading.Semaphore:

threading.Semaphore представляет собой классический семафор, который позволяет ограничить количество потоков, имеющих доступ к общему ресурсу (в данном случае, к выполнению задачи). Семафор инициализируется с максимальным количеством разрешенных потоков. Каждый поток, прежде чем начать выполнение, вызывает метод acquire(), который блокирует поток, если счетчик семафора равен 0. После завершения работы поток вызывает метод release(), который увеличивает счетчик, позволяя другому потоку получить доступ.

import threading

max_threads = 5
semaphore = threading.Semaphore(max_threads)

def worker(task_id):
    semaphore.acquire()
    try:
        print(f"Thread {task_id} started.")
        # Выполнение задачи
        import time
        time.sleep(2) # Имитация работы
        print(f"Thread {task_id} finished.")
    finally:
        semaphore.release()

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

for t in threads:
    t.join() # Ожидание завершения всех потоков
  

2. Использование concurrent.futures.ThreadPoolExecutor:

ThreadPoolExecutor предоставляет более удобный и современный способ управления потоками. Он автоматически управляет пулом потоков и позволяет отправлять задачи на выполнение. При инициализации ThreadPoolExecutor указывается максимальное количество потоков, которое он может использовать. Задачи отправляются с помощью методов submit() или map(). submit() возвращает объект Future, который позволяет отслеживать состояние задачи и получать ее результат. map() применяет функцию ко всем элементам итерируемого объекта и возвращает итератор с результатами.

import concurrent.futures

max_threads = 5

def worker(task_id):
    print(f"Thread {task_id} started.")
    # Выполнение задачи
    import time
    time.sleep(2) # Имитация работы
    print(f"Thread {task_id} finished.")
    return f"Result from thread {task_id}"

with concurrent.futures.ThreadPoolExecutor(max_workers=max_threads) as executor:
    futures = [executor.submit(worker, i) for i in range(10)]

    for future in concurrent.futures.as_completed(futures):
        try:
            result = future.result()
            print(f"Result: {result}")
        except Exception as e:
            print(f"Exception: {e}")
  

3. Комбинация Queue и threading:

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

import threading
import queue
import time

max_threads = 5
task_queue = queue.Queue()

def worker():
    while True:
        task_id = task_queue.get()
        if task_id is None: # Сигнал завершения потока
            break
        print(f"Thread {threading.current_thread().name} started task {task_id}.")
        # Выполнение задачи
        time.sleep(2) # Имитация работы
        print(f"Thread {threading.current_thread().name} finished task {task_id}.")
        task_queue.task_done() # Уведомляем очередь о завершении задачи

threads = []
for i in range(max_threads):
    t = threading.Thread(target=worker, name=f"Worker-{i}")
    threads.append(t)
    t.start()

# Добавляем задачи в очередь
for i in range(10):
    task_queue.put(i)

# Ждем завершения всех задач
task_queue.join()

# Сигнализируем потокам о завершении работы
for i in range(max_threads):
    task_queue.put(None)

# Ожидаем завершения всех потоков
for t in threads:
    t.join()
  

Выбор подхода:

• threading.Semaphore: Подходит для простых случаев, когда требуется ограничить доступ к ресурсу, но требует явного управления acquire/release.
• concurrent.futures.ThreadPoolExecutor: Самый удобный и рекомендуемый подход для большинства случаев. Автоматически управляет пулом потоков и предоставляет удобные инструменты для отправки задач и получения результатов.
• Queue и threading: Подходит для более сложных сценариев, когда требуется гибкое управление задачами и возможность добавления новых задач во время выполнения.

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