Неблокирующая очередь потоков

Question

Я хочу создать простую очередь из потоков.

РЕЗЬБЫ НАЧИНАЮТСЯ ПОСЛЕ ЗАПРОСА

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

def a():
    print('start a')
    sleep(5)
    print('end a')


def b():
    print('start b')
    sleep(5)
    print('end b')


t = Thread(target=a)
t.start()
t.join()
print('test1')
t = Thread(target=b)
t.start()
t.join()
print('test2')

Результат кода:

start a
end a
test1
start b
end b
test2

Expectation:

start a
test1
end a
start b
test2
end b

Я знаю, что могу печатать ('test1') перед соединением, но когда я использую пост-запрос, поток может начаться во время печати ('test1')

t.start()
print('test1') <- here comes post request
               t1.start()
               print('test2')
               t1.join()
t.join()

В этом случае потоки будут работать параллельно

Answer 1

Вы можете передать timeout аргумент join(), это заблокирует вызывающий поток максимум timeout секунд. «Максимальный», потому что поток может завершиться раньше. Если вы не передадите тайм-аут, он будет блокироваться до завершения потока. Итак, что вам нужно сделать в вашем примере, это присоединиться к потоку дважды, в первый раз с таймаутом, который позволяет печатать test1.

from threading import Thread
from time import sleep

def a():
    print('start a')
    sleep(5)
    print('end a')


def b():
    print('start b')
    sleep(5)
    print('end b')


t = Thread(target=a)
t.start()
t.join(0.1)
print('test1')
t.join()

t = Thread(target=b)
t.start()
t.join(0.1)
print('test2')
t.join()

Выход:

start a
test1
end a
start b
test2
end b

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

import logging
from time import sleep
from multiprocessing.pool import ThreadPool


def make_request(*args):
    logger.debug(f'making request number {args[0]}')
    sleep(5)


def init_logging(log_level=logging.DEBUG):
    fmt = '[%(asctime)s %(levelname)-8s %(threadName)s' \
          ' %(funcName)s()] --- %(message)s'
    logging.basicConfig(format=fmt, level=log_level)


if __name__ == '__main__':

    N_THREADS = 5
    N_REQUESTS = 10

    arguments = [*zip(range(N_REQUESTS))]  # [(0,), (1,), (2,) ...]

    init_logging()
    logger = logging.getLogger()

    with ThreadPool(N_THREADS) as pool:
        for args in arguments:
            pool.apply_async(make_request, args=args)
            sleep(1)  # heartbeat

Пример вывода:

[2018-09-09 03:17:06,303 DEBUG    Thread-1 make_request()] --- making request number 0
[2018-09-09 03:17:07,304 DEBUG    Thread-2 make_request()] --- making request number 1
[2018-09-09 03:17:08,306 DEBUG    Thread-3 make_request()] --- making request number 2
[2018-09-09 03:17:09,307 DEBUG    Thread-4 make_request()] --- making request number 3
[2018-09-09 03:17:10,308 DEBUG    Thread-5 make_request()] --- making request number 4
[2018-09-09 03:17:11,309 DEBUG    Thread-1 make_request()] --- making request number 5
[2018-09-09 03:17:12,310 DEBUG    Thread-2 make_request()] --- making request number 6
[2018-09-09 03:17:13,311 DEBUG    Thread-3 make_request()] --- making request number 7
[2018-09-09 03:17:14,312 DEBUG    Thread-4 make_request()] --- making request number 8
[2018-09-09 03:17:15,313 DEBUG    Thread-5 make_request()] --- making request number 9

Process finished with exit code 0