1

Пишу консольный многопоточный чекер для пркси, используя встроеный в python threading Раньше код выглядел так:

# Number of threads = Number of proxies
t_list = generate_threads(proxies, checker=checker, goods=goods)
for t in t_list:
    t.start()

for t in t_list:
    t.join()  # Waiting for process end checking
save_goods(goods, filename=goods_dest)

Но так как прокси много ~300 штук проверка идет порядка 30 - 40 секунд. И неплохо было бы запилить прогрессбар, который показывал бы состояние проверки.

Я решил использовать библиотеку tqdm. Переписал код так:

# Number of threads = Number of proxies
t_list = generate_threads(proxies, checker=checker, goods=goods)
for t in t_list:
    t.start()

for t in tqdm(t_list):
    t.join()  # Waiting for process end checking
save_goods(goods, filename=goods_dest)

Но у этого похода есть проблема. Т.к. потоки идут по очереди, и главный поток ждёт пока завершится следующий, то получается что если первый поток "долгий", то прогрессбар долго остаётся на 0, так как ждёт пока он выполнится. Так же я пытался вынести прогрессбар в отдельный процесс, но у меня ничего не получилось.

2 ответа 2

3

Можно воспользоваться явным изменением индикатора прогресса. Нужно завернуть функцию потока в обертку, которая по окончанию работы функции явно изменит индикатор:

from functools import wraps

lock = threading.Lock()

def execute_and_update_progress(f, progress):
    def wrapped(*args, **kwargs):
        try:
          return f(*args, **kwargs)
        finally: 
            with lock:  
                progress.update(1)
    return wrapped

# В предположении что checker это функция, которую выполняют потоки
with tqdm(total=len(proxies)) as progress:
    t_list = generate_threads(
        proxies, checker=execute_and_update_progress(checker, progress), goods=goods)
    for t in t_list:
        t.start()

    for t in t_list:
        t.join()

save_goods(goods, filename=goods_dest)

Второй вариант, это вместо явного создания потоков воспользоваться Executor-ами из concurrent.futures:

with ThreadPoolExecutor(max_workers=len(proxy)) as executor:
  lock = threading.Lock()
  futures = []
  with tqdm(total=len(proxies)) as progress:
    def update_progress():
       with lock:
         progress.update(1)
    for proxy in proxies: 
       # тут добавляем задачу на выполнение на proxy
       # нужно передать правильные параметры в функцию потока
       future = executor.submit(checker, param1, param2)
       future.add_done_callback(update_progress)
       futures.add(future)
    concurrent.futures.wait(futures)
1
  • 1
    Потому я больше люблю multiprocessing.dummy.Pool для обработки тредов
    – eri
    17 дек 2020 в 8:05
2

Тогда нужно проверять, какие потоки уже завершились с помощью метода is_alive(), прежде чем делать им join(). Мне кажется, можно сделать как-то так (код не проверял):

def get_finished_threads(threads):
    t_copy = threads.copy()
    while t_copy:
        finished = filter(lambda x: not x.is_alive(), t_copy)
        if not finished:
            time.sleep(0.1) # чтобы не съесть всё процессорное время на проверки
        for t in finished:
            t_copy.remove(t)
            yield t

# ...
for t in tqdm(get_finished_threads(t_list), total=len(t_list)):
    t.join()  # Waiting for process end checking
1
  • Да, код работает, спасибо) 17 дек 2020 в 8:04

Ваш ответ

By clicking “Отправить ответ”, you agree to our terms of service and acknowledge that you have read and understand our privacy policy and code of conduct.

Всё ещё ищете ответ? Посмотрите другие вопросы с метками или задайте свой вопрос.