Flask：通过写入客户端来流式传输数据？

Question

我有将数据序列化为类文件对象的现有代码：

def some_serialization_function(file):
    file.write(...)

在 Flask 中，我希望能够将序列化的数据直接发送到客户端， 无需先将其缓冲在内存中。

我查看了 werkzeug 的 ResponseStreamMixin，但我不认为它可以在没有缓冲的情况下工作：

class StreamResponse(flask.Response, werkzeug.wrappers.ResponseStreamMixin):
   pass

@app.route("/data")
def get_data():
   r = StreamResponse()
   some_serialization_function(r.stream) # everything is buffered into memory
   return r # buffered data is sent after return

我发现的所有流数据示例都基于生成器，它们以相反的方向工作（即数据是从生成器“拉出”，而不是通过写入调用“推出”），所以我想知道，有没有办法在 Flask 中直接向客户端“写入”？

编辑 - 更清楚一点：我正在寻找一种方法来提供由“some_serialization_function(...)”（我无法轻易更改）生成的数据，而没有让该函数全部写入的内存/IO开销首先将数据保存到缓冲区/文件中。

（我怀疑临时文件最终会是一种方式，因为与通过网络实际发送数据的开销相比，IO 开销不会很大。我主要关心的是内存开销）。

Answer 1

如果我理解你，你想要

• Flask Web 应用程序提供数据流
• 客户端一块一块地获取数据，而不是一大块
• Flask Web 应用程序处于控制之中，从而启动 write 应用程序。

我认为，这是无法做到的，因为必须有人控制流程，如果是 Web 应用程序，它是客户端，正在读取数据。

另一方面，如果您想防止在 Web 应用程序上缓冲要提供给客户端的整个内容，您可以在 Web 服务器上逐段读取数据并逐段输出。

逐个服务器提供内容

from flask import Flask, Response, request
app = Flask(__name__)

@app.route('/')
def hello_world():
    return 'Hello World!'

@app.route('/loop')
def loop():
    def generate():
        yield "Hello"
        yield "World"
    return Response(generate())

@app.route('/longloop/<int:rows>')
def longloop(rows):
    def generate(rows):
        for i in xrange(rows):
            yield "{i}: Hello World".format(i=i)
    return Response(generate(rows))

if __name__ == '__main__':
    app.run(debug=True)

诀窍是使用Response 对象和生成器生成输出。

如果您访问http://localhost:5000/longloop/100，您将收到100个问候。

使用curl 从命令行尝试此操作，并更好地将输出重定向到/dev/null：

$ curl -X GET http://localhost:5000/longloop/120000000 > /dev/null                                                                                                                              
  % Total    % Received % Xferd  Average Speed   Time    Time     Time  Current
                                 Dload  Upload   Total   Spent    Left  Speed
100  538M    0  538M    0     0  1056k      0 --:--:--  0:08:41 --:--:-- 1079k

正如我们所见，脚本现在运行了 8 多分钟，flask 应用程序消耗的内存仍然大致相同，在我的例子中它保持在总 RAM 的 0.4%。

Answer 2

您可以创建一个特殊的类似文件的对象，该对象将流向客户端的生成器提供给该生成器。这是一个使用队列的快速而肮脏的实现：

from queue import Queue

class StreamWriter(object):
    def __init__(self):
        self.queue = Queue()

    def write(self, str):
        self.queue.put(str)

    def read(self):
        str = self.queue.get()
        self.queue.task_done()
        if str == '~':
            return None
        return str

    def close(self):
        self.write('~')  # indicate EOF

这只不过是一个 pub-sub 类型的队列。 read() 方法将阻塞，直到在另一个线程中写入了某些内容。

现在您可以使用生成器流式传输响应。以下示例显示了一个将序列化函数作为参数的生成器。序列化函数在后台线程中执行，并接收类文件对象作为参数。

def generate_response(serialize):
    file = StreamWriter()
    def serialize_task():
        serialize(file)
        file.close()
    threading.Thread(target=serialize_task).start()
    while True:
        chunk = file.read()
        if chunk is None:
            break
        yield chunk

我希望这会有所帮助！