celery 每个 worker 在执行任务时，如何配置一定数量的 task？

我这边本机起了两个 worker ，然后设置了 CELERYD_PREFETCH_MULTIPLIER 为 1 。
但是跑起来以后，发现 STARTED 和 RECEVED 都有一大把，结果把我机器给跑崩了。

本来觉得是在 celery 起的 task 函数，里面的线程可能没优化好，导致这种情况，但后来觉得应该不会一次性 start 这么多的 task 啊。

个人觉得应该是一个 worker 同时只会解决一个 task 才对，或者这个解决的量是可以配置的，我这里没找到配置选项。。。

大家有没有什么好的办法，跪求解决方案！

task

Worker

Celery

receved

23 条回复 • 2019-01-14 11:46:57 +08:00

saggit

2017-02-27 13:19:43 +08:00

新版本的 settings 名子改了， worker_prefetch_multiplier = 1 试试

julyclyde

2017-02-27 13:54:53 +08:00

你设置的这是 fetch 的数量啊，不是 execute 的数量

akmonde

2017-02-27 14:02:39 +08:00

@saggit 先谢谢这位兄弟，不过我设置了 worker_prefetch_multiplier = 1 并没有起作用。

另外，我看到网上有解释，“ celery 中的一个 worker 其实是代表一个进程池，一个进程池是由一个父进程和多个子进程组成，貌似父进程不干事，只用于分配 task ，子进程数默认是 CPU 核数”

故而我也试着配置了 CELERYD_MAX_TASKS_PER_CHILD = 2 ，但是貌似没起作用。

还是每个 worker 分配 start 了好几个 task ， RECEVED 也一大把。

另外，我想问问大家怎么一次性，把那么多起来的 task 全部结束掉，每次 terminate 好多次。

akmonde

2017-02-27 14:03:08 +08:00

@julyclyde 兄弟求配置参数。。我没找到。。

julyclyde

2017-02-27 14:04:52 +08:00

@akmonde http://docs.celeryproject.org/en/latest/userguide/workers.html#concurrency

akmonde

2017-02-27 14:24:28 +08:00

@julyclyde 非常感谢兄弟，确实有效，要是能直接写到配置里就更好了， hhhh
另外兄弟知道把那么 start 和 receive 后 task 全部结束掉么?
我每次都是直接 ctrl+C 把 worker 结束掉，因为用的 redis 作中间件，最后还会手动清理 redis 的缓存内容，不然它还会一直运行下去。
不过这样好像不太正规，也有点 bug 。

julyclyde

2017-02-27 17:42:14 +08:00

犯不着写配置文件里啊，反正你也得从 celery worker 命令行启动，多个-c 参数的小事而已

听不懂“把那么 start 和 receive 后 task 全部结束掉”

akmonde

2017-02-27 18:57:25 +08:00

@julyclyde 我的意思是：前台已经发了任务，存在了中间件（ redis ），然后后台取了任务，有的已经 start ，有的还在 receive 状态，如何把这些废弃的任务全部结束掉。
我直接把那些 worker ctrl+c 掉的话，重新启动 worker 时，那些被废弃的任务还会继续跑。

julyclyde

2017-02-27 21:20:54 +08:00

这细节动作我就没注意观察过了
诶按说 redis 作为队列的话，取出来是不会放回去的啊？

maemo

2017-02-27 21:45:39 +08:00

@akmonde 你有用 multi 方式起 worker 吗，如果有的话，直接把起 worker 的命令中的 start 换成 stop 。要清理废弃的任务的话，可以用 celery purge -Q queuename, 具体用法可以查看 celery purge --help

akmonde

2017-02-27 23:14:51 +08:00

@maemo 并没有用 multi 喔，至于您说的 queuename 是 task 名么？类似于“ e1e10f99-fb7b-42a4-b627-7ea0e74daf90 ”？我那边用 flower 控制的，但也没有看见批量清除的法子，有点头疼。。

akmonde

2017-02-27 23:30:00 +08:00

@julyclyde 没，我猜是放到了 redis 里面的，清除 redis 后过会儿会结束。
但清除 redis 后再启动时， celery 还会调度一会儿 task ，这部分不知道是存在哪儿的，我一直很奇怪。

xnightsky

2017-02-27 23:50:41 +08:00

看配置项`result_backend `, task 信息存在 result_backend 配置对应位置，还有`result_expires `配置项，所有 task 默认存 1 天

xnightsky

2017-02-27 23:51:30 +08:00

不过 celelry 命令行有参数能干掉所有任务

akmonde

2017-02-28 00:37:39 +08:00

@xnightsky 谢谢兄弟，刚找了下干掉所有任务的参数，不过没找到。。有点伤

xnightsky

