Queue类有一个empty()方法，可按如下方式访问：

import django_rq
q = django_rq.get_failed_queue()
q.empty()

然而，在我的测试中，这只是清除了Redis中的失败列表键，而不是作业键本身。所以你的数千个作业仍然会占用Redis的内存。为了防止这种情况发生，你必须逐个删除这些作业：

import django_rq
q = django_rq.get_failed_queue()
while True:
    job = q.dequeue()
    if not job:
        break
    job.delete()  # Will delete key from Redis

至于在管理界面中有一个按钮，你必须改变django-rq/templates/django-rq/jobs.html模板，它扩展了admin/base_site.html，似乎没有给予任何定制的空间。

redis-cli允许FLUSHDB，伟大的为我的本地环境，因为我产生了一个bizzallion工作.
随着Django集成的工作，我会更新。只是增加了0.02美元。

由于@Augusto-men方法似乎不再起作用了，这里有另一种解决办法：
您可以使用原始连接删除失败的作业。只需遍历rq：job键并检查作业状态。

from django_rq import get_connection
from rq.job import Job

# delete failed jobs
con = get_connection('default')
for key in con.keys('rq:job:*'):
    job_id = key.decode().replace('rq:job:', '')
    job = Job.fetch(job_id, connection=con)
    if job.get_status() == 'failed':
        con.delete(key)
con.delete('rq:failed:default')  # reset failed jobs registry

可以使用以下代码示例按名称清空任何队列：

import django_rq

queue = "default"
q = django_rq.get_queue(queue)
q.empty()

或者甚至有Django命令：

import django_rq

from django.core.management.base import BaseCommand

class Command(BaseCommand):
    def add_arguments(self, parser):
        parser.add_argument("-q", "--queue", type=str)

    def handle(self, *args, **options):
        q = django_rq.get_queue(options.get("queue"))
        q.empty()

由于RQ更新实现了Registries，其他答案已过时。
现在，您需要执行此操作来循环并删除failed作业。这也适用于任何特定的Registry。

import django_rq
from rq.registry import FailedJobRegistry

failed_registry = FailedJobRegistry('default', connection=django_rq.get_connection())

for job_id in failed_registry.get_job_ids():
    try:
        failed_registry.remove(job_id, delete_job=True)
    except:
        # failed jobs expire in the queue. There's a
        # chance this will raise NoSuchJobError
        pass

Source

您可以使用以下命令从命令行清空队列：
rq empty [queue-name]
运行rq info将列出所有队列。