Gunicorn+Eventlet环境下SocketIO后台任务Redis Pub/Sub订阅无故停止问题求助

大家好，我最近碰到一个头疼的问题，想请教下有没有同行遇到过类似的情况。

我当前的服务架构是：用Gunicorn启动1个Eventlet worker，通过SocketIO和前端做WebSocket通信；同时需要监听一个外部Pod发出的Redis Pub/Sub事件，所以我用SocketIO的后台线程来启动Redis订阅监听逻辑。

以下是我的关键代码：

1. SocketIO初始化（extensions.py） #

socketio = SocketIO(
    logger=False, 
    engineio_logger=False, 
    path='/socket.io', 
    cors_allowed_origins='*', 
    async_mode='eventlet', 
    ping_timeout=10, 
    ping_interval=60
)

2. Redis订阅监听类 #

class RedisConnection:
    def __init__(self):
        # 初始化逻辑省略
        pass

    def subscription_listener(self):
        while True:
            try:
                pubsub = self.redis.pubsub()
                pubsub.subscribe(self.websocket_channel)
                try:
                    for message in pubsub.listen():
                        try:
                            m = TypeAdapter(RedisEnvelopeMessage).validate_python(message)
                            self.logger.debug(f"Received from REDIS: {message}")
                            if m.type == 'message':
                                d = m.data
                                self.logger.debug(f"Received message from REDIS: {d}")
                                with self.app.app_context():
                                    # 从Flask扩展中获取对应的WS服务实例
                                    WSS = self.app.extensions.get(d.module, "RedisWS").ws_services.get(d.company_id)
                                    # 根据消息类型调用对应方法
                                    if d.message is not None:
                                        if isinstance(d.message, list):
                                            getattr(WSS, d.method)(*d.message)
                                        elif isinstance(d.message, dict):
                                            getattr(WSS, d.method)(d.message)
                                        elif isinstance(d.message, str):
                                            getattr(WSS, d.method)(d.message)
                                        else:
                                            getattr(WSS, d.method)()
                            elif m.type == 'subscribe':
                                self.logger.info(f"Subscribed to REDIS channel: {m.channel}")
                            else:
                                self.logger.info(f"Received message from REDIS but NOT PROCESSED: {message}")
                        except Exception as e:
                            self.logger.catch_exception(e)
                            self.logger.error(f"Pubsub parsing error: {e}").save("pubsub_listener")
                except Exception as e:
                    self.logger.catch_exception(e, level="critical")
                    self.logger.error(f"Pubsub listener error: {e}").save("pubsub_listener")
                    # socketio.sleep(2)
            except Exception as e:
                self.logger.catch_exception(e, level="critical")
                self.logger.error(f"Pubsub loop error: {e}").save("pubsub_listener")

3. Flask应用初始化时启动后台任务 #

def create_app(config_class=Config):
    # 其他初始化逻辑省略
    socketio.init_app(app)
    with app.app_context():
        app.extensions["pubsub"] = socketio.start_background_task(redis_db_static.subscription_listener)

问题现象 #

代码刚启动时一切正常，Redis订阅能正常接收并处理事件，但运行一段时间后，这个订阅监听就毫无征兆地停止了，而且我检查了所有日志，没有任何异常信息输出。重启Gunicorn服务后，订阅功能又能恢复正常。

我原本期望这个后台线程能持续运行，直到Gunicorn主进程被杀死或重启时才会终止重启，但现在它会中途挂掉，完全找不到原因。

有没有朋友遇到过类似的问题？麻烦给点排查思路或者解决方案，非常感谢！

内容来源于stack exchange