基于EC2+Nginx+Gunicorn的Django Redis过期事件监听器最优实现问询

你的问题核心是gunicorn多worker模式下，每个worker都会执行Django的ready()方法，导致多线程监听器重复订阅Redis过期事件，既浪费资源又可能重复执行数据库写入操作。下面是两种高效的解决思路，优先推荐第一种：

一、将Redis监听器独立于Django服务运行（最推荐） #

直接把监听器做成独立的后台进程，和gunicorn托管的Django服务完全解耦，从根源上避免多实例重复监听的问题。

具体实现步骤： #

1. 编写独立的监听器脚本（比如redis_expiry_listener.py）：

   import os
   import redis
   from django.conf import settings
   
   # 初始化Django环境，让脚本能调用Django的ORM和配置
   os.environ.setdefault('DJANGO_SETTINGS_MODULE', '你的项目名.settings')
   import django
   django.setup()
   
   from 你的app.models import 要操作的模型
   
   def listen_redis_expiry():
       # 连接Redis，根据你的配置调整参数
       r = redis.Redis(host='localhost', port=6379, db=0, decode_responses=True)
       # 订阅过期事件（需确保Redis开启了notify-keyspace-events Ex）
       pubsub = r.pubsub()
       pubsub.psubscribe('__keyevent@0__:expired')  # @0表示监听db0，根据你的实际db调整
   
       print("Redis过期事件监听器已启动")
       for message in pubsub.listen():
           if message['type'] == 'pmessage':
               expired_key = message['data']
               # 过滤你需要的特定模式的键，比如以"order_"开头的键
               if expired_key.startswith('order_'):
                   # 执行数据库写入操作，比如更新订单状态
                   order_id = expired_key.split('_')[1]
                   try:
                       order = 要操作的模型.objects.get(id=order_id)
                       order.status = 'expired'
                       order.save()
                       print(f"处理过期订单：{order_id}")
                   except 要操作的模型.DoesNotExist:
                       print(f"订单{order_id}不存在，跳过")
   
   if __name__ == '__main__':
       listen_redis_expiry()
   

2. 配置进程托管工具，保证监听器持续运行：
   用systemd或者supervisor来托管这个脚本，进程挂了会自动重启。比如用systemd：

   • 创建服务文件/etc/systemd/system/redis-expiry-listener.service：

     [Unit]
     Description=Redis Expiry Event Listener for Django App
     After=network.target redis.service
     
     [Service]
     User=ec2-user  # 换成你的EC2实例用户名
     WorkingDirectory=/path/to/your/django/project  # Django项目根目录
     ExecStart=/path/to/your/python/env/bin/python redis_expiry_listener.py  # 虚拟环境的Python路径
     Restart=always
     
     [Install]
     WantedBy=multi-user.target
     

   • 启动并设置开机自启：

     sudo systemctl daemon-reload
     sudo systemctl start redis-expiry-listener.service
     sudo systemctl enable redis-expiry-listener.service
     

3. Redis配置检查：
   打开Redis配置文件（通常是/etc/redis/redis.conf），确保开启了键空间事件通知：

   notify-keyspace-events Ex
   

   修改后重启Redis服务：sudo systemctl restart redis.service

二、在gunicorn环境下只启动一个监听器（适合不想完全独立的场景） #

如果一定要和Django服务绑定，可以利用gunicorn的钩子或者文件锁，保证只有一个worker启动监听器，但这种方式可靠性不如独立部署。

方式1：用gunicorn的post_fork钩子 #

在Django项目的配置文件（比如gunicorn.conf.py）里添加钩子，只让第一个worker启动监听器：

import os
import threading
from django.conf import settings
import redis

def post_fork(server, worker):
    # 只让worker_id为0的进程启动监听器
    if worker.id == 0:
        def listen_redis():
            r = redis.Redis(host='localhost', port=6379, db=0, decode_responses=True)
            pubsub = r.pubsub()
            pubsub.psubscribe('__keyevent@0__:expired')
            for message in pubsub.listen():
                if message['type'] == 'pmessage':
                    expired_key = message['data']
                    if expired_key.startswith('order_'):
                        # 执行数据库操作，逻辑同独立脚本
                        pass
        # 启动线程运行监听器
        listener_thread = threading.Thread(target=listen_redis, daemon=True)
        listener_thread.start()

启动gunicorn时指定这个配置文件：gunicorn --config gunicorn.conf.py your_project.wsgi

方式2：用文件锁保证单实例 #

在Django的ready()方法里添加文件锁，只有获取到锁的进程才启动监听器：

import fcntl
import os
import threading
from django.apps import AppConfig
import redis

class YourAppConfig(AppConfig):
    default_auto_field = 'django.db.models.BigAutoField'
    name = 'your_app'

    def ready(self):
        # 创建锁文件路径
        lock_file_path = '/tmp/redis_listener.lock'
        lock_file = open(lock_file_path, 'w')
        try:
            # 尝试获取排他锁
            fcntl.flock(lock_file, fcntl.LOCK_EX | fcntl.LOCK_NB)
            # 获取锁成功，启动监听器线程
            def listen_redis():
                r = redis.Redis(host='localhost', port=6379, db=0, decode_responses=True)
                pubsub = r.pubsub()
                pubsub.psubscribe('__keyevent@0__:expired')
                for message in pubsub.listen():
                    # 处理逻辑
                    pass
            listener_thread = threading.Thread(target=listen_redis, daemon=True)
            listener_thread.start()
        except BlockingIOError:
            # 其他进程已获取锁，跳过启动
            pass
        finally:
            # 不要关闭文件，否则锁会释放
            pass

这种方式要注意，如果worker进程异常退出，锁文件可能需要手动清理，否则新启动的worker可能无法获取锁。

总结 #

优先选择独立部署监听器的方案，原因如下：

• 完全解耦，不占用gunicorn worker的资源，避免影响Django服务的请求处理；
• 不会受gunicorn worker重启、扩缩容的影响，保证监听器始终单实例运行；
• 排查问题更方便，监听器的日志可以单独管理。

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