线程的非阻塞队列答案

【问题标题】：Nonblocking queue of threads线程的非阻塞队列
【发布时间】：2019-02-12 02:00:05
【问题描述】：

我想创建简单的线程队列。

线程将从 POST 请求开始

我创建了没有请求的简单示例。我尝试加入线程，但它没有按我的意愿工作。

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

期望：

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()

在这种情况下线程将并行运行

【问题讨论】：

您是否总是希望 first 线程在第二个线程开始之前完成？完成顺序重要吗？
@wwii 线程的顺序并不那么重要。我正在使用限制每秒请求数的 api，所以最重要的是避免并行运行线程。
如果你想“避免并行运行线程”，为什么要使用线程？
我的回答没有解决您的问题吗？

标签： python multithreading concurrency queue python-multithreading

【解决方案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

【讨论】：