Flask部署Salesforce CDC前端WebSocket连接异常排查求助

问题原因 #

当通过http://localhost:5000/change访问前端时，浏览器发起的WebSocket握手请求被HTTP服务器错误处理，导致Connection请求头被设置为keep-alive而非WebSocket要求的Upgrade，从而触发连接失败。而file://路径直接打开时无HTTP服务器介入，请求头符合WebSocket规范，所以连接正常。

修复步骤 #

1. 检查前端WebSocket连接地址 #

确保前端代码中WebSocket的初始化URL使用ws://协议，且指向后端WebSocket服务的正确端口和路径。例如：

// 错误示例：指向普通HTTP端口且未指定WebSocket路径
// const ws = new WebSocket('http://localhost:5000');

// 正确示例：使用ws协议，指定WebSocket路由
const ws = new WebSocket('ws://localhost:5000/ws');

2. 使用支持WebSocket的后端框架 #

普通Flask等HTTP服务器无法直接处理WebSocket升级请求，需改用支持WebSocket的框架或扩展：

方案一：Flask + Flask-SocketIO（推荐）

这是最简便的Python WebSocket集成方式，能在同一端口同时提供HTTP页面和WebSocket服务：

1. 安装依赖

pip install flask flask-socketio eventlet

2. 修改app.py代码

from flask import Flask, render_template
from flask_socketio import SocketIO, emit
# 导入你的Salesforce CDC对接模块
import your_salesforce_cdc_module

app = Flask(__name__)
app.config['SECRET_KEY'] = 'your-secret-key'
# 允许跨域（本地调试用）
socketio = SocketIO(app, cors_allowed_origins="*")

# 处理Salesforce CDC消息并推送给前端
def forward_salesforce_changes(message):
    socketio.emit('account_update', message)

# 启动Salesforce CDC监听
def start_cdc_listener():
    # 替换为你的实际CDC监听逻辑，收到消息时调用forward_salesforce_changes
    your_salesforce_cdc_module.listen(forward_salesforce_changes)

@app.route('/change')
def change_page():
    return render_template('change.html')

if __name__ == '__main__':
    # 后台启动CDC监听线程
    import threading
    threading.Thread(target=start_cdc_listener, daemon=True).start()
    # 启动带WebSocket支持的服务器
    socketio.run(app, host='0.0.0.0', port=5000, debug=True)

3. 修改change.html的WebSocket客户端代码

<!-- 引入SocketIO客户端库 -->
<script src="https://cdnjs.cloudflare.com/ajax/libs/socket.io/4.0.1/socket.io.js"></script>
<script>
    const socket = io('http://localhost:5000');
    // 监听后端推送的Account变更消息
    socket.on('account_update', function(data) {
        // 将消息展示到页面（示例：追加到列表）
        const list = document.getElementById('change-log');
        const item = document.createElement('li');
        item.textContent = JSON.stringify(data);
        list.appendChild(item);
    });
</script>

方案二：原生WebSocket + aiohttp

若需使用原生WebSocket，可采用aiohttp框架同时提供HTTP静态服务和WebSocket处理：

1. 安装依赖

pip install aiohttp

2. 修改app.py代码

from aiohttp import web
import your_salesforce_cdc_module

# 存储所有活跃的WebSocket连接
active_ws_connections = set()

async def serve_change_page(request):
    return web.FileResponse('change.html')

async def handle_websocket(request):
    ws = web.WebSocketResponse()
    await ws.prepare(request)
    active_ws_connections.add(ws)
    try:
        async for _ in ws:
            pass
    finally:
        active_ws_connections.remove(ws)
    return ws

# 推送Salesforce变更消息到所有客户端
async def broadcast_cdc_message(message):
    for ws in active_ws_connections:
        if not ws.closed:
            await ws.send_json(message)

# 启动Salesforce CDC监听
async def start_cdc_listener(app):
    your_salesforce_cdc_module.listen(lambda msg: broadcast_cdc_message(msg))

app = web.Application()
app.add_routes([
    web.get('/change', serve_change_page),
    web.get('/ws', handle_websocket)
])
app.on_startup.append(start_cdc_listener)

if __name__ == '__main__':
    web.run_app(app, port=5000)

3. 前端代码保持原生WebSocket初始化：

const ws = new WebSocket('ws://localhost:5000/ws');
ws.onmessage = function(event) {
    const data = JSON.parse(event.data);
    // 处理并展示消息
};

3. 排除中间件干扰 #

若你使用了Flask等框架的第三方中间件，检查是否有修改HTTP请求头的逻辑，确保WebSocket握手请求的Connection和Upgrade头未被篡改。

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