帮助文档
专业提供香港服务器、香港云服务器、香港高防服务器租用、香港云主机、台湾服务器、美国服务器、美国云服务器vps租用、韩国高防服务器租用、新加坡服务器、日本服务器租用 一站式全球网络解决方案提供商!专业运营维护IDC数据中心,提供高质量的服务器托管,服务器机房租用,服务器机柜租用,IDC机房机柜租用等服务,稳定、安全、高性能的云端计算服务,实时满足您的多样性业务需求。 香港大带宽稳定可靠,高级工程师提供基于服务器硬件、操作系统、网络、应用环境、安全的免费技术支持。
服务器资讯 / 香港服务器租用 / 香港VPS租用 / 香港云服务器 / 美国服务器租用 / 台湾服务器租用 / 日本服务器租用 / 官方公告 / 帮助文档
Flask——基于python完整实现客户端和服务器后端流式请求及响应
发布时间:2024-03-06 04:53:10   分类:帮助文档
Flask——基于python完整实现客户端和服务器后端流式请求及响应






文章目录
本地客户端Flask服务器后端客户端/服务器端流式接收[打字机]效果




看了很多相关博客,但是都没有本地客户端和服务器后端的完整代码示例,有的也只说了如何流式获取后端结果,基本没有讲两端如何同时实现流式输入输出,特此整理总结,给大家交流学习和使用!

本地客户端
requests.post得到流式回复的重要参数:
stream:需要设置为True;response.iter_content:使用该函数接收返回的流式数据。
import requests
import time
import json

def generate_stream_data():
# 假设这是要发送的文本列表
is_end = False
lines = ["Hello", "world", "this", "is", "a", "stream", "of", "text"]
for line in lines:
print(line)
if lines.index(line) == len(lines) - 1:
is_end = True
yield json.dumps({'line': line, 'is_end': is_end}) + '\n'
time.sleep(0.5)
# 模拟数据处理时间

def get_stream_response(response):
# 流式接收response
rec_data_list = []
temp_data = ''
for chunk in response.iter_content(chunk_size=1):
temp_data += chunk.decode('utf-8')
if temp_data.endswith('\n'):
temp_json = json.loads(temp_data)
rec_data_list.append(temp_json)
print(temp_data)
temp_data = ''
if temp_json['is_end']:
break
print(rec_data_list)
print("----------------------------")
print(temp_data)
return rec_data_list

def stream_upload(url):

# 流式接收response
response = requests.post(url, data=generate_stream_data(), stream=True)

final_response = get_stream_response(response)

return final_response

url = 'http://127.0.0.1:5000/stream'
response = stream_upload(url)

Flask服务器后端
flask.request流式获取数据::
使用request.stream.read读取数据,而不是get_data()等一次性函数。
from flask import Flask, Response, request
import time
import json
import requests

app = Flask(__name__)

def process_stream_data(stream_data):
# 假设这是要发送的数据
print("开始生成新的数据流")
is_end = False
print(stream_data)
for idx, line in enumerate(stream_data):
if idx == len(stream_data)-1:
is_end = True
print(line)
yield json.dumps(line)+"\n"
time.sleep(0.5)
# 模拟数据处理时间

def get_stream_request(chunk_size=1):
req_data_list = []
temp_data = ''
while True:
chunk = request.stream.read(chunk_size)
temp_data += chunk.decode('utf-8')
if temp_data.endswith('\n'):
temp_json = json.loads(temp_data)
req_data_list.append(temp_json)
print(temp_data)
temp_data = ''
if temp_json['is_end']:
return req_data_list

@app.route('/stream', methods=['POST'])
def stream_text():

data = get_stream_request()

print("----------------------------")

return Response(process_stream_data(data))

if __name__ == "__main__":
app.run(host='0.0.0.0', port=5000, debug=True)

客户端/服务器端流式接收[打字机]效果





香港云服务器租用推荐
服务器租用资讯
·广东云服务有限公司怎么样
·广东云服务器怎么样
·广东锐讯网络有限公司怎么样
·广东佛山的蜗牛怎么那么大
·广东单位电话主机号怎么填写
·管家婆 花生壳怎么用
·官网域名过期要怎么办
·官网邮箱一般怎么命名
·官网网站被篡改怎么办
服务器租用推荐
·美国服务器租用
·台湾服务器租用
·香港云服务器租用
·香港裸金属服务器
·香港高防服务器租用
·香港服务器租用特价