网站建设和管理情况自查报告,做网站需要去哪里备案,做运动特卖的网站,在线编辑文章目录 本地客户端Flask服务器后端客户端/服务器端流式接收[打字机]效果 看了很多相关博客#xff0c;但是都没有本地客户端和服务器后端的完整代码示例#xff0c;有的也只说了如何流式获取后端结果#xff0c;基本没有讲两端如何同时实现流式输入输出#xff0c;特此整… 文章目录 本地客户端Flask服务器后端客户端/服务器端流式接收[打字机]效果 看了很多相关博客但是都没有本地客户端和服务器后端的完整代码示例有的也只说了如何流式获取后端结果基本没有讲两端如何同时实现流式输入输出特此整理总结给大家交流学习和使用 本地客户端
requests.post得到流式回复的重要参数 stream需要设置为Trueresponse.iter_content使用该函数接收返回的流式数据。
import requests
import time
import jsondef generate_stream_data():# 假设这是要发送的文本列表is_end Falselines [Hello, world, this, is, a, stream, of, text]for line in lines:print(line)if lines.index(line) len(lines) - 1:is_end Trueyield json.dumps({line: line, is_end: is_end}) \ntime.sleep(0.5)# 模拟数据处理时间def get_stream_response(response):# 流式接收responserec_data_list []temp_data for chunk in response.iter_content(chunk_size1):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]:breakprint(rec_data_list)print(----------------------------)print(temp_data)return rec_data_listdef stream_upload(url):# 流式接收responseresponse requests.post(url, datagenerate_stream_data(), streamTrue)final_response get_stream_response(response)return final_responseurl 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 requestsapp Flask(__name__)def process_stream_data(stream_data):# 假设这是要发送的数据print(开始生成新的数据流)is_end Falseprint(stream_data)for idx, line in enumerate(stream_data):if idx len(stream_data)-1:is_end Trueprint(line)yield json.dumps(line)\ntime.sleep(0.5)# 模拟数据处理时间def get_stream_request(chunk_size1):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_listapp.route(/stream, methods[POST])
def stream_text():data get_stream_request()print(----------------------------)return Response(process_stream_data(data))if __name__ __main__:app.run(host0.0.0.0, port5000, debugTrue)客户端/服务器端流式接收[打字机]效果
