Запуск ограниченного количества потоков одновременно

Question

Всех приветствую. Задача такая:

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

Я решил эту задачу так:

import time
import threading


class TheLongTask:
    def __init__(self, name, seconds):
        self.name = name
        self._seconds = seconds

    def run(self, callback):
        print(f'Task with name {self.name} started')
        time.sleep(self._seconds+2)
        print(f'Task with name {self.name} ended\n')
        callback()


class TaskRunner:
    def __init__(self):
        self._tasks = []
        self._runned_task = 0

    def _start_tasks(self):
        for i in range(0, len(self._tasks)):
            if self._runned_task <= 1:
                task = self._tasks.pop(i)
                threading.Thread(target=task.run,
                                 args=(self._on_task_complete, )).start()
                self._runned_task += 1
            else:
                #print("2 threads already running. Waiting for end")
                break

    def _on_task_complete(self):
        self._runned_task -= 1
        self._start_tasks()

    def add_task(self, task: TheLongTask):
        self._tasks.append(task)
        self._start_tasks()


if __name__ == '__main__':
    runner = TaskRunner()

    for i in range(0, 100):
        runner.add_task(TheLongTask(f'task number {i}', i))

Но я более чем уверен, что есть более изящные методы решения.

Как можно это сделать красиво?)

Answer 1

Для подобной задачи, элегантнее всего было бы использовать Semaphore() это абстрактный ограничитель одновременно выполняемых задач.

import threading, random, time

class ActivePool:
"""Воображаемый пул соединений"""

    start = time.time()

    def __init__(self):
        super(ActivePool, self).__init__()
        self.active = []
        self.lock = threading.Lock()

    def makeActive(self, name):
        with self.lock:
            self.active.append(name)
            tm = time.time() - self.start
            print(f'Время: {round(tm, 3)} Running: {self.active}')

    def makeInactive(self, name):
        with self.lock:
            self.active.remove(name)
            tm = time.time() - self.start
            print(f'Время: {round(tm, 3)} Running: {self.active}')


def worker(sem, pool):
    with sem:
        th_name = threading.current_thread().name
        print(f'{th_name} ожидает присоединения к пулу')
        pool.makeActive(th_name)
        time.sleep(0.5)
        pool.makeInactive(th_name)

# переменная семафора    
sem = threading.Semaphore(2)


# воображаемый пул соединений
pool = ActivePool()
# запускаем потоки
for i in range(4):
    t = threading.Thread(
        target=worker,
        args=(sem, pool),
    )
    t.start()

Подробнее можете изучить тут -> : Источник

Answer 2

Более изящное решение это imap_unordered метод на multiprocessing.dummy.Pool - создаст группу тредов и корми этих воркеров последовательность, забирай результат по мере выполнения.

Как альтернатива thread_pool_executor из модуля concurent

Но если слова 'сам манипулировать потоками и прочим' значит что нельзя использовать стандартные модули кроме тех что проходили, и надо написать индусский код, то стоит посмотреть на queue - во времена python 2.5 делал взаимодействие с треда и через него

БЛОГ НА HUSL

Запуск ограниченного количества потоков одновременно

Ответы (2 шт):