2017-02-28 01:21:29 +08:00

`celery -h` 其他配置项就不说了
看一下命令行说明
````
celery purge

返回
WARNING: This will remove all tasks from queue: celery.
There is no undo for this operation!

(to skip this prompt use the -f option)

Are you sure you want to delete all tasks (yes/NO)?
````
这是我本机的提示

xnightsky

2017-02-28 01:23:01 +08:00

不知他能否干掉已经被 fetch 的命令，需要你排查

akmonde

2017-02-28 01:24:02 +08:00

@xnightsky 3ks ，我明天看看~~

Stony

2017-08-03 00:36:30 +08:00

http://docs.celeryproject.org/en/latest/userguide/configuration.html
参数配置都在这里，新旧配置的 key 对照表也有。

CELERYD_PREFETCH_MULTIPLIER = n
worker_prefetch_multiplier = n
这个是任务预取功能，就是每个工作的进程／线程／绿程在获取任务的时候，会尽量多拿 n 个，以保证获取的通讯成本可以压缩，在每个任务很短（明显小于 1 秒）情况下，是值得调大的，而且推荐是 2 的幂。0 表示尽可能多拿。如果 1 个都不想多拿，那么除了设置 1 外，还需要设置 task_acks_late 为 true，如果你的任务不是幂等（可以重复调用）的话，可能会有问题。详细解释参考： http://docs.celeryproject.org/en/latest/userguide/optimizing.html
task_acks_late = True
worker_prefetch_multiplier = 1

CELERYD_MAX_TASKS_PER_CHILD = n
worker_max_tasks_per_child = n
这个表示每个工作的进程／线程／绿程在执行 n 次任务后，主动销毁，之后会起一个新的。主要解决一些资源释放的问题。

CELERY_TASK_RESULT_EXPIRES = s
result_expires = s
这个表示保存任务结果的时长，这个时间会被设置到 redis 里面（假设 backend 是 redis ），如果抓取数据量大的话，是可以缩短保存的时间，节省 backend 的资源（ redis 主要是内存）消耗，默认是 24 小时（ 86400 ），单位是秒。

===分割线===
已经 fetch 的任务，要么等执行完毕，要么等 kill 掉。要注意 soft kill 和 hard kill 的区别，没记错的话是 TERMINAL 和 kill -9 的区别。soft kill 会执行完当前的 task，但是，prefetch 的会丢失。在 pycharm 里，我第一次 ctrl+c 是软 kill，可能会等上一会儿，第二次 ctrl+c 是硬 kill，直接停掉进程。

没有 fetch 的任务可以 purge 清理调队列中的。

根据选择的 pool 类型不同，可以有 solo （单进程） prefork （进程池） threading （线程池） gevent （协程池）和 eventlet （协程池）。4.x 好像废弃了线程池，主要的原因我猜可能和 GIL 有关。进程池比较耗内存，好处是所有工作线程资源都是隔离的，如果配置动态数量的挺好用。协程池则对于 IO 密集型工作比较有优势。所有的并发数量默认是 cpu core 的数量，4 核就是 4，可以根据实际情况调大。

akmonde

2017-08-03 12:56:23 +08:00

@Stony 非常感谢您，很详细，这帖子我差点都忘了。。另外，我想请教下您，如果想批量 kill 一批机器的 task，应该咋弄，我通常是直接 grep celery 的所有进程，然后 kill -9。
但是这样就有个毛病，我批量 kill 集群中的 task 时会出现问题，只有一台一台机器去手动操作。
另外，您说如果 celery 卡住的时候，重启所有 pool 有用么？话说我以前配置了 pool 的安全选项，还是不能在 flower 直接重启。

mengdodo

2018-11-15 11:23:49 +08:00

@akmonde
#查看 celery 启动的 work 进程 pid
ps aux | grep celery
#批量 kill 所有的 celery 生成的 work 方法 1
ps auxww | grep 'worker -l info' | awk '{print $2}' | xargs kill -9
#批量 kill 掉 celery 的 work 方法 2
详细文档： https://celery.readthedocs.io/en/latest/userguide/workers.html#stopping-the-worker
pkill -9 -f 'celery -A [项目名] worker'

akmonde

2018-11-19 16:50:11 +08:00

@mengdodo 这是 17 年的帖子了...不过还是谢谢兄 dei~

louyuguang

2019-01-14 11:46:57 +08:00

@Stony 兄弟，你的 task_acks_late 这个参数对我帮助太大了，这个问题我找了两个礼拜。
因为我的一个 task 可能要执行 10s~600s 不等，我的 worker 同时只能执行一个 task。所以我不希望任何一个 worker 在执行一个任务的时候，再接收其他任务（除非所有 worker 都在执行任务。）