Комментарии пользователя Ичи Ня

Ошибка в логике многопоточного парсера?

Ичи Ня @Ichi

Как вариант можно запустить несколько экземпляров кода. Да, это больше ресурсов больше "сожрёт", процентов на 50. Но зато реально запускать столько потоков, сколько надо.

Другой вариант через callback и async всё делать. Там простая проверка завершен код или нет.

Кидаю код одного из проектов:

import threading
import time


class ThreadingController:
    _threads = {}
    __thread_max = 1
    __delay_works = []
    __start_time = 0
    __run_delay_works = False

    def __init__(self, thread_max: int = 1):
        self.set_max_thread(thread_max=thread_max)
        self.__start_time = time.monotonic()

    def add_work(self, coro):
        self.clean_threads()
        thread_id = self.find_free_thread()

        if thread_id == 0 and self.__run_delay_works:
            count = len([x for x in self._threads if self._threads[x] is not None])
            if count == 0:
                self.run_delay_works()
            else:
                return False

        if len(self._threads) >= self.__thread_max and thread_id == 0:
            return False
        if thread_id == 0:
            thread_id = len(self._threads) + 1
        print(self.__thread_max, len(self._threads), thread_id, self._threads)
        threading.Event()
        t1 = threading.Thread(target=coro, args=[thread_id], daemon=True, name=str(thread_id))
        t1.start()
        threading.Event()
        self._threads[thread_id] = t1
        # t1.join()

    def add_delay_work(self, coro, *args, **kwargs):
        for work in self.__delay_works:
            if work['coro'] == coro and work['args'] == args and work['kwargs'] == kwargs:
                return
        self.__delay_works.append({'coro': coro, 'args': args, 'kwargs': kwargs})

    def run_delay_works(self):
        from ldplayer import LDPlayer
        from utils import close_process
        from gui import GUI
        GUI.log('')
        if self.__thread_max > 1:
            GUI.log('Запуск отложенных задач')
        LDPlayer().close_all()
        close_process(['adb.exe'])
        close_process()
        for work in self.__delay_works:
            if work['kwargs'].get('title'):
                GUI.log(work['kwargs'].get('title'))
            try:
                work['coro'](*work['args'], **work['kwargs'])
            except:
                pass
        self.__delay_works = []
        self.__start_time = time.monotonic()
        self.__run_delay_works = False

    def clean_threads(self):
        for idx in self._threads:
            if isinstance(self._threads[idx], threading.Thread) and not self._threads[idx].is_alive():
                self._threads[idx].join()
                self._threads[idx] = None

    def find_free_thread(self) -> int:
        delay_run_time = self.__thread_max * 600
        if delay_run_time > 3600:
            delay_run_time = 3600
        if time.monotonic() - self.__start_time > delay_run_time:
            self.__run_delay_works = True
            return 0
        alive_num = 1
        for idx in self._threads:
            if alive_num == idx and self._threads[idx] is not None:
                alive_num = 0
            if self._threads[idx] is None:
                alive_num = idx
        print(alive_num, self._threads)
        return alive_num


__all__ = ['ThreadingController']

Суть довольно простая: создаем контроллер (указывая максимум). А потом просто цикле запускаем th.add_work(work, ). Если есть свободный поток, он добавится, если нет, то ничего не делается. Раз в час прекращаются запускаться потоки, чтобы запустить эксклюзивные задачи в один поток (у меня банально - закрытие всех приложений и удаление временных файлов).

clean_threads - если задача выполнилась, то она будет "освобождена" и удалена из списка

Написано 14 июл. 2023

Как называется нормальный парсер прокси?

Ичи Ня @Ichi

Kristal98,

Войдите на сайт