搭建每分钟更新的HTTP服务器 求Python替代microhttpd的模块建议

你之前用C语言的microhttpd实现过每分钟更新并广播温度的HTTP服务，现在转用Python的话，我推荐几个适配性很强的方案，结合你的定时更新需求逐一说明：

1. Flask + APScheduler（快速上手首选） #

Flask是轻量级Web框架，语法简洁，非常适合快速搭建小型HTTP服务。搭配APScheduler这个专门的定时任务库，能完美实现每分钟更新数据的需求。

示例代码： #

首先安装依赖：

pip install flask apscheduler

然后编写服务代码：

from flask import Flask
from apscheduler.schedulers.background import BackgroundScheduler
import random  # 模拟温度数据更新

app = Flask(__name__)
current_temperature = 25.0  # 初始温度

def update_temperature():
    """每分钟更新温度数据"""
    global current_temperature
    current_temperature = round(random.uniform(20.0, 30.0), 1)
    print(f"Updated temperature: {current_temperature}°C")

# 启动定时任务，每分钟执行一次更新
scheduler = BackgroundScheduler()
scheduler.add_job(update_temperature, 'interval', minutes=1)
scheduler.start()

@app.route('/temperature')
def get_temperature():
    return {"temperature": current_temperature, "update_frequency": "every minute"}

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=8000, debug=False)

这个方案的优势是代码量少、学习成本低，适合快速迭代你的需求。

2. FastAPI + APScheduler（高性能进阶选择） #

FastAPI是现代、高性能的异步Web框架，支持自动生成API文档，如果你以后需要扩展服务功能（比如增加更多传感器数据接口），这个框架的扩展性更好。同样搭配APScheduler实现定时任务。

示例代码： #

安装依赖：

pip install fastapi uvicorn apscheduler

服务代码：

from fastapi import FastAPI
from apscheduler.schedulers.background import BackgroundScheduler
import random

app = FastAPI()
current_temperature = 25.0

def update_temperature():
    global current_temperature
    current_temperature = round(random.uniform(20.0, 30.0), 1)
    print(f"Updated temperature: {current_temperature}°C")

scheduler = BackgroundScheduler()
scheduler.add_job(update_temperature, 'interval', minutes=1)
scheduler.start()

@app.get('/temperature')
async def get_temperature():
    return {"temperature": current_temperature, "update_frequency": "every minute"}

if __name__ == '__main__':
    import uvicorn
    uvicorn.run(app, host='0.0.0.0', port=8000)

FastAPI的异步特性在高并发场景下表现更出色，而且自动生成的Swagger文档（访问/docs）能方便你调试接口。

3. aiohttp（纯异步轻量方案） #

如果你想要更底层的异步控制，aiohttp是纯异步的HTTP框架，适合对性能要求较高的场景。可以结合asyncio的原生定时能力或者APScheduler的异步调度器。

示例代码： #

安装依赖：

pip install aiohttp

服务代码：

from aiohttp import web
import asyncio
import random

current_temperature = 25.0

async def update_temperature():
    global current_temperature
    while True:
        current_temperature = round(random.uniform(20.0, 30.0), 1)
        print(f"Updated temperature: {current_temperature}°C")
        await asyncio.sleep(60)  # 每分钟更新一次

async def get_temperature(request):
    return web.json_response({"temperature": current_temperature, "update_frequency": "every minute"})

async def main():
    app = web.Application()
    app.add_routes([web.get('/temperature', get_temperature)])
    # 启动定时任务
    asyncio.create_task(update_temperature())
    runner = web.AppRunner(app)
    await runner.setup()
    site = web.TCPSite(runner, '0.0.0.0', 8000)
    await site.start()
    print("Server started at http://0.0.0.0:8000")
    await asyncio.Event().wait()

if __name__ == '__main__':
    asyncio.run(main())

这个方案完全基于异步IO，没有额外的定时任务库依赖，适合喜欢自己控制流程的开发者。

4. 标准库http.server（极简无依赖方案） #

如果你的需求极其简单，甚至不想安装任何第三方库，可以用Python标准库的http.server，配合线程来实现定时更新：

示例代码： #

from http.server import BaseHTTPRequestHandler, HTTPServer
import threading
import time
import random

current_temperature = 25.0

def update_temperature():
    global current_temperature
    while True:
        current_temperature = round(random.uniform(20.0, 30.0), 1)
        print(f"Updated temperature: {current_temperature}°C")
        time.sleep(60)

class TemperatureHandler(BaseHTTPRequestHandler):
    def do_GET(self):
        if self.path == '/temperature':
            self.send_response(200)
            self.send_header('Content-type', 'application/json')
            self.end_headers()
            response = f'{{"temperature": {current_temperature}, "update_frequency": "every minute"}}'
            self.wfile.write(response.encode('utf-8'))
        else:
            self.send_response(404)
            self.end_headers()

if __name__ == '__main__':
    # 启动定时更新线程
    update_thread = threading.Thread(target=update_temperature, daemon=True)
    update_thread.start()

    server_address = ('0.0.0.0', 8000)
    httpd = HTTPServer(server_address, TemperatureHandler)
    print("Server started at http://0.0.0.0:8000")
    try:
        httpd.serve_forever()
    except KeyboardInterrupt:
        httpd.shutdown()

这个方案不需要任何第三方依赖，适合快速验证原型，但功能扩展性较差。

总结一下：如果是快速搭建选Flask，追求高性能和扩展性选FastAPI，喜欢异步底层控制选aiohttp，极简无依赖用标准库http.server。

内容的提问来源于stack exchange，提问作者Hua Ye