python - 使用 Twisted 或使用线程、Python 中的队列处理大量流数据

Question

我正在以极快的速度从与 Twitter API 流媒体服务器的长期连接中获取推文。我继续进行一些繁重的文本处理，并将推文保存在我的数据库中。

我将 PyCurl 用于连接和回调函数，负责文本处理和保存在数据库中。请参阅下面我无法正常工作的方法。

我对网络编程不熟悉，所以想知道:如何使用 Threads、Queue 或 Twisted 框架来解决这个问题？

def process_tweet():
    # do some heaving text processing


def open_stream_connection():
    connect = pycurl.Curl()
    connect.setopt(pycurl.URL, STREAMURL)
    connect.setopt(pycurl.WRITEFUNCTION, process_tweet)
    connect.setopt(pycurl.USERPWD, "%s:%s" % (TWITTER_USER, TWITTER_PASS))
    connect.perform()

Answer 1

您应该有许多线程在消息传入时接收它们。如果您使用的是 pycurl，该数字可能应该是 1，但如果您使用的是 httplib，则该数字应该更高 - 这个想法是您希望能够拥有一次在 Twitter API 上有多个查询，因此需要处理的工作量稳定。

当每条推文到达时，它被推送到 Queue.Queue。队列确保通信中存在线程安全 - 每条推文将仅由一个工作线程处理。

工作线程池负责从队列中读取并处理推文。只应将有趣的推文添加到数据库中。

由于数据库可能是瓶颈，因此池中值得添加的线程数量是有限制的——更多的线程不会使其处理速度更快，这只会意味着更多的线程在队列中等待访问数据库。

这是一个相当常见的 Python 习语。这种架构只会在一定程度上扩展 - 即一台机器可以处理的内容。