如何为Django RQ特定队列的任务添加自定义后置执行逻辑

方案1：使用RQ信号按需过滤指定队列（最简便） #

RQ原生提供了任务生命周期信号，你可以只监听job_success信号，在回调逻辑中判断任务所属队列，仅匹配目标队列时执行发邮件逻辑：

• 首先在Django应用的apps.py的ready方法中注册信号回调：

from django.apps import AppConfig
import rq.signals
from django.core.mail import mail_admins

def notify_admin_on_job_success(sender, job, **kwargs):
    # 替换为你需要触发逻辑的队列名称
    target_queue = "your_target_queue_name"
    if job.origin == target_queue:
        # 自定义邮件内容，可按需取job的参数、返回值等信息
        subject = f"队列 {target_queue} 任务执行成功"
        message = f"任务ID: {job.id}\n任务方法: {job.func_name}\n返回结果: {job.return_value()}"
        mail_admins(subject, message, fail_silently=True)

class YourAppConfig(AppConfig):
    default_auto_field = "django.db.models.BigAutoField"
    name = "your_app_name"

    def ready(self):
        rq.signals.job_success.connect(notify_admin_on_job_success)

这个方案不需要修改Job基类，也不需要调整队列初始化逻辑，适配所有已存在的任务

方案2：为指定队列单独配置自定义Job类 #

RQ的队列实例支持单独指定job_class参数，不需要全局修改默认Job类，仅需要在初始化目标队列时传入自定义的Job类即可：

• 第一步定义自定义Job类：

from rq.job import Job
from django.core.mail import mail_admins

class NotifyAdminJob(Job):
    def perform(self):
        # 先执行原任务逻辑
        result = super().perform()
        # 执行完成后的额外逻辑
        subject = f"任务 {self.func_name} 执行完成"
        message = f"任务ID: {self.id}\n返回结果: {result}"
        mail_admins(subject, message, fail_silently=True)
        return result

• 第二步在初始化目标队列时指定job_class：

from rq import Queue
from redis import Redis
from your_module import NotifyAdminJob

# 仅目标队列使用自定义Job类，其他队列仍使用默认Job类
target_queue = Queue(
    name="your_target_queue_name",
    connection=Redis(),
    job_class=NotifyAdminJob
)
# 任务入队逻辑和之前完全一致
target_queue.enqueue(your_task_func, *args, **kwargs)

如果是用Django RQ的@job装饰器指定队列，可以直接在装饰器中传入job_class参数：

from django_rq import job

@job("your_target_queue_name", job_class=NotifyAdminJob)
def your_task_func(*args, **kwargs):
    # 任务原有逻辑
    pass

方案3：使用装饰器封装目标队列的任务 #

如果你的目标队列只有少数固定任务，可以直接用装饰器封装任务逻辑，不需要修改RQ的基础配置：

from functools import wraps
from django.core.mail import mail_admins

def notify_admin_after_execution(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        result = func(*args, **kwargs)
        # 执行完成后逻辑
        subject = f"任务 {func.__name__} 执行完成"
        message = f"返回结果: {result}"
        mail_admins(subject, message, fail_silently=True)
        return result
    return wrapper

# 仅给需要触发通知的任务加装饰器即可
@notify_admin_after_execution
def your_target_queue_task(*args, **kwargs):
    # 原有任务逻辑
    pass

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