开发者社区 > 云原生 > Serverless > 正文

函数计算,python环境flask框架,返回text/event-stream流数据的时候总是把所

函数计算,python环境flask框架,返回text/event-stream流数据的时候总是把所有chunk合在一起返回,而不是逐chunk返回,与真实环境不符。请问有办法解决吗?浏览器请求该接口,真实环境会逐字打印到浏览器页面,而函数计算FC会下载一个文件

展开
收起
古拉古拉 2023-08-01 10:49:35 1737 0
3 条回答
写回答
取消 提交回答
  • 北京阿里云ACE会长

    为了解决这个问题,您可以尝试使用以下两种方法:

    禁用 Flask 框架的响应缓存:可以通过设置 Cache-Control 头信息来禁用 Flask 框架的响应缓存,从而实现逐 chunk 返回的效果。例如,您可以在返回响应时添加以下头信息:
    python
    Copy
    from flask import Response

    @app.route('/stream')
    def stream():
    def generate():
    yield "data: chunk1\n\n"
    yield "data: chunk2\n\n"
    yield "data: chunk3\n\n"

    return Response(generate(), mimetype='text/event-stream', headers={'Cache-Control': 'no-cache', 'X-Accel-Buffering': 'no'})
    

    使用其他框架:如果您不想禁用 Flask 框架的响应缓存,可以尝试使用其他框架来实现逐 chunk 返回的效果。例如,您可以使用 Tornado 框架来实现 text/event-stream 流数据的返回,Tornado 框架天生支持流式响应,可以实现逐 chunk 返回的效果。
    python
    Copy
    import tornado.ioloop
    import tornado.web

    class MainHandler(tornado.web.RequestHandler):
    @tornado.web.asynchronous
    def get(self):
    self.set_header('Content-Type', 'text/event-stream')
    self.set_header('Cache-Control', 'no-cache')
    self.write('data: chunk1\n\n')
    self.flush()
    tornado.ioloop.IOLoop.current().add_timeout(2, self.chunk2)

    def chunk2(self):
        self.write('data: chunk2\n\n')
        self.flush()
        tornado.ioloop.IOLoop.current().add_timeout(2, self.chunk3)
    
    def chunk3(self):
        self.write('data: chunk3\n\n')
        self.flush()
        self.finish()
    

    def make_app():
    return tornado.web.Application([
    (r"/stream", MainHandler),
    ])

    if name == "main":
    app = make_app()
    app.listen(8888)
    tornado.ioloop.IOLoop.current().start()

    2023-08-03 19:38:03
    赞同 展开评论 打赏
  • 回答5.png

    此答案来自钉钉群“阿里函数计算官网客户"

    2023-08-01 19:35:39
    赞同 展开评论 打赏
  • 存在即是合理

    你可以尝试使用flask.Response对象的stream_with_context方法来逐块返回数据。示例:

    from flask import Flask, Response, stream_with_context
    import time
    
    app = Flask(__name__)
    
    def generate_data():
        for i in range(10):
            yield f"data: {i}
    
    "
            time.sleep(1)
    
    @app.route('/stream')
    def stream():
        return Response(stream_with_context(generate_data()), content_type='text/event-stream')
    
    if __name__ == '__main__':
        app.run(debug=True)
    

    这个示例中,generate_data函数会生成一个包含10个数字的数据流,每个数字之间间隔1秒。通过使用stream_with_context,我们可以确保在计算环境中逐块返回数据。

    2023-08-01 13:03:28
    赞同 展开评论 打赏

快速交付实现商业价值。

相关产品

  • 函数计算
  • 相关电子书

    更多
    Hologres Serverless之路:揭秘弹性计算组 立即下载
    Serverless开发平台: 让研发效能再提升 立即下载
    Serverless 引领云上研发新范式 立即下载