python - 在任务中运行组时， celery 运行线程卡住

Question

我与 celery 团体有一些问题。当没有。任务更多是一些线程卡住了，再也不会返回。

@celery.task(bind=True, ignore_result=False, soft_time_limit=TASK_TIME_OUT*15)
def schedule(self, data):
    try:
        job = group(process.s(i[0], i[1]) for i in data)
        result = job.apply_async()
        result_data = None

        with allow_join_result():
            result_data = result.get()

        callback.delay(result_data)
    except SoftTimeLimitExceeded as ex:
        LOGGER.error("Timeout schedule Failed. " + str(ex))

然后，我有另一种方法，该方法需要约1分钟来处理一些数据，也有一些I / O调用，然后返回结果。

@celery.task(bind=True, ignore_result=False, soft_time_limit=TASK_TIME_OUT)
def process(self, param1, param2):
    .......
    .......
    return result

使用Eventlet的Celery Worker命令

celery -A proj worker --loglevel=debug -P eventlet --concurrency=100 -n worker@%h

我在kubernetes上运行了两个worker，每个pod的 CPU = 1和MEM = 512MB 。

我将解释流程:

有一个api调用会触发一个 celery 任务“时间表”。在“计划”方法中，我创建了一个组(我也尝试了和弦，但没有起作用)，并根据输入内容创建了任务，如果任务少于400个，那么它可以正常工作。

如果任务超过500，那么我有一些线程被卡住，再也不会返回，因此，后续步骤永远不会在“计划”方法中运行。

我在做什么错？

为什么celery运行线程卡住了，它说它是 Activity 的，但根本没有运行，因此我的CPU使用率一直很高。以下是检查统计结果。

worker@service-worker-lbv46: OK
{
    "broker": {
        "alternates": [],
        "connect_timeout": 4,
        "failover_strategy": "round-robin",
        "heartbeat": 120.0,
        "hostname": "redis-service-stg",
        "insist": false,
        "login_method": null,
        "port": 6379,
        "ssl": false,
        "transport": "redis",
        "transport_options": {},
        "uri_prefix": null,
        "userid": null,
        "virtual_host": "/"
    },
    "clock": "1243",
    "pid": 7,
    "pool": {
        "free-threads": 98,
        "max-concurrency": 100,
        "running-threads": 2
    },
    "prefetch_count": 400,
    "rusage": {
        "idrss": 0,
        "inblock": 0,
        "isrss": 0,
        "ixrss": 0,
        "majflt": 0,
        "maxrss": 376584,
        "minflt": 4313949,
        "msgrcv": 0,
        "msgsnd": 0,
        "nivcsw": 12441,
        "nsignals": 0,
        "nswap": 0,
        "nvcsw": 1214,
        "oublock": 1840,
        "stime": 18.567744,
        "utime": 585.98883
    },
    "total": {
        "worker.tasks.process": 402,
        "worker.tasks.schedule": 1
    }
}
-> worker@service-worker-g9kh7: OK
{
    "broker": {
        "alternates": [],
        "connect_timeout": 4,
        "failover_strategy": "round-robin",
        "heartbeat": 120.0,
        "hostname": "redis-service-stg",
        "insist": false,
        "login_method": null,
        "port": 6379,
        "ssl": false,
        "transport": "redis",
        "transport_options": {},
        "uri_prefix": null,
        "userid": null,
        "virtual_host": "/"
    },
    "clock": "1243",
    "pid": 7,
    "pool": {
        "free-threads": 99,
        "max-concurrency": 100,
        "running-threads": 1
    },
    "prefetch_count": 400,
    "rusage": {
        "idrss": 0,
        "inblock": 0,
        "isrss": 0,
        "ixrss": 0,
        "majflt": 0,
        "maxrss": 348324,
        "minflt": 3903269,
        "msgrcv": 0,
        "msgsnd": 0,
        "nivcsw": 14085,
        "nsignals": 0,
        "nswap": 0,
        "nvcsw": 28288,
        "oublock": 1840,
        "stime": 11.887338,
        "utime": 291.123721
    },
    "total": {
        "worker.tasks.process": 382
    }
}

在这里您可以看到worker1中有2个正在运行的线程，而worker2中有1个正在运行的线程，它们从很长的时间就开始卡住了。

我知道一个任务不应该等待另一个任务，如果有更好的方法可以建议我。如果架构发生变化。

=================================================== =========

也尝试过使用和弦，但线程仍然停留在执行“进程”任务中，celery inspect stats显示1个正在运行的线程，它什么也没做。问题是它不会执行该线程返回的回调。

另一个问题是它也没有超时。

eventlet == 0.23.0
kombu == 4.1.0
台球== 3.5.0.2
celery == 4.0.2
redis == 2.10.5

提前致谢。

Answer 1

您正在启动任务，并从另一个任务中获取其结果。 This is known to cause deadloack-一旦您的工作人员池耗尽，您的第一个任务就是等待其他因缺少可用工作人员而无法执行的任务。

解决方案是改为use callbacks or more complex workflows。在您的情况下，您确实应该使用和弦，这是正确的解决方案，因此请再试一次，如果仍然遇到问题，请发布有关该问题的信息。