1

Использую следующую конструкцию:

def request():
    try:
        r = requests.get(url)
    except requests.exceptions.HTTPError as err:
        print(err)
    except requests.exceptions.ChunkedEncodingError:
        print('Server returned nothing(0 bytes). Breaking connection.')
        break
    except requests.exceptions.ConnectionError as err:
        print('Connection error!')
        break

q = Queue()
try:
    for i in range(len(ips)):
        t = threading.Thread(target=request)
        t.start()
except KeyboardInterrupt:
    print('Stopping...')
    sys.exit()

Что меня в этой конструкции не устраивает:

  1. Если отправлять запросы через прокси с теле while, то скрипт нельзя остановить через ctrl+c, пока не будут выполнены все запросы.
  2. Если всё-таки попытаться остановить скрипт, то вылетит исключение KeyboardInterrupt, которое обрабатывается, но почему-то исключение всё равно появляется.

Как я понимаю, мне нужно сперва остановить все процессы в queue, чтобы остановить основной процесс. Собственно, как мне это сделать? Документации по-настоящему мало.

1
  • Самый простой выход - поставить t.daemon = True перед t.start().
    – faoxis
    14 фев 2017 в 14:15

1 ответ 1

1
import time, random
from threading import Thread
from queue import Queue

class Worker(Thread):
    def __init__(self, parent):
        self.parent = parent
        Thread.__init__(self)
        Thread.start(self)

    def run(self):
        '''
        выполнить target в потоке
        выйти, при self.parent.is_running=False, после выполнения текущей target
        '''
        while self.parent.is_running:
            t = self.parent.queue_in.get()  # получить
            try:
                r = t['target'](*t['args'], **t['kwargs'])  # выполнить
                print(r, self)

            except Exception:
                if t is None: return  # выход из queue_in.get()
                else: raise
            finally: self.parent.queue_in.task_done()

class Pool:
    def __init__(self, size: int):
        self.is_running = True
        self.queue_in = Queue()
        self.threads = [Worker(self) for _ in range(size)]

    def __call__(self, target: callable):
        '''queue_in.put(dict)'''
        def put(*args, **kwargs):
            self.queue_in.put(dict(target=target, args=args, kwargs=kwargs))
        return put

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        for _ in self.threads:
            self.queue_in.put(None)

def request(i, t=0):
    '''target'''
    time.sleep(t)
    return i

if __name__ == '__main__':
    # создать 2 потока
    with Pool(size=2) as pool:
        # добавить 100 вызовов request()
        for i in range(100):
            pool(request)(i, t=random.randrange(3))

        try:  # 'бесконечно' ожидать(пока запушен любой из pool.threads)
            while any(th.is_alive() for th in pool.threads):
                time.sleep(1)
                if random.randrange(2):  # выйти
                    raise KeyboardInterrupt  # ctrl+c
        except KeyboardInterrupt:
            # pass  # выйти после выполнения всех request
            pool.is_running = False  # выйти после выполнения текущих request

добавить 100 вызовов request, и прервать в процессе выполнения KeyboardInterrupt :

0 <Worker(Thread-1, started 4936)>
2 <Worker(Thread-1, started 4936)>
1 <Worker(Thread-2, started 1984)>

Ваш ответ

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

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