python - Python 中的异步流处理

Question

让我们从一个简单的例子开始。 HTTP 数据流采用以下格式:

MESSAGE_LENGTH, 2 bytes
MESSAGE_BODY, 
REPEAT...

目前，我使用 urllib2 来检索和处理流式数据，如下所示:

length = response.read(2)
while True:
    data = response.read(length)
    DO DATA PROCESSING

它有效，但由于所有消息的大小都是 50-100 字节，上述方法每次读取时都会限制缓冲区大小，因此可能会影响性能。

是否可以使用单独的线程进行数据检索和处理？

Answer 1

是的，可以做到，而且并不难，如果您的格式基本上是固定的。

我在 Python 2.2.3 中将它与 httplib 一起使用，发现它在我们一起破解它的方式中有一些糟糕的性能(基本上是猴子将基于 select() 的套接字层修补到 httplib 中)。

诀窍是获取套接字并自己进行缓冲，因此您不会与中间层争夺缓冲(当我们有用于分块 http 解码的 httplib 缓冲区、用于 read() 的套接字层缓冲区时，性能非常糟糕) ).

然后有一个状态机，它在需要时从套接字中获取新数据，并将完成的 block 推送到 Queue.Queue 中，为您的处理线程提供数据。

我用它来传输文件，在一个额外的线程中对它们进行校验和 (zlib.ADLER32)，然后在第三个线程中将它们写入文件系统。通过套接字和 HTTP/分块开销在我的本地机器上实现大约 40 MB/s 的持续吞吐量。