拓冰网站定制 
我最早正经写服务端接收数据是因为要给一组环境传感器做数据上报。当时第一反应是找个现成的后端框架翻了半天企业级框架的教程被那一堆配置直接劝退——程序能不能跑起来另说光理解为什么一个接收数据的工程需要这么多文件就够头疼。后来我换了个思路我其实只需要一个一直运行的程序监听某个端口等数据进来收下来存起来再给发送方回一句收到了。就这点事不夸张地说Python标准库就能做到换Flask写也就三十行代码。这篇文章就把这套最简单服务端从头到尾讲透服务端接收数据到底接收的是什么、不同场景怎么选方案、代码怎么写、怎么自测以及我在实际项目里被数据接收坑过的几个点。内容不追求大而全追求的是你照着做十分钟内能跑通一个真正能收数据的服务端。适合刚接触前后端联调、在做小项目、搞硬件上报或者只是临时想验证某个接口的人。1. 先把接收数据这件事拆明白1.1 我们口中的接收数据到底是在接收什么很多人一听服务端接收数据脑子里浮现的是后台管理系统、数据库集群、消息队列这些东西。但实际上绝大多数场景根本没到那个复杂度。我把遇到过的接收数据需求归了下类基本都是下面几种网页表单提交前端页面填了个表单点提交后端要拿到这些字段存下来。App或小程序上报手机端采集了用户行为、定位、日志定时往服务端发送。硬件设备上报温湿度传感器、GPS追踪器、摄像头状态通过HTTP接口周期性上报数据。系统间回调通知支付平台、短信平台在某个事件发生后回调你的接口告诉你订单已支付。脚本或爬虫抓取结果另一个脚本跑完任务后把结果POST给你。这些场景有个共同点都是有一个东西主动把数据发给你运行的程序。你的程序不需要去连接别人只需要老老实实在一个端口上等着。所以接收数据这个动作本质上就三件事持续监听某个网络端口收到请求后把请求里的数据解析出来给发送方一个明确的响应。仅此而已。至于数据库、事务、权限这些都是收到数据之后的事不属于接收这个动作本身。1.2 一次请求从发出到被接收中间发生了什么为了后面调试方便这里有必要把链路简单过一遍。你可以把它理解成去餐厅吃饭客户端浏览器、手机、传感器就是顾客它知道你的服务端地址IP加端口相当于知道饭馆开在哪条街几号。顾客进门喊服务员发起HTTP请求这一步走的是你服务端正在监听的端口比如5000。你的程序就是服务员听到喊声后要看顾客点了什么菜读取请求里的数据然后去后厨做业务处理最后把菜端上桌返回响应。在这个过程里几个关键要素要记住请求行POST /receive HTTP/1.1 请求头Content-Type: application/json 告诉服务端数据是什么格式 请求体{temp: 26.5, humidity: 60} 真正要接收的数据服务端接收数据做的就是两件事解析请求头按格式读请求体。至于0.0.0.0这类监听地址、5000这种端口号都是你开启监听时需要先理解的参数——监听地址决定谁能连、端口决定连哪里。1.3 GET和POST两种最常见的递纸条方式初学者最容易混淆的就是GET和POST的区别。其实从服务端接收数据的角度看区别非常直观方式数据放在哪里常见用途特点GETURL末尾的查询字符串里比如?nameabcvalue123查询、调试、参数简单可被浏览器地址栏直接访问会留在历史记录里POSTHTTP请求体的body里提交表单、上报JSON数据、文件上传数据不在URL中可传大容量和复杂结构举个例子同样是上报一条温度数据GET方式 http://localhost:5000/receive?deviceesp32temp26.5 POST方式 http://localhost:5000/receive body: {device: esp32, temp: 26.5}如果你只是自己调试GET快得多浏览器里直接打开链接就能触发。但准备正式接收业务数据我建议一律用POST原因有三点数据结构可以复杂一点点、长度不受URL限制、不会因为数据里的特殊字符导致URL解析出错。提示根本就不存在所谓的GET更安全这种说法。真正的数据保密靠的是HTTPS而不是靠把数据塞进body里掩耳盗铃。2. 别一上来就整企业级框架先选一个真简单的方案2.1 常见方案横向对比简单这两个字不同背景的人理解不同。一个Java后端朋友觉得Spring Boot已经很简单了但对一个刚转行、或者只是搞个硬件Demo的人来说光把Maven依赖拉下来就得折腾半天。我按从零开始跑通一个接收数据的服务端需要付出的成本把常见方案排了个队方案依赖启动成本适合场景短板Python内置http.server零依赖最低Python自带临时调试、一次性任务要手动解析报文不适合做业务Python Flask需装Flask低一个pip命令快速原型、小型项目、硬件上报性能上限不如专门的高性能框架Node.js Express需装Node依赖低前端顺手、Node环境已有回调风格需要熟悉Go 标准库net/http零第三方依赖低要单文件部署、追求性能语法对新手略陌生Java Spring Boot重高大型团队、复杂业务很多人就是被它劝退的你可能会问既然Python内置的http.server零依赖为什么不直接用它因为它把接收数据这件事做成了最原始的样子收到的是一个完整HTTP报文你要从字符串里手动抠出请求头、body、参数还得自己处理url解码、JSON解析。写个Demo没问题做正经功能就是在给自己挖坑。2.2 为什么我推荐Flask作为起步方案Flask是一个微框架Micro Framework所谓微指的是它只解决你递纸条的核心需求路由、请求解析、响应返回。它不需要你理解依赖注入中间件配置中心这些花哨的概念。我用Flask已经很多年总结它的优势就三条路由写起来像在报菜名。app.route(/receive, methods[POST])底下直接跟处理的函数一眼就能看懂哪个路径对应哪段逻辑。请求数据解析是现成的。request.args拿GET参数、request.get_json()拿JSON、request.form拿表单数据不用自己写解析逻辑。返回JSON也方便。直接返回一个dictFlask自动帮你序列化并加上对应的Content-Type头。还有一个不太被提及但很现实的好处Flask的教程、问答、现成代码是Python后端框架里最多的。你遇到任何小问题比如Flask接收不到POST的body搜一下基本就有答案。对于新手或者做快速验证的人来说这个生态价值比框架本身的性能重要得多。注意如果你确实只是想在终端里快速看一眼某个地址发来的原始请求长什么样那用http.server反而快。在Python环境里执行python -m http.server 8000浏览器访问一下终端会输出请求日志。这个技巧可以用来应急。3. 三十行代码搭出第一个能收数据的服务端3.1 环境准备就两步先确认Python已经装好在终端里执行python3 --version看到版本号之后安装Flaskpip install flask国内网络如果下载慢可以临时换一下Python软件源但不要为了图快装来路不明的包。装完验证一下python3 -c import flask; print(flask.__version__)没报错就是装好了。3.2 第一版把GET参数原样打印出来创建一个文件server.py先写个最基础的版本from flask import Flask, request app Flask(__name__) app.route(/receive, methods[GET]) def receive(): name request.args.get(name, ) value request.args.get(value, ) print(收到GET请求, name, value) return ok if __name__ __main__: app.run(host0.0.0.0, port5000, debugTrue)这段代码要拆开看app Flask(__name__)创建了一个Flask应用后面所有路由都挂在它上。app.route(/receive, methods[GET])声明了路径和方法只有访问/receive并且是GET下面的函数才会执行。request.args是所有GET查询参数的集合request.args.get(name)拿到?name后面的值。app.run(host0.0.0.0, port5000)启动监听0.0.0.0表示允许局域网内其他设备访问如果只想本机访问就改成127.0.0.1。启动python3 server.py然后另开一个终端用curl模拟发数据curl http://localhost:5000/receive?nametempvalue26.5回到服务端终端你会看到打印出收到GET请求 temp 26.5到这里你的服务端已经真正意义上接收到了数据。虽然是GET但骨架已经有了。3.3 第二版接收POST JSON数据实际业务里最常遇到的就是客户端POST一个JSON过来。把代码升级一下加点容错from flask import Flask, request app Flask(__name__) app.route(/api/data, methods[POST]) def api_data(): # silentTrue: 如果body不是合法JSON不会抛异常而是返回None data request.get_json(silentTrue) if data is None: return {code: 1, message: JSON解析失败}, 400 print(收到JSON数据, data) return {code: 0, message: ok, received: data} if __name__ __main__: app.run(host0.0.0.0, port5000, debugTrue)这次的改动要点request.get_json(silentTrue)直接解析请求体里的JSON。silentTrue这一步很重要——如果客户端发了个我不是JSON的东西过来框架会抛一个400错误但直接用它会在调试时看到一屏幕堆栈。你更希望的是拿到一个None然后自己决定怎么回复。返回的是字典Flask会自动把它变成JSON字符串还会自动设置Content-Type: application/json。返回400状态码语义上告诉客户端你发的东西不合规范。测试命令curl -X POST \ -H Content-Type: application/json \ -d {device:esp32,temp:26.5,humidity:60} \ http://localhost:5000/api/data服务端打印收到JSON数据 {device: esp32, temp: 26.5, humidity: 60}curl返回{code:0,message:ok,received:{device:esp32,temp:26.5,humidity:60}}3.4 第三版一个统一入口GET、POST、JSON、表单全接住真实的接收场景往往没有你想的那么规范。有可能前端一会儿用JSON一会儿用表单也有可能是你临时给别的部门同学调试对方习惯用GET带参数。我的做法是做一个通用接收器不管什么格式先接住把内容打印出来并原样返回from flask import Flask, request app Flask(__name__) app.route(/receive, methods[GET, POST]) def receive_all(): # 记录请求方法 method request.method # 1. GET查询参数比如 /receive?a1b2 query_params dict(request.args) # 2. 判断Content-Type决定怎么解析body content_type request.headers.get(Content-Type, ) body_data None if application/json in content_type: body_data request.get_json(silentTrue) elif application/x-www-form-urlencoded in content_type: body_data dict(request.form) else: # 其他类型先按原文读出来 body_data request.get_data(as_textTrue) print(f[{method}] 查询参数: {query_params}) print(f[{method}] body内容: {body_data}) return { method: method, query_params: query_params, body: body_data, status: received } if __name__ __main__: app.run(host0.0.0.0, port5000, debugTrue)这个版本好在哪它把接收和业务处理解耦了。不管你收到什么先把内容记录清楚调试的时候一目了然。后面想接数据库也好、转发给别的服务也好都是在拿到body_data之后再加逻辑。提示request.get_data(as_textTrue)是最后的兜底方案它把原始body当字符串读出来。碰到文件上传时这个不适用文件要用request.files这点后面单独说。4. 数据收到之后拿它干什么4.1 别用print改用日志我最早偷懒直接print结果服务端跑了一晚上第二天想查昨晚哪些设备上报了数据终端翻半天全被滚动刷没了。你要记住print 是给调试用的日志是给运行用的。先用Python自带的logging模块把日志写到文件里import logging from logging.handlers import RotatingFileHandler # 配置日志按大小轮转单个文件超过1MB自动切分 handler RotatingFileHandler(server.log, maxBytes1024*1024, backupCount5) formatter logging.Formatter(%(asctime)s - %(levelname)s - %(message)s) handler.setFormatter(formatter) logger logging.getLogger(my_server) logger.setLevel(logging.INFO) logger.addHandler(handler)然后在接收函数里把原来的print替换成logger.info(f收到请求: {method} 查询参数: {query_params} body: {body_data})这样数据接收完服务端程序爱怎么重启都没事日志文件里什么都在。4.2 把数据存进SQLite不用装任何数据库很多时候接收完数据是要落库的。对小型项目、个人工具、硬件上报这类场景我强烈推荐直接用Python标准库自带的SQLite——不用装MySQL不用配置连接信息一个文件就是一个数据库。接收数据并存库的完整示例import sqlite3 import json from datetime import datetime from flask import Flask, request app Flask(__name__) def init_db(): conn sqlite3.connect(data.db) conn.execute( CREATE TABLE IF NOT EXISTS sensor_data ( id INTEGER PRIMARY KEY AUTOINCREMENT, device TEXT, temp REAL, humidity REAL, raw TEXT, created_at TEXT ) ) conn.commit() conn.close() app.route(/api/data, methods[POST]) def api_data(): data request.get_json(silentTrue) if data is None: return {code: 1, message: JSON解析失败}, 400 conn sqlite3.connect(data.db) conn.execute( INSERT INTO sensor_data (device, temp, humidity, raw, created_at) VALUES (?, ?, ?, ?, ?), ( data.get(device, ), data.get(temp, 0), data.get(humidity, 0), json.dumps(data, ensure_asciiFalse), datetime.now().strftime(%Y-%m-%d %H:%M:%S) ) ) conn.commit() conn.close() return {code: 0, message: ok, id: 已入库} if __name__ __main__: init_db() app.run(host0.0.0.0, port5000, debugTrue)这段代码把接收和存储串在了一起先解析JSON再插入SQLite表最后返回成功响应。SQLite单文件、无需运维特别适合先跑起来再考虑扩展的阶段。4.3 收到数据后要回话学会用HTTP状态码很多初学的人有个坏习惯服务端收到数据后不管什么情况都返回200。这会让发送方误以为数据一定处理成功了到头来数据丢了都不知道在哪丢的。我的建议是最少区分三种响应情况状态码响应体示例发送方应该怎么做接收成功处理成功200{code:0, message:ok}不要重发格式错误或参数不对400{code:1, message:设备ID不能为空}检查代码修正后重发服务端内部出错500{code:2, message:数据库写入失败}等一会重试或报障碍尤其要注意400和500的区别。我遇到过很多硬件项目设备端只会收到非200就重发结果因为服务端代码有个字段没判断导致设备疯狂重发把日志刷爆。正确做法是服务端自己把参数校验做好校验不通过明确返回400让发送方知道不要再盲目重试先去改配置。4.4 文件上传怎么接收接收数据不只是JSON偶尔还要接收文件比如设备拍的照片、App上传的日志压缩包。Flask处理这个也很直接from flask import Flask, request import os app Flask(__name__) UPLOAD_FOLDER uploads os.makedirs(UPLOAD_FOLDER, exist_okTrue) app.route(/upload, methods[POST]) def upload_file(): if file not in request.files: return {code: 1, message: 缺少file字段}, 400 file request.files[file] # file.filename是原始文件名使用时严格注意别直接用用户给的路径 safe_name os.path.basename(file.filename or unnamed) save_path os.path.join(UPLOAD_FOLDER, safe_name) file.save(save_path) return {code: 0, message: 文件已保存, path: save_path}注意我用os.path.basename把文件名取了一遍——这是安全习惯防止用户传一个包含../../的路径把你的服务器目录穿透。无论接收文件还是JSON凡是用户输入都要假设它是恶意的。5. 实测用各种方式把数据砸向你的服务端5.1 curl最快的自测方式启动服务端后我调试接口默认先上curl因为不需要打开任何图形界面一条命令完事# 发GET带参数 curl http://localhost:5000/receive?nametempvalue26.5 # 发POST JSONLinux/macOS的bash环境 curl -X POST -H Content-Type: application/json \ -d {device:esp32,temp:26.5} \ http://localhost:5000/api/data # 如果是在Windows的cmd里最外面的引号要改成双引号JSON里用单引号转义 # curl -X POST -H Content-Type: application/json -d {\device\:\esp32\,\temp\:26.5} http://localhost:5000/api/datacurl的关键参数就三个-X指定方法、-H指定请求头、-d指定body。只要这条命令能通说明服务端接收链路基本没问题。5.2 图形化工具Postman、Apifox或Apipost如果你觉得命令行不直观或者要调试的字段很多可以用图形化的接口请求工具。不管哪个配置方式都一样选择请求方法POST输入请求地址比如http://192.168.1.100:5000/api/data在Headers里加Content-Type: application/jsonJSON格式选择后很多工具会自动加在Body里选raw格式选JSON填上要发送的数据点击发送看返回结果。这类工具最大的价值是可以把一组请求参数保存下来下次直接复用。做接口测试、给别人演示接口怎么用都方便。我平时会在项目目录里保留一份导出的接口集合传给协作的硬件工程师对方打开就能调试省得来回要参数格式。5.3 在代码里模拟客户端给硬件端或前端看的示例如果你是那个需要对接别人服务端的人可以用脚本快速把测试请求发出来。Python的requests库最直白import requests url http://localhost:5000/api/data data { device: esp32-001, temp: 26.5, humidity: 60, timestamp: 2024-01-01 12:00:00 } resp requests.post(url, jsondata, timeout5) print(resp.status_code) print(resp.json())我一般建议用json关键字参数而不是data手动指定Header因为requests会在传dict给json时自动帮你序列化并设置Content-Type: application/json少一步手动操作就少一处出错的可能。如果是浏览器端JavaScript要发数据最简写法是fetch(http://localhost:5000/api/data, { method: POST, headers: { Content-Type: application/json }, body: JSON.stringify({ device: web, temp: 22.5 }) }) .then(resp resp.json()) .then(json console.log(json));把服务端跑起来再用上面任意一种方式发数据看到服务端成功打印、返回正常响应整个接收数据的闭环就通了。6. 踩坑实录我接收数据时掉过的五个坑6.1 中文乱码要么是发送方没告诉你怎么编码要么是终端在骗你现象是服务端打印出来的Json里有中文的地方全变成\u4f60\u597d或者乱码。排查链路是这样的第一步先确认服务端收到的原始字节是什么。在接收函数里临时加一行print(repr(request.get_data()))看字节流里是不是合法的UTF-8。如果原始字节就是\\u4f60\\u597d这种转义形式那其实是好的——JSON规范允许这种写法只是Python打印时默认显示转义不代表乱码。想要显示中文在打印前加print(json.dumps(data, ensure_asciiFalse, indent2))第二步如果原始字节真的是乱码说明发送方没有按UTF-8编码。这时需要对方在请求头里显式指定Content-Type: application/json; charsetutf-8或者严格使用标准的JSON库发送数据。第三步还有一个常常被忽略的终端编码。Windows的cmd默认可能是GBK服务端程序输出UTF-8的中文到了终端里就会显示乱码。这时不是数据错了是显示错了。可以在终端执行chcp 65001切到UTF-8再跑。6.2 浏览器跨域被拦curl没问题前端用的fetch却报CORS错误这个坑几乎是每个做前后端分离的人都会遇到的服务端curl测得好好的前端一调用就报blocked by CORS policy。原因是浏览器的安全策略只有同协议、同域名、同端口的页面才允许访问你的接口。你的前端跑在http://localhost:5173服务端跑在http://localhost:5000端口不同就算跨域。排查链路确认报错是否来自浏览器控制台——如果你用手机App、curl、requests发数据都没问题基本可以判定就是跨域问题在服务端加CORS支持最简单的方式是装Flask的扩展pip install flask-cors然后在代码初始化的时候加一行from flask_cors import CORS CORS(app)CORS(app)默认允许所有来源访问开发调试够用。生产环境要收紧的话可以指定origins[http://your-frontend-domain.com]。如果是生产环境不想引入额外依赖也可以自己写一个after_request钩子给所有响应都加上Access-Control-Allow-Origin头。但既然Flask-Cors是官方维护的扩展直接用扩展是最省事的。注意CORS是浏览器的行为不是服务端的强制限制。加了CORS头不代表任何人都能跨域访问你的接口服务端该做的鉴权和校验不能省。6.3 端口被占用Error: Address already in use现象很明确启动服务时直接报OSError: [Errno 98] Address already in use或 Windows 下类似提示。原因不外乎三种端口被上一个没关干净的进程占用有其他程序恰好用了这个端口服务端自己把自己重复启动了因为Flask的debug模式在某些编辑器里会启动两个进程。排查命令Linux/macOSlsof -i :5000Windows:netstat -ano | findstr :5000找到占用端口的进程ID后按需杀进程或换端口。我个人更建议调试时固定一个唯一端口比如 5000到了部署阶段服务数量多了容易撞再统一规划。另外提醒一点Flask的debug模式会启动一个额外的reloader进程你改了代码它会自动重启服务。如果这个reloader没被正确回收可能出现我明明改了代码但访问的还是老版本的情况。排查时用ps看一下有没有多个python进程有的话全部终止再重新启动。6.4 数据收到了但不完整body只有一半现象是服务端拿到的JSON缺少后半截解析直接失败。这种情况我在对接某些定制化硬件时遇到过几次多数原因出在客户端要么没有设置正确的Content-Length头要么是采用流式chunked传输但发送方没有按规范结束要么是超时设置太短body还没发完连接就被掐断了。排查链路在服务端用request.get_data(cacheTrue)拿原始body看实际长度对比请求头里客户端声明的Content-Length不一致就是客户端的问题如果确认是客户端流式发送服务端可以设置app.config[MAX_CONTENT_LENGTH]定义上限超过就直接拒绝避免收不完整还硬解析。Flask里设置请求体大小上限的写法app.config[MAX_CONTENT_LENGTH] 1024 * 1024 # 限制1MB超过这个大小的请求Flask会直接返回413发送方就知道自己发的东西超过限制了而不是收到一个模糊的解析失败。6.5 服务端一收到脏数据就崩最常见的崩溃是KeyError或TypeError或者body解析出错。比如客户端该传temp字段结果没传你代码里直接data[temp]就会抛异常返回一个500和一堆堆栈。解决思路很简单分三步取值用.get()并给默认值不要用[]下标方式将解析和业务处理包在 try-except 里捕获异常后返回结构化错误信息在日志中记录原始body这样哪怕出了错也能从日志里看到当时到底收到了什么。改造后的接收函数app.route(/api/data, methods[POST]) def api_data(): raw request.get_data(as_textTrue) data request.get_json(silentTrue) if data is None: logger.warning(fJSON解析失败, 原始数据: {raw}) return {code: 1, message: JSON解析失败}, 400 try: device data.get(device, unknown) temp float(data.get(temp, 0)) except ValueError: logger.warning(f字段类型错误, 原始数据: {raw}) return {code: 2, message: temp必须是数字}, 400 # 继续业务处理... return {code: 0, message: ok, device: device, temp: temp}这套写法保证了两件事任何异常情况下客户端拿到的都是能看懂的错误信息服务端不会因为一条脏数据就整个挂掉。7. 从能接收到像一个正经服务端7.1 关闭debug模式明确绑定地址debugTrue在开发阶段很好用能实时重载代码、显示详细报错。但也意味着任何人访问你的地址触发了一个未捕获的异常都能看到一个包含源码片段和堆栈的调试页面这在生产环境就是信息泄露。启动时把它关掉if __name__ __main__: # 生产环境不要开debug app.run(host0.0.0.0, port5000, debugFalse)关于host0.0.0.0它的含义是监听所有网卡上的请求。部署在服务器上时这样写才能让局域网或公网访问到。但如果只是在你自己电脑上调试建议用127.0.0.1别把服务暴露给局域网里的其他设备。7.2 让服务端在后台常驻运行用python3 server.py启动服务一旦关闭终端服务就跟着停了。临时跑个Demo无所谓但如果你要它长期收数据需要让它脱离终端运行。Linux服务器上最简单的做法nohup python3 server.py server.log 21 这样启动后关掉SSH会话服务还在跑。不过更规范、也推荐的做法是交给systemd管理这样服务崩溃了能自动重启、开机自启也方便。写一个服务单元文件比如/etc/systemd/system/my-server.service[Unit] DescriptionMy Data Receiver [Service] ExecStart/usr/bin/python3 /path/to/server.py Restartalways Useryour_user [Install] WantedBymulti-user.target然后sudo systemctl daemon-reload sudo systemctl start my-server sudo systemctl enable my-server在开发环境不想管systemd的可以直接用上面的nohup方案。至于生产环境要不要引入gunicorn这类WSGI服务器这个要看流量和需求。如果只是每秒接收几次上报Flask自带的服务器完全够用真要扛高并发再考虑上gunicorn可以做进程多开。7.3 下一步还能怎么扩展接收数据只是第一步链路往后延伸的空间很大。根据我自己的经验按优先级排列加校验和鉴权至少用一个请求头里的Token来识别发送方是不是你自己。设备上报场景更要注意别让任何人POST点假数据就把你的库写满。数据转发接收端可以再作为一个客户端把数据推给消息队列、其他内部服务配合第三方队列库或一个简单的HTTP转发就能做。接口文档给别人对接的时候一份说明清楚路径、方法、请求头、body格式、返回格式的文档能省很多沟通成本。先写个最少三行的README说明用POST还是GET、JSON字段叫什么就比让对方猜强得多。健康检查接口加一个/ping返回pong方便别人确认你的服务还活着。负载均衡和监控脚本都会用到这个。我个人的体会是接收数据这个功能90%的项目根本不需要一开始就用企业级框架。一个Flask应用、一张表、一个日志文件足够对付绝大多数中小型需求。真正要去琢磨架构和高性能的时候说明你的服务每天收到的数据量已经大到需要认真对待了——到那个阶段再迁移成本完全可控因为接收数据的逻辑本来就该保持简单。 用Flask做最简单的服务端接收数据：从零搭建HTTP接口
我最早正经写服务端接收数据是因为要给一组环境传感器做数据上报。当时第一反应是找个现成的后端框架翻了半天企业级框架的教程被那一堆配置直接劝退——程序能不能跑起来另说光理解为什么一个接收数据的工程需要这么多文件就够头疼。后来我换了个思路我其实只需要一个一直运行的程序监听某个端口等数据进来收下来存起来再给发送方回一句收到了。就这点事不夸张地说Python标准库就能做到换Flask写也就三十行代码。这篇文章就把这套最简单服务端从头到尾讲透服务端接收数据到底接收的是什么、不同场景怎么选方案、代码怎么写、怎么自测以及我在实际项目里被数据接收坑过的几个点。内容不追求大而全追求的是你照着做十分钟内能跑通一个真正能收数据的服务端。适合刚接触前后端联调、在做小项目、搞硬件上报或者只是临时想验证某个接口的人。1. 先把接收数据这件事拆明白1.1 我们口中的接收数据到底是在接收什么很多人一听服务端接收数据脑子里浮现的是后台管理系统、数据库集群、消息队列这些东西。但实际上绝大多数场景根本没到那个复杂度。我把遇到过的接收数据需求归了下类基本都是下面几种网页表单提交前端页面填了个表单点提交后端要拿到这些字段存下来。App或小程序上报手机端采集了用户行为、定位、日志定时往服务端发送。硬件设备上报温湿度传感器、GPS追踪器、摄像头状态通过HTTP接口周期性上报数据。系统间回调通知支付平台、短信平台在某个事件发生后回调你的接口告诉你订单已支付。脚本或爬虫抓取结果另一个脚本跑完任务后把结果POST给你。这些场景有个共同点都是有一个东西主动把数据发给你运行的程序。你的程序不需要去连接别人只需要老老实实在一个端口上等着。所以接收数据这个动作本质上就三件事持续监听某个网络端口收到请求后把请求里的数据解析出来给发送方一个明确的响应。仅此而已。至于数据库、事务、权限这些都是收到数据之后的事不属于接收这个动作本身。1.2 一次请求从发出到被接收中间发生了什么为了后面调试方便这里有必要把链路简单过一遍。你可以把它理解成去餐厅吃饭客户端浏览器、手机、传感器就是顾客它知道你的服务端地址IP加端口相当于知道饭馆开在哪条街几号。顾客进门喊服务员发起HTTP请求这一步走的是你服务端正在监听的端口比如5000。你的程序就是服务员听到喊声后要看顾客点了什么菜读取请求里的数据然后去后厨做业务处理最后把菜端上桌返回响应。在这个过程里几个关键要素要记住请求行POST /receive HTTP/1.1 请求头Content-Type: application/json 告诉服务端数据是什么格式 请求体{temp: 26.5, humidity: 60} 真正要接收的数据服务端接收数据做的就是两件事解析请求头按格式读请求体。至于0.0.0.0这类监听地址、5000这种端口号都是你开启监听时需要先理解的参数——监听地址决定谁能连、端口决定连哪里。1.3 GET和POST两种最常见的递纸条方式初学者最容易混淆的就是GET和POST的区别。其实从服务端接收数据的角度看区别非常直观方式数据放在哪里常见用途特点GETURL末尾的查询字符串里比如?nameabcvalue123查询、调试、参数简单可被浏览器地址栏直接访问会留在历史记录里POSTHTTP请求体的body里提交表单、上报JSON数据、文件上传数据不在URL中可传大容量和复杂结构举个例子同样是上报一条温度数据GET方式 http://localhost:5000/receive?deviceesp32temp26.5 POST方式 http://localhost:5000/receive body: {device: esp32, temp: 26.5}如果你只是自己调试GET快得多浏览器里直接打开链接就能触发。但准备正式接收业务数据我建议一律用POST原因有三点数据结构可以复杂一点点、长度不受URL限制、不会因为数据里的特殊字符导致URL解析出错。提示根本就不存在所谓的GET更安全这种说法。真正的数据保密靠的是HTTPS而不是靠把数据塞进body里掩耳盗铃。2. 别一上来就整企业级框架先选一个真简单的方案2.1 常见方案横向对比简单这两个字不同背景的人理解不同。一个Java后端朋友觉得Spring Boot已经很简单了但对一个刚转行、或者只是搞个硬件Demo的人来说光把Maven依赖拉下来就得折腾半天。我按从零开始跑通一个接收数据的服务端需要付出的成本把常见方案排了个队方案依赖启动成本适合场景短板Python内置http.server零依赖最低Python自带临时调试、一次性任务要手动解析报文不适合做业务Python Flask需装Flask低一个pip命令快速原型、小型项目、硬件上报性能上限不如专门的高性能框架Node.js Express需装Node依赖低前端顺手、Node环境已有回调风格需要熟悉Go 标准库net/http零第三方依赖低要单文件部署、追求性能语法对新手略陌生Java Spring Boot重高大型团队、复杂业务很多人就是被它劝退的你可能会问既然Python内置的http.server零依赖为什么不直接用它因为它把接收数据这件事做成了最原始的样子收到的是一个完整HTTP报文你要从字符串里手动抠出请求头、body、参数还得自己处理url解码、JSON解析。写个Demo没问题做正经功能就是在给自己挖坑。2.2 为什么我推荐Flask作为起步方案Flask是一个微框架Micro Framework所谓微指的是它只解决你递纸条的核心需求路由、请求解析、响应返回。它不需要你理解依赖注入中间件配置中心这些花哨的概念。我用Flask已经很多年总结它的优势就三条路由写起来像在报菜名。app.route(/receive, methods[POST])底下直接跟处理的函数一眼就能看懂哪个路径对应哪段逻辑。请求数据解析是现成的。request.args拿GET参数、request.get_json()拿JSON、request.form拿表单数据不用自己写解析逻辑。返回JSON也方便。直接返回一个dictFlask自动帮你序列化并加上对应的Content-Type头。还有一个不太被提及但很现实的好处Flask的教程、问答、现成代码是Python后端框架里最多的。你遇到任何小问题比如Flask接收不到POST的body搜一下基本就有答案。对于新手或者做快速验证的人来说这个生态价值比框架本身的性能重要得多。注意如果你确实只是想在终端里快速看一眼某个地址发来的原始请求长什么样那用http.server反而快。在Python环境里执行python -m http.server 8000浏览器访问一下终端会输出请求日志。这个技巧可以用来应急。3. 三十行代码搭出第一个能收数据的服务端3.1 环境准备就两步先确认Python已经装好在终端里执行python3 --version看到版本号之后安装Flaskpip install flask国内网络如果下载慢可以临时换一下Python软件源但不要为了图快装来路不明的包。装完验证一下python3 -c import flask; print(flask.__version__)没报错就是装好了。3.2 第一版把GET参数原样打印出来创建一个文件server.py先写个最基础的版本from flask import Flask, request app Flask(__name__) app.route(/receive, methods[GET]) def receive(): name request.args.get(name, ) value request.args.get(value, ) print(收到GET请求, name, value) return ok if __name__ __main__: app.run(host0.0.0.0, port5000, debugTrue)这段代码要拆开看app Flask(__name__)创建了一个Flask应用后面所有路由都挂在它上。app.route(/receive, methods[GET])声明了路径和方法只有访问/receive并且是GET下面的函数才会执行。request.args是所有GET查询参数的集合request.args.get(name)拿到?name后面的值。app.run(host0.0.0.0, port5000)启动监听0.0.0.0表示允许局域网内其他设备访问如果只想本机访问就改成127.0.0.1。启动python3 server.py然后另开一个终端用curl模拟发数据curl http://localhost:5000/receive?nametempvalue26.5回到服务端终端你会看到打印出收到GET请求 temp 26.5到这里你的服务端已经真正意义上接收到了数据。虽然是GET但骨架已经有了。3.3 第二版接收POST JSON数据实际业务里最常遇到的就是客户端POST一个JSON过来。把代码升级一下加点容错from flask import Flask, request app Flask(__name__) app.route(/api/data, methods[POST]) def api_data(): # silentTrue: 如果body不是合法JSON不会抛异常而是返回None data request.get_json(silentTrue) if data is None: return {code: 1, message: JSON解析失败}, 400 print(收到JSON数据, data) return {code: 0, message: ok, received: data} if __name__ __main__: app.run(host0.0.0.0, port5000, debugTrue)这次的改动要点request.get_json(silentTrue)直接解析请求体里的JSON。silentTrue这一步很重要——如果客户端发了个我不是JSON的东西过来框架会抛一个400错误但直接用它会在调试时看到一屏幕堆栈。你更希望的是拿到一个None然后自己决定怎么回复。返回的是字典Flask会自动把它变成JSON字符串还会自动设置Content-Type: application/json。返回400状态码语义上告诉客户端你发的东西不合规范。测试命令curl -X POST \ -H Content-Type: application/json \ -d {device:esp32,temp:26.5,humidity:60} \ http://localhost:5000/api/data服务端打印收到JSON数据 {device: esp32, temp: 26.5, humidity: 60}curl返回{code:0,message:ok,received:{device:esp32,temp:26.5,humidity:60}}3.4 第三版一个统一入口GET、POST、JSON、表单全接住真实的接收场景往往没有你想的那么规范。有可能前端一会儿用JSON一会儿用表单也有可能是你临时给别的部门同学调试对方习惯用GET带参数。我的做法是做一个通用接收器不管什么格式先接住把内容打印出来并原样返回from flask import Flask, request app Flask(__name__) app.route(/receive, methods[GET, POST]) def receive_all(): # 记录请求方法 method request.method # 1. GET查询参数比如 /receive?a1b2 query_params dict(request.args) # 2. 判断Content-Type决定怎么解析body content_type request.headers.get(Content-Type, ) body_data None if application/json in content_type: body_data request.get_json(silentTrue) elif application/x-www-form-urlencoded in content_type: body_data dict(request.form) else: # 其他类型先按原文读出来 body_data request.get_data(as_textTrue) print(f[{method}] 查询参数: {query_params}) print(f[{method}] body内容: {body_data}) return { method: method, query_params: query_params, body: body_data, status: received } if __name__ __main__: app.run(host0.0.0.0, port5000, debugTrue)这个版本好在哪它把接收和业务处理解耦了。不管你收到什么先把内容记录清楚调试的时候一目了然。后面想接数据库也好、转发给别的服务也好都是在拿到body_data之后再加逻辑。提示request.get_data(as_textTrue)是最后的兜底方案它把原始body当字符串读出来。碰到文件上传时这个不适用文件要用request.files这点后面单独说。4. 数据收到之后拿它干什么4.1 别用print改用日志我最早偷懒直接print结果服务端跑了一晚上第二天想查昨晚哪些设备上报了数据终端翻半天全被滚动刷没了。你要记住print 是给调试用的日志是给运行用的。先用Python自带的logging模块把日志写到文件里import logging from logging.handlers import RotatingFileHandler # 配置日志按大小轮转单个文件超过1MB自动切分 handler RotatingFileHandler(server.log, maxBytes1024*1024, backupCount5) formatter logging.Formatter(%(asctime)s - %(levelname)s - %(message)s) handler.setFormatter(formatter) logger logging.getLogger(my_server) logger.setLevel(logging.INFO) logger.addHandler(handler)然后在接收函数里把原来的print替换成logger.info(f收到请求: {method} 查询参数: {query_params} body: {body_data})这样数据接收完服务端程序爱怎么重启都没事日志文件里什么都在。4.2 把数据存进SQLite不用装任何数据库很多时候接收完数据是要落库的。对小型项目、个人工具、硬件上报这类场景我强烈推荐直接用Python标准库自带的SQLite——不用装MySQL不用配置连接信息一个文件就是一个数据库。接收数据并存库的完整示例import sqlite3 import json from datetime import datetime from flask import Flask, request app Flask(__name__) def init_db(): conn sqlite3.connect(data.db) conn.execute( CREATE TABLE IF NOT EXISTS sensor_data ( id INTEGER PRIMARY KEY AUTOINCREMENT, device TEXT, temp REAL, humidity REAL, raw TEXT, created_at TEXT ) ) conn.commit() conn.close() app.route(/api/data, methods[POST]) def api_data(): data request.get_json(silentTrue) if data is None: return {code: 1, message: JSON解析失败}, 400 conn sqlite3.connect(data.db) conn.execute( INSERT INTO sensor_data (device, temp, humidity, raw, created_at) VALUES (?, ?, ?, ?, ?), ( data.get(device, ), data.get(temp, 0), data.get(humidity, 0), json.dumps(data, ensure_asciiFalse), datetime.now().strftime(%Y-%m-%d %H:%M:%S) ) ) conn.commit() conn.close() return {code: 0, message: ok, id: 已入库} if __name__ __main__: init_db() app.run(host0.0.0.0, port5000, debugTrue)这段代码把接收和存储串在了一起先解析JSON再插入SQLite表最后返回成功响应。SQLite单文件、无需运维特别适合先跑起来再考虑扩展的阶段。4.3 收到数据后要回话学会用HTTP状态码很多初学的人有个坏习惯服务端收到数据后不管什么情况都返回200。这会让发送方误以为数据一定处理成功了到头来数据丢了都不知道在哪丢的。我的建议是最少区分三种响应情况状态码响应体示例发送方应该怎么做接收成功处理成功200{code:0, message:ok}不要重发格式错误或参数不对400{code:1, message:设备ID不能为空}检查代码修正后重发服务端内部出错500{code:2, message:数据库写入失败}等一会重试或报障碍尤其要注意400和500的区别。我遇到过很多硬件项目设备端只会收到非200就重发结果因为服务端代码有个字段没判断导致设备疯狂重发把日志刷爆。正确做法是服务端自己把参数校验做好校验不通过明确返回400让发送方知道不要再盲目重试先去改配置。4.4 文件上传怎么接收接收数据不只是JSON偶尔还要接收文件比如设备拍的照片、App上传的日志压缩包。Flask处理这个也很直接from flask import Flask, request import os app Flask(__name__) UPLOAD_FOLDER uploads os.makedirs(UPLOAD_FOLDER, exist_okTrue) app.route(/upload, methods[POST]) def upload_file(): if file not in request.files: return {code: 1, message: 缺少file字段}, 400 file request.files[file] # file.filename是原始文件名使用时严格注意别直接用用户给的路径 safe_name os.path.basename(file.filename or unnamed) save_path os.path.join(UPLOAD_FOLDER, safe_name) file.save(save_path) return {code: 0, message: 文件已保存, path: save_path}注意我用os.path.basename把文件名取了一遍——这是安全习惯防止用户传一个包含../../的路径把你的服务器目录穿透。无论接收文件还是JSON凡是用户输入都要假设它是恶意的。5. 实测用各种方式把数据砸向你的服务端5.1 curl最快的自测方式启动服务端后我调试接口默认先上curl因为不需要打开任何图形界面一条命令完事# 发GET带参数 curl http://localhost:5000/receive?nametempvalue26.5 # 发POST JSONLinux/macOS的bash环境 curl -X POST -H Content-Type: application/json \ -d {device:esp32,temp:26.5} \ http://localhost:5000/api/data # 如果是在Windows的cmd里最外面的引号要改成双引号JSON里用单引号转义 # curl -X POST -H Content-Type: application/json -d {\device\:\esp32\,\temp\:26.5} http://localhost:5000/api/datacurl的关键参数就三个-X指定方法、-H指定请求头、-d指定body。只要这条命令能通说明服务端接收链路基本没问题。5.2 图形化工具Postman、Apifox或Apipost如果你觉得命令行不直观或者要调试的字段很多可以用图形化的接口请求工具。不管哪个配置方式都一样选择请求方法POST输入请求地址比如http://192.168.1.100:5000/api/data在Headers里加Content-Type: application/jsonJSON格式选择后很多工具会自动加在Body里选raw格式选JSON填上要发送的数据点击发送看返回结果。这类工具最大的价值是可以把一组请求参数保存下来下次直接复用。做接口测试、给别人演示接口怎么用都方便。我平时会在项目目录里保留一份导出的接口集合传给协作的硬件工程师对方打开就能调试省得来回要参数格式。5.3 在代码里模拟客户端给硬件端或前端看的示例如果你是那个需要对接别人服务端的人可以用脚本快速把测试请求发出来。Python的requests库最直白import requests url http://localhost:5000/api/data data { device: esp32-001, temp: 26.5, humidity: 60, timestamp: 2024-01-01 12:00:00 } resp requests.post(url, jsondata, timeout5) print(resp.status_code) print(resp.json())我一般建议用json关键字参数而不是data手动指定Header因为requests会在传dict给json时自动帮你序列化并设置Content-Type: application/json少一步手动操作就少一处出错的可能。如果是浏览器端JavaScript要发数据最简写法是fetch(http://localhost:5000/api/data, { method: POST, headers: { Content-Type: application/json }, body: JSON.stringify({ device: web, temp: 22.5 }) }) .then(resp resp.json()) .then(json console.log(json));把服务端跑起来再用上面任意一种方式发数据看到服务端成功打印、返回正常响应整个接收数据的闭环就通了。6. 踩坑实录我接收数据时掉过的五个坑6.1 中文乱码要么是发送方没告诉你怎么编码要么是终端在骗你现象是服务端打印出来的Json里有中文的地方全变成\u4f60\u597d或者乱码。排查链路是这样的第一步先确认服务端收到的原始字节是什么。在接收函数里临时加一行print(repr(request.get_data()))看字节流里是不是合法的UTF-8。如果原始字节就是\\u4f60\\u597d这种转义形式那其实是好的——JSON规范允许这种写法只是Python打印时默认显示转义不代表乱码。想要显示中文在打印前加print(json.dumps(data, ensure_asciiFalse, indent2))第二步如果原始字节真的是乱码说明发送方没有按UTF-8编码。这时需要对方在请求头里显式指定Content-Type: application/json; charsetutf-8或者严格使用标准的JSON库发送数据。第三步还有一个常常被忽略的终端编码。Windows的cmd默认可能是GBK服务端程序输出UTF-8的中文到了终端里就会显示乱码。这时不是数据错了是显示错了。可以在终端执行chcp 65001切到UTF-8再跑。6.2 浏览器跨域被拦curl没问题前端用的fetch却报CORS错误这个坑几乎是每个做前后端分离的人都会遇到的服务端curl测得好好的前端一调用就报blocked by CORS policy。原因是浏览器的安全策略只有同协议、同域名、同端口的页面才允许访问你的接口。你的前端跑在http://localhost:5173服务端跑在http://localhost:5000端口不同就算跨域。排查链路确认报错是否来自浏览器控制台——如果你用手机App、curl、requests发数据都没问题基本可以判定就是跨域问题在服务端加CORS支持最简单的方式是装Flask的扩展pip install flask-cors然后在代码初始化的时候加一行from flask_cors import CORS CORS(app)CORS(app)默认允许所有来源访问开发调试够用。生产环境要收紧的话可以指定origins[http://your-frontend-domain.com]。如果是生产环境不想引入额外依赖也可以自己写一个after_request钩子给所有响应都加上Access-Control-Allow-Origin头。但既然Flask-Cors是官方维护的扩展直接用扩展是最省事的。注意CORS是浏览器的行为不是服务端的强制限制。加了CORS头不代表任何人都能跨域访问你的接口服务端该做的鉴权和校验不能省。6.3 端口被占用Error: Address already in use现象很明确启动服务时直接报OSError: [Errno 98] Address already in use或 Windows 下类似提示。原因不外乎三种端口被上一个没关干净的进程占用有其他程序恰好用了这个端口服务端自己把自己重复启动了因为Flask的debug模式在某些编辑器里会启动两个进程。排查命令Linux/macOSlsof -i :5000Windows:netstat -ano | findstr :5000找到占用端口的进程ID后按需杀进程或换端口。我个人更建议调试时固定一个唯一端口比如 5000到了部署阶段服务数量多了容易撞再统一规划。另外提醒一点Flask的debug模式会启动一个额外的reloader进程你改了代码它会自动重启服务。如果这个reloader没被正确回收可能出现我明明改了代码但访问的还是老版本的情况。排查时用ps看一下有没有多个python进程有的话全部终止再重新启动。6.4 数据收到了但不完整body只有一半现象是服务端拿到的JSON缺少后半截解析直接失败。这种情况我在对接某些定制化硬件时遇到过几次多数原因出在客户端要么没有设置正确的Content-Length头要么是采用流式chunked传输但发送方没有按规范结束要么是超时设置太短body还没发完连接就被掐断了。排查链路在服务端用request.get_data(cacheTrue)拿原始body看实际长度对比请求头里客户端声明的Content-Length不一致就是客户端的问题如果确认是客户端流式发送服务端可以设置app.config[MAX_CONTENT_LENGTH]定义上限超过就直接拒绝避免收不完整还硬解析。Flask里设置请求体大小上限的写法app.config[MAX_CONTENT_LENGTH] 1024 * 1024 # 限制1MB超过这个大小的请求Flask会直接返回413发送方就知道自己发的东西超过限制了而不是收到一个模糊的解析失败。6.5 服务端一收到脏数据就崩最常见的崩溃是KeyError或TypeError或者body解析出错。比如客户端该传temp字段结果没传你代码里直接data[temp]就会抛异常返回一个500和一堆堆栈。解决思路很简单分三步取值用.get()并给默认值不要用[]下标方式将解析和业务处理包在 try-except 里捕获异常后返回结构化错误信息在日志中记录原始body这样哪怕出了错也能从日志里看到当时到底收到了什么。改造后的接收函数app.route(/api/data, methods[POST]) def api_data(): raw request.get_data(as_textTrue) data request.get_json(silentTrue) if data is None: logger.warning(fJSON解析失败, 原始数据: {raw}) return {code: 1, message: JSON解析失败}, 400 try: device data.get(device, unknown) temp float(data.get(temp, 0)) except ValueError: logger.warning(f字段类型错误, 原始数据: {raw}) return {code: 2, message: temp必须是数字}, 400 # 继续业务处理... return {code: 0, message: ok, device: device, temp: temp}这套写法保证了两件事任何异常情况下客户端拿到的都是能看懂的错误信息服务端不会因为一条脏数据就整个挂掉。7. 从能接收到像一个正经服务端7.1 关闭debug模式明确绑定地址debugTrue在开发阶段很好用能实时重载代码、显示详细报错。但也意味着任何人访问你的地址触发了一个未捕获的异常都能看到一个包含源码片段和堆栈的调试页面这在生产环境就是信息泄露。启动时把它关掉if __name__ __main__: # 生产环境不要开debug app.run(host0.0.0.0, port5000, debugFalse)关于host0.0.0.0它的含义是监听所有网卡上的请求。部署在服务器上时这样写才能让局域网或公网访问到。但如果只是在你自己电脑上调试建议用127.0.0.1别把服务暴露给局域网里的其他设备。7.2 让服务端在后台常驻运行用python3 server.py启动服务一旦关闭终端服务就跟着停了。临时跑个Demo无所谓但如果你要它长期收数据需要让它脱离终端运行。Linux服务器上最简单的做法nohup python3 server.py server.log 21 这样启动后关掉SSH会话服务还在跑。不过更规范、也推荐的做法是交给systemd管理这样服务崩溃了能自动重启、开机自启也方便。写一个服务单元文件比如/etc/systemd/system/my-server.service[Unit] DescriptionMy Data Receiver [Service] ExecStart/usr/bin/python3 /path/to/server.py Restartalways Useryour_user [Install] WantedBymulti-user.target然后sudo systemctl daemon-reload sudo systemctl start my-server sudo systemctl enable my-server在开发环境不想管systemd的可以直接用上面的nohup方案。至于生产环境要不要引入gunicorn这类WSGI服务器这个要看流量和需求。如果只是每秒接收几次上报Flask自带的服务器完全够用真要扛高并发再考虑上gunicorn可以做进程多开。7.3 下一步还能怎么扩展接收数据只是第一步链路往后延伸的空间很大。根据我自己的经验按优先级排列加校验和鉴权至少用一个请求头里的Token来识别发送方是不是你自己。设备上报场景更要注意别让任何人POST点假数据就把你的库写满。数据转发接收端可以再作为一个客户端把数据推给消息队列、其他内部服务配合第三方队列库或一个简单的HTTP转发就能做。接口文档给别人对接的时候一份说明清楚路径、方法、请求头、body格式、返回格式的文档能省很多沟通成本。先写个最少三行的README说明用POST还是GET、JSON字段叫什么就比让对方猜强得多。健康检查接口加一个/ping返回pong方便别人确认你的服务还活着。负载均衡和监控脚本都会用到这个。我个人的体会是接收数据这个功能90%的项目根本不需要一开始就用企业级框架。一个Flask应用、一张表、一个日志文件足够对付绝大多数中小型需求。真正要去琢磨架构和高性能的时候说明你的服务每天收到的数据量已经大到需要认真对待了——到那个阶段再迁移成本完全可控因为接收数据的逻辑本来就该保持简单。