Flask-SocketIO搭配Gunicorn部署Koyeb云端后无法工作求助

问题概述 #

开发了一款处理第三方POST请求的REST API测试应用，本地PyCharm运行时一切正常：POST请求可触发前端SocketIO监听函数，实时更新页面元素。但部署到Koyeb平台（强制使用Gunicorn WSGI服务器）后，POST请求能正常抵达后端并执行对应逻辑，却无法通过SocketIO推送更新到前端。

相关代码与配置 #

app.py核心代码 #

import eventlet
eventlet.monkey_patch()

from flask import Flask, request, jsonify
from flask_sqlalchemy import SQLAlchemy
from flask_socketio import SocketIO, emit

app = Flask(__name__)
socketio = SocketIO(app, async_mode='eventlet')

# 数据库模型（省略其他代码）
class ApiDb(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    nrVisits = db.Column(db.Integer, default=0)

@app.route("/endpoint", methods=['GET', 'POST'])
def endpoint():
    if request.method == 'POST':
        try:
            dbVisits = ApiDb.query.filter_by(id=1).first()
            dbVisits.nrVisits += 1
        except:
            db.session.add(ApiDb(id=1, nrVisits=1))
        db.session.commit()

        handle_trigger_update()
        return {'odgovor': 'OK'}

@socketio.on('trigger_update')
def handle_trigger_update():
    dbVisits = ApiDb.query.filter_by(id=1).first()
    total = dbVisits.nrVisits
    socketio.emit('update', {'total': total})

@app.route('/trigger_backend')
def trigger_backend():
    handle_trigger_update()
    return 'Backend function triggered successfully'

if __name__=='__main__':
    wsgi.server(eventlet.listen(('', 8000)), app)

前端HTML核心代码 #

<script src="https://cdnjs.cloudflare.com/ajax/libs/socket.io/4.3.1/socket.io.js" integrity="sha512-Y5MDU6RaF5h5HE5BgqJlKkV12kbkbIgWHutcT+XHHNOUzr+HHjWZGC02sqEguuPglmFms3cc08WH2PhQ5rF8Cw==" crossorigin="anonymous"></script>

<script>
    document.addEventListener('DOMContentLoaded', () => {
        const socket = io.connect('http://' + document.domain + ':' + location.port, {
          transports: ['websocket']
        });

        console.log('sckIO start');
        socket.on('update', function(data) {
            console.log('sckIO update');
            document.getElementById('output3').innerText = data.total;
        });

        function triggerBackend() {
            fetch('/trigger_backend')
                .then(response => response.text())
                .then(data => console.log(data))
                .catch(error => console.error('Error:', error));
        }

        triggerBackend();
    });
</script>

Koyeb Procfile配置 #

gunicorn --worker-class eventlet -w 1 app:app

解决步骤 #

1. 修正Gunicorn启动入口 #

Gunicorn需要绑定SocketIO的WSGI兼容实例，而非原始Flask app。修改Procfile为：

gunicorn --worker-class eventlet -w 1 app:socketio.wsgi_app

2. 修复后端广播逻辑 #

直接调用SocketIO事件处理函数无法在WSGI环境下正确广播，需在路由中直接使用socketio.emit并开启广播模式：

@app.route("/endpoint", methods=['GET', 'POST'])
def endpoint():
    if request.method == 'POST':
        try:
            dbVisits = ApiDb.query.filter_by(id=1).first()
            dbVisits.nrVisits += 1
        except:
            db.session.add(ApiDb(id=1, nrVisits=1))
        db.session.commit()
        
        # 直接广播更新
        dbVisits = ApiDb.query.filter_by(id=1).first()
        socketio.emit('update', {'total': dbVisits.nrVisits}, broadcast=True)
        return {'odgovor': 'OK'}

@app.route('/trigger_backend')
def trigger_backend():
    dbVisits = ApiDb.query.filter_by(id=1).first()
    socketio.emit('update', {'total': dbVisits.nrVisits}, broadcast=True)
    return 'Backend function triggered successfully'

同时移除@socketio.on('trigger_update')装饰的handle_trigger_update函数。

3. 适配前端SocketIO连接 #

Koyeb使用HTTPS，需自动适配协议与域名，无需手动指定端口：

const socket = io({ 
    transports: ['websocket'],
    secure: true,
    reconnection: true
});

4. 锁定依赖版本兼容性 #

在requirements.txt中明确指定兼容版本：

flask-socketio==5.3.5
eventlet==0.33.3
gunicorn==21.2.0
flask==2.3.3

5. Koyeb平台配置检查 #

• 确认服务端口设置为8000（Koyeb默认监听端口）
• 验证平台防火墙允许WebSocket连接（默认开启）

验证流程 #

1. 重新部署修改后的代码到Koyeb
2. 打开前端页面，检查控制台是否输出sckIO start
3. 发送POST请求到/endpoint，确认output3实时更新
4. 访问/trigger_backend，验证前端同样能接收更新

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