如何终止Celery中已在执行的任务？（基于Django视图、gevent及Amazon SQS场景）

咱们先搞清楚为啥你用revoke没效果——因为你选了gevent作为worker池。Celery的revoke带terminate=True的终止逻辑，对gevent这类协程池的支持特别有限：gevent是单进程跑多协程，所有任务都挤在同一个进程里，SIGKILL信号没法精准杀掉单个协程任务，worker进程根本不会中断正在跑的协程，所以任务该跑多久还是跑多久，跑完还给你返回结果。

给你两个可行的解决路子：

路子一：切换到prefork池（推荐，如果你能换操作系统） #

正如你后来发现的，prefork是Celery默认的worker池，每个任务都是独立的子进程，SIGKILL能直接干掉对应的子进程，终止任务的逻辑就能正常工作了。步骤很简单：

1. 切换到类Unix系统（比如你提到的CentOS7，Windows确实不支持prefork的完整功能）
2. 启动worker的时候去掉-P gevent，用默认的prefork模式：

celery -A myproj -l info

3. 之后你在Django视图里的撤销代码就能正常生效了：

import myproj.tasks as tasks

task = tasks.mytask.delay(...)
tasks.app.control.revoke(task.task_id, terminate=True, signal="SIGKILL")

路子二：给任务加主动检查机制（适合没法换池的情况） #

如果实在没法切换到prefork，那只能让任务自己“感知”要不要终止——这其实是协程场景下常用的方案，并没有你想的那么繁琐：
核心思路就是在任务的关键节点（比如循环里、耗时操作前后），主动检查自己是否被标记为需要终止，如果是就立刻退出。

举个具体的例子改造你的任务：

# 任务文件 tasks.py
from celery import shared_task, current_task
import time
from your_module import myclass

@shared_task
def mytask(**kwargs):
    # 假设你的任务有一系列耗时操作，比如循环处理
    for step in range(100):
        # 关键：检查当前任务是否被标记为撤销
        if current_task.request.is_aborted():
            # 这里可以加一些清理工作，比如关闭资源
            return "任务已主动终止"
        
        # 模拟你的业务逻辑：调用自定义类
        result = myclass(**kwargs)
        # 模拟耗时操作
        time.sleep(2.5)
    
    return result

然后在Django视图里，调用revoke的时候不需要加terminate=True，只需要标记任务为待终止就行，任务自己会检查：

import myproj.tasks as tasks

task = tasks.mytask.delay(...)
# 标记任务为待终止，任务会主动检查并退出
tasks.app.control.revoke(task.task_id, terminate=False)

小提示：current_task.request.is_aborted()需要Celery 4.0及以上版本支持。如果你用的是旧版本，可以换个思路：用Redis这类缓存存终止标记——调用revoke时往Redis里存{task_id: "terminated"}，任务里定期去查这个键，存在就退出。

再补一句为啥原方法不行 #

gevent池是单进程多协程，所有任务共享同一个进程资源，要是给这个进程发SIGKILL，整个worker都会挂掉，Celery为了避免这种情况，不会给gevent池发送终止信号，所以你的revoke操作根本没起到作用。再加上SQS作为broker，它本身不支持任务的“撤销标记”存储，revoke的信息只存在worker内存里，就算标记了，gevent协程也不会主动响应。

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