117 lines
3.3 KiB
Python
117 lines
3.3 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
"""
|
|
测试SocketIO异步模式问题的诊断脚本
|
|
"""
|
|
|
|
import sys
|
|
import os
|
|
|
|
# 添加当前目录到Python路径
|
|
sys.path.append(os.path.dirname(os.path.abspath(__file__)))
|
|
|
|
print("=== SocketIO 异步模式诊断 ===")
|
|
print(f"Python版本: {sys.version}")
|
|
print(f"工作目录: {os.getcwd()}")
|
|
print()
|
|
|
|
# 检查可用的异步模式
|
|
print("检查可用的异步驱动模块:")
|
|
|
|
# 检查 threading
|
|
try:
|
|
import threading
|
|
print("✓ threading 模块可用")
|
|
except ImportError as e:
|
|
print(f"✗ threading 模块不可用: {e}")
|
|
|
|
# 检查 eventlet
|
|
try:
|
|
import eventlet
|
|
print("✓ eventlet 模块可用")
|
|
except ImportError as e:
|
|
print(f"✗ eventlet 模块不可用: {e}")
|
|
|
|
# 检查 gevent
|
|
try:
|
|
import gevent
|
|
print("✓ gevent 模块可用")
|
|
except ImportError as e:
|
|
print(f"✗ gevent 模块不可用: {e}")
|
|
|
|
print()
|
|
print("检查 engineio 异步驱动:")
|
|
|
|
# 检查 engineio.async_drivers.threading
|
|
try:
|
|
import engineio.async_drivers.threading
|
|
print("✓ engineio.async_drivers.threading 可用")
|
|
except ImportError as e:
|
|
print(f"✗ engineio.async_drivers.threading 不可用: {e}")
|
|
|
|
# 检查 engineio.async_drivers.eventlet
|
|
try:
|
|
import engineio.async_drivers.eventlet
|
|
print("✓ engineio.async_drivers.eventlet 可用")
|
|
except ImportError as e:
|
|
print(f"✗ engineio.async_drivers.eventlet 不可用: {e}")
|
|
|
|
# 检查 engineio.async_eventlet
|
|
try:
|
|
import engineio.async_eventlet
|
|
print("✓ engineio.async_eventlet 可用")
|
|
except ImportError as e:
|
|
print(f"✗ engineio.async_eventlet 不可用: {e}")
|
|
|
|
# 检查 engineio.async_threading
|
|
try:
|
|
import engineio.async_threading
|
|
print("✓ engineio.async_threading 可用")
|
|
except ImportError as e:
|
|
print(f"✗ engineio.async_threading 不可用: {e}")
|
|
|
|
print()
|
|
print("测试 SocketIO 初始化:")
|
|
|
|
try:
|
|
from flask import Flask
|
|
from flask_socketio import SocketIO
|
|
|
|
app = Flask(__name__)
|
|
app.config['SECRET_KEY'] = 'test-key'
|
|
|
|
# 测试不指定 async_mode
|
|
print("测试 SocketIO() 不指定 async_mode...")
|
|
try:
|
|
socketio = SocketIO(app, cors_allowed_origins='*')
|
|
print(f"✓ SocketIO 初始化成功，自动选择的异步模式: {socketio.async_mode}")
|
|
except Exception as e:
|
|
print(f"✗ SocketIO 初始化失败: {e}")
|
|
|
|
# 测试指定 threading
|
|
print("测试 SocketIO(async_mode='threading')...")
|
|
try:
|
|
app2 = Flask(__name__)
|
|
app2.config['SECRET_KEY'] = 'test-key-2'
|
|
socketio2 = SocketIO(app2, cors_allowed_origins='*', async_mode='threading')
|
|
print(f"✓ SocketIO threading 模式初始化成功")
|
|
except Exception as e:
|
|
print(f"✗ SocketIO threading 模式初始化失败: {e}")
|
|
|
|
# 测试指定 eventlet
|
|
print("测试 SocketIO(async_mode='eventlet')...")
|
|
try:
|
|
app3 = Flask(__name__)
|
|
app3.config['SECRET_KEY'] = 'test-key-3'
|
|
socketio3 = SocketIO(app3, cors_allowed_origins='*', async_mode='eventlet')
|
|
print(f"✓ SocketIO eventlet 模式初始化成功")
|
|
except Exception as e:
|
|
print(f"✗ SocketIO eventlet 模式初始化失败: {e}")
|
|
|
|
except Exception as e:
|
|
print(f"✗ Flask/SocketIO 导入失败: {e}")
|
|
|
|
print()
|
|
print("=== 诊断完成 ===")
|
|
print("按回车键退出...")
|
|
input() |