如何实现HTTP服务器推送？Python服务端推送如何感知新数据可用？

阿华AIGC实验室

2026-5-21

好问题！其实在Python里做服务器推送时，怎么感知新数据可用，确实是很多人刚接触时会困惑的点——毕竟Python不像有些语言自带标准化事件系统，但咱们有不少实用的替代方案，核心就是围绕「事件触发」或者「状态监听」来做，我给你拆解几种常见的实现思路：

一、用消息队列/中间件实现事件触发

这是分布式场景下最常用的方案，比如Redis的Pub/Sub、RabbitMQ这类中间件。当新数据产生时，生产者把数据推到队列/频道里，你的推送控制器订阅这个频道，一旦有新消息进来就立刻触发发送逻辑。

举个简单的Redis Pub/Sub示例（用Flask做HTTP服务）：

import redis
from flask import Flask, Response

app = Flask(__name__)
# 连接Redis
r = redis.Redis(host='localhost', port=6379)
pubsub = r.pubsub()
# 订阅新数据频道
pubsub.subscribe('new_data_updates')

@app.route('/stream')
def server_sent_events():
    def generate_events():
        # 阻塞监听频道消息
        for message in pubsub.listen():
            if message['type'] == 'message':
                # 用Server-Sent Events格式返回数据
                yield f"data: {message['data'].decode('utf-8')}\n\n"
    return Response(generate_events(), mimetype='text/event-stream')

# 模拟数据生产者接口
@app.route('/push_data/<content>')
def push_data(content):
    # 往频道发布新数据
    r.publish('new_data_updates', content)
    return "数据已推送！"

if __name__ == '__main__':
    app.run(debug=True)

这里pubsub.listen()会一直阻塞，直到有新数据被发布到new_data_updates频道，这时就会立刻把数据推给客户端。

二、单进程/多线程场景：用条件变量做事件通知

如果你的服务不需要分布式，只是单进程或多线程运行，可以用Python标准库的threading.Condition来实现轻量的事件通知。它相当于一个“信号开关”，生产者产生数据时发出信号，等待的请求处理线程就会被唤醒。

示例代码：

from flask import Flask, jsonify
from threading import Condition, Thread
import time

app = Flask(__name__)
# 存储新数据的线程安全列表
pending_data = []
# 条件变量，用于通知等待的线程
data_condition = Condition()

def mock_data_producer():
    """模拟定时产生新数据的生产者线程"""
    while True:
        time.sleep(5)
        new_content = f"新数据：{time.ctime()}"
        with data_condition:
            pending_data.append(new_content)
            # 通知所有等待的线程：有新数据了！
            data_condition.notify_all()

# 启动生产者线程（后台运行）
Thread(target=mock_data_producer, daemon=True).start()

@app.route('/long_poll')
def long_polling():
    with data_condition:
        # 阻塞等待，直到pending_data里有数据
        data_condition.wait_for(lambda: len(pending_data) > 0)
        # 取出所有新数据并清空列表
        send_data = pending_data.copy()
        pending_data.clear()
        return jsonify({"data": send_data})

if __name__ == '__main__':
    app.run(debug=True)

这里data_condition.wait_for()会让请求线程进入休眠，直到生产者调用notify_all()唤醒它，这样就能及时响应新数据，避免无效的轮询。

三、WebSocket框架：自带事件推送机制

如果用WebSocket实现推送，很多成熟的框架已经帮你封装好了底层的事件逻辑，比如Flask-SocketIO或者websockets库，你只需要在数据产生时调用推送接口就行。

举个Flask-SocketIO的例子：

from flask import Flask
from flask_socketio import SocketIO, emit
import time
from threading import Thread

app = Flask(__name__)
# 初始化SocketIO，允许跨域
socketio = SocketIO(app, cors_allowed_origins="*")

def mock_data_producer():
    while True:
        time.sleep(5)
        new_content = f"WebSocket推送：{time.ctime()}"
        # 广播新数据给所有连接的客户端
        socketio.emit('new_data_arrived', {'content': new_content}, broadcast=True)

Thread(target=mock_data_producer, daemon=True).start()

# 监听客户端连接事件
@socketio.on('connect')
def handle_client_connect():
    print("有客户端连接了！")

if __name__ == '__main__':
    socketio.run(app, debug=True)

框架内部已经维护了客户端连接池，当你调用emit()时，会自动把数据推送给所有在线客户端，完全不需要自己处理等待逻辑。

四、监听文件/数据库的变更

如果你的数据来自文件或者数据库，还可以用工具监听变更：