Flask-SocketIO应用中使用PyMongo时出现gevent AssertionError

• gevent版本：24.2.1（pip install gevent）
• Python版本：Python 3.12.3（python --version）
• 操作系统：

No LSB modules are available.
Distributor ID: Ubuntu
Description:    Ubuntu 24.04.1 LTS
Release:        24.04
Codename:       noble

使用gunicorn运行Flask-SocketIO应用时出现如下错误：

[2024-09-19 19:40:20 +0530] [7874] [INFO] Starting gunicorn 23.0.0
[2024-09-19 19:40:20 +0530] [7874] [INFO] Listening at: http://0.0.0.0:8000 (7874)
[2024-09-19 19:40:20 +0530] [7874] [INFO] Using worker: geventwebsocket.gunicorn.workers.GeventWebSocketWorker
[2024-09-19 19:40:20 +0530] [7893] [INFO] Booting worker with pid: 7893
Traceback (most recent call last):
  File "src/gevent/_abstract_linkable.py", line 287, in gevent._gevent_c_abstract_linkable.AbstractLinkable._notify_links
  File "src/gevent/_abstract_linkable.py", line 333, in gevent._gevent_c_abstract_linkable.AbstractLinkable._notify_links
AssertionError: (None, <callback at 0x769b75efca40 args=([],)>)
2024-09-19T14:10:20Z <callback at 0x769b75efca40 args=([],)> failed with AssertionError

启动命令：gunicorn -c gunicorn.conf.py app:app

Gunicorn配置：

workers = 1
worker_class = 'geventwebsocket.gunicorn.workers.GeventWebSocketWorker'
bind = '0.0.0.0:8000'
worker_connections = 1000
loglevel = 'info'
accesslog = '-'
errorlog = '-'
timeout = 120
preload_app = True

app.py代码：

from gevent import monkey
_ = monkey.patch_all()

import os
import jwt
from functools import wraps
from flask import Flask, render_template, request, redirect, make_response, url_for, send_from_directory
from utils.data_utils import extract_data as extract_data1688, get_driver as get_driver1688, LOAD_DELAY as LOAD_DELAY1688
from flask_socketio import SocketIO, emit
import random
import traceback
from pymongo.mongo_client import MongoClient
from pymongo.server_api import ServerApi
from dotenv import load_dotenv
load_dotenv()

app = Flask(__name__)
app.config['SECRET_KEY'] = os.getenv("JWT_SECRET")
socketio = SocketIO(app)
client = MongoClient("<exactly_correct_uri>", server_api=ServerApi('1'))

JWT_SECRET = os.getenv("JWT_SECRET")

def token_required(fn):
    @wraps(fn)
    def wrapper(*args, **kwargs):
        token = request.cookies.get('auth_token')
        if not token:
            return redirect('/login')
        try:
            decoded_token = jwt.decode(token, JWT_SECRET, algorithms=['HS256'])
            if decoded_token.get('role') != 'MOD':
                return redirect('/login')
        except jwt.ExpiredSignatureError:
            return redirect('/login')
        except jwt.InvalidTokenError:
            return redirect('/login')
        return fn(*args, **kwargs)
    return wrapper

@app.route('/')
@token_required
def index():
    return render_template('index.html')

@app.route('/login', methods=['GET', 'POST'])
def login():
    message = request.args.get('message', "")
    if request.method == 'POST':
        token = request.form.get('auth_token')
        try:
            decoded_token = jwt.decode(token, JWT_SECRET, algorithms=['HS256'])
            if decoded_token.get('role') != 'MOD':
                return redirect(url_for('login', message="Unauthorized role."))
        except jwt.ExpiredSignatureError:
            return redirect(url_for('login', message="Token expired. Please try again."))
        except jwt.InvalidTokenError:
            return redirect(url_for('login', message="Invalid token. Please try again."))

        resp = make_response(redirect('/'))
        resp.set_cookie('auth_token', token)
        return resp

    return render_template('login.html', message=message)

@app.route('/public/<path:filename>')
def serve_public_file(filename):
    return send_from_directory('public', filename)

@socketio.on("data_scrape_start")
@token_required
def data_scrape_start(url):
    print("[scraper] Scraping:", url)
    driver = None
    try:  
        emit("log", {"message": f"Scraping: {url}", "level": "i"})
        socketio.sleep(random.uniform(2, 5))
        driver = get_driver1688(True)
        driver.get(url)
        socketio.sleep(LOAD_DELAY1688)
        data = extract_data1688(driver, url, socketio)
        emit("data", data)
        emit("log", {"message": "Fetched all data!", "level": "s"})
        emit("log", {"message": "Uploading to database...", "level": "i"})
        # collection.insert_one(data)
        emit("log", {"message": "Uploaded all data!", "level": "s"})
    except Exception as e:
        emit("log", {"message": f"Error: {str(e)}", "level": "e"})
        traceback.print_exc()
    finally:
        emit("data_scrape_end")
        if driver:
            driver.quit()


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

补充信息：直接使用python app.py运行应用时无错误，可正常工作；且仅在包含数据库代码时才会出现该错误。

1. 禁用preload_app配置 #

Gunicorn配置中启用的preload_app = True会导致MongoDB客户端在主进程初始化后被fork到子进程，而gevent的协程环境无法正确处理这种跨进程的资源共享，进而触发断言错误。

修改gunicorn.conf.py，将preload_app设置为False：

preload_app = False

2. 延迟MongoDB客户端初始化 #

如果必须使用preload_app，可以将MongoDB客户端的初始化逻辑延迟到Worker进程启动后执行，避免在主进程中初始化。示例如下：

# 移除全局的client初始化，改为在需要时创建
def get_mongo_client():
    return MongoClient("<exactly_correct_uri>", server_api=ServerApi('1'))

# 在需要使用数据库的地方调用
@socketio.on("data_scrape_start")
@token_required
def data_scrape_start(url):
    client = get_mongo_client()
    collection = client.your_database.your_collection
    # ... 后续操作

3. 检查版本兼容性 #

确保使用的pymongo版本支持gevent的协程环境，建议安装pymongo[srv]最新稳定版，避免版本不兼容导致的异步回调问题。若问题仍存在，可尝试将gevent版本降级到23.9.1，该版本对协程回调的处理更稳定。

验证 #

修改配置后，重新执行gunicorn -c gunicorn.conf.py app:app启动应用，检查是否还会出现断言错误。

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