Python 多线程与共享变量

Question

我正在尝试并行化我的工作，但我对多线程很陌生，因此对具体实现感到困惑。

我有一个套接字监听器，它将数据保存到缓冲区。当缓冲区达到其容量时，我需要将其数据保存到数据库中。在一个线程上，我想启动套接字监听器，而在并行任务上，我想检查缓冲区状态。

BufferQueue 只是 python list 的扩展，具有允许检查列表是否达到指定大小的方法。

SocketManager 是我正在收听的 STREAM_URL 的流数据提供者。它使用回调函数来处理消息

但是当我使用回调来检索数据时，我不确定使用共享变量是正确且最佳的决定

buffer = BufferQueue(buffer_size=10000)

def start_listening_to_sokcet(client):
    s = SocketManager(client)
    s.start_socket(cb_new)
    s.start()

def cb_new(message):
    print("New message")
    global buffer
    for m in message:
        #save data to buffer

def is_buffer_ready(buffer):
    global buffer
    print("Buffer state")
    if buffer.ready():
         #save buffer data to db

如果您能帮助我处理此案，我将不胜感激

Answer 1

我认为您正在寻找的只是 queue模块。

queue.Queue是一个自同步队列，专门用于在线程之间传递对象。

默认情况下，在队列上调用 get 将会阻塞，直到有对象可用为止，这正是您通常想要做的 — 在网络应用程序中使用线程进行并发的要点是您的线程所有这些看起来都像普通的同步代码，但大部分时间都花在等待套接字、文件、队列或其他无事可做的东西上。但您可以使用 block=False 进行检查而不阻塞，或者在等待时设置 timeout。

您还可以在构造队列时指定maxsize。然后，默认情况下，put 将阻塞，直到队列不太满而无法接受新对象。但是，同样，您可以使用 block 或 timeout 来尝试，如果太满则失败。

所有同步均在 get 和 put 内部处理，因此您不需要 Lock 来保证线程安全或向服务员发出信号的条件。

队列甚至可以为您处理关闭操作。生产者可以放置一个特殊的值，告诉消费者在get上看到它时退出。

对于生产者需要等待消费者完成的正常关闭，您可以在消费者完成处理每个排队对象后使用可选的 task_done 方法，并让生产者阻塞在join 方法。但如果您不需要这个，或者有其他方法来等待关闭，例如加入消费者线程，您可以跳过这一部分。