如何实现Celery消费者与Redis Broker断连时的邮件通知？

步骤1：自定义Consumer类重写连接错误处理方法 #

Celery的celery.worker.consumer.consumer类中，on_connection_error_after_connected方法会在Worker与Broker已建立连接后断开时触发。我们可以继承该类，重写方法添加邮件告警逻辑：

# 在Django项目的自定义模块（如celery_extensions.py）中编写
from celery.worker.consumer import Consumer
from django.core.mail import send_mail
from django.conf import settings

class CustomConsumer(Consumer):
    def on_connection_error_after_connected(self, exc, interval):
        # 先执行父类原有逻辑，保证重连机制正常运行
        super().on_connection_error_after_connected(exc, interval)
        
        # 构造告警邮件内容
        subject = "Celery Worker与Redis Broker断开连接警告"
        message = f"""
Celery Worker检测到与Redis Broker的连接已断开，正在尝试重连。
错误详情: {str(exc)}
下次重连间隔: {interval} 秒
        """.strip()
        from_email = settings.DEFAULT_FROM_EMAIL
        recipient_list = ["alert@your-domain.com"]  # 替换为实际接收告警的邮箱
        
        # 发送告警邮件，捕获异常避免影响Celery重连逻辑
        try:
            send_mail(subject, message, from_email, recipient_list, fail_silently=False)
        except Exception as mail_exc:
            self.logger.error(f"断连告警邮件发送失败: {str(mail_exc)}")

步骤2：配置Celery使用自定义Consumer #

在项目的celery.py文件中，指定Worker使用我们自定义的Consumer类：

from celery import Celery
from django.conf import settings
from .celery_extensions import CustomConsumer

app = Celery('your_project_name')

# 加载Django配置
app.config_from_object('django.conf:settings', namespace='CELERY')

# 绑定自定义Consumer
app.conf.worker_consumer = CustomConsumer

# 自动发现项目中的Celery任务
app.autodiscover_tasks(lambda: settings.INSTALLED_APPS)

步骤3：配置Django邮件发送参数 #

确保在settings.py中正确配置邮件后端（以SMTP为例）：

# settings.py
EMAIL_BACKEND = 'django.core.mail.backends.smtp.EmailBackend'
EMAIL_HOST = 'smtp.your-domain.com'  # 替换为你的SMTP服务器地址
EMAIL_PORT = 587
EMAIL_USE_TLS = True
EMAIL_HOST_USER = 'sender@your-domain.com'
EMAIL_HOST_PASSWORD = 'your-email-password-or-app-token'
DEFAULT_FROM_EMAIL = 'sender@your-domain.com'

额外注意事项 #

• 可以添加频率控制逻辑（比如用内存变量或Redis缓存记录上次发送邮件的时间），避免短时间内多次断连导致邮件轰炸。
• 确保Celery Worker进程能访问Django的邮件配置，且网络环境允许Worker连接到SMTP服务器。
• 测试时可手动停止Redis服务，观察Worker日志和邮箱，验证告警逻辑是否正常触发。

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