带队列的Python线程: how to avoid to use join?

Question

我有一个有 2 个线程的场景:

1. 一个线程等待来自套接字的消息(嵌入在 C 库中 - 阻塞调用是“Barra.ricevi”)，然后将一个元素放入队列

2. 一个线程等待从队列中获取元素并执行某些操作

示例代码

import Barra
import Queue    
import threading

posQu = Queue.Queue(maxsize=0)

def threadCAN():
    while True:
        canMsg = Barra.ricevi("can0")
        if canMsg[0] == 'ERR':
            print (canMsg)
        else:
            print ("Enqueued message"), canMsg
            posQu.put(canMsg)

thCan = threading.Thread(target = threadCAN)
thCan.daemon = True
thCan.start()

while True:
    posMsg = posQu.get()
    print ("Messagge from the queue"), posMsg

结果是，每次从套接字发出新消息时，都会将新元素添加到队列中，但应该从队列中获取项目的主线程永远不会被唤醒。

输出如下:

Enqueued message

Enqueued message

Enqueued message

Enqueued message

我期望有:

Enqueued message

Messagge from the queue

Enqueued message

Messagge from the queue

解决此问题的唯一方法是添加以下行:

posQu.join()

在等待来自套接字的消息的线程末尾，以及该行:

posQu.task_done()

在主线程的末尾。

在这种情况下，从套接字接收到新消息后，线程将阻塞等待主线程处理排队的项目。

不幸的是，这不是所需的行为，因为我希望线程始终准备好从套接字获取消息，而不是等待另一个线程完成作业。

我做错了什么？ 谢谢

安德鲁 (意大利)

Answer 1

这可能是因为您的 Barra 在 Barra.ricevi 时没有释放全局解释器锁 (GIL)。不过，您可能想检查一下。

GIL 确保任一时刻只能运行一个线程(限制了多处理器系统中线程的有用性)。 GIL 每 100 个“刻度”切换一次线程——一个松散映射到字节码指令的刻度。请参阅here了解更多详情。

在您的生产者线程中，C 库调用之外没有发生太多事情。这意味着生产者线程将在 GIL 切换到另一个线程之前多次调用 Barra.ricevi。

就增加复杂性而言，解决方案是:

• 将项目添加到队列后调用 time.sleep(0)。这会产生该线程，以便另一个线程可以运行。
• 使用sys.setcheckinterval()来减少切换线程之前执行的“ticks”数量。这是以程序的计算成本更高为代价的。
• 使用多处理而不是线程。这包括使用 multiprocessing.Queue 而不是 Queue.Queue。
• 修改 Barra，以便在调用其函数时释放 GIL。

使用多处理的示例。请注意，使用多处理时，您的进程不再具有隐含的共享状态。您需要查看多处理以了解如何在进程之间传递信息。

import Barra  
import multiprocessing

def threadCAN(posQu):
    while True:
        canMsg = Barra.ricevi("can0")
        if canMsg[0] == 'ERR':
            print(canMsg)
        else:
            print("Enqueued message", canMsg)
            posQu.put(canMsg)

if __name__ == "__main__":
    posQu = multiprocessing.Queue(maxsize=0)
    procCan = multiprocessing.Process(target=threadCAN, args=(posQu,))
    procCan.daemon = True
    procCan.start()

    while True:
        posMsg = posQu.get()
        print("Messagge from the queue", posMsg)