Celery 里面可以使用多进程、多线程、协程吗？

测试积点老人

本人使用 Django + Celery 搭建了一套测试平台，其中有个需求是运行一个任务，这个任务需要处理几万至上百万张图片，为此使用 celery 进行任务调度和执行。现在的问题是一次任务需要花半小时才能完成 10W 张图片的处理，我想缩短任务执行时间，所以想在 celery 中的一个任务中使用多进程 + 协程的方式，但我在网上没有找到 celery 中使用多进程、协程的示例，官方文档也没有涉及这一块。
我尝试在 celery 直接使用多进程，运行后报错，不通过 celery 是没问题的，以下是代码：

1. import time
   
2. 
   
3. from celery import Celery
   
4. from multiprocessing import Pool
   
5. 
   
6. app = Celery('tasks', broker='redis://:docserver123456!@172.17.10.175:6379/3')
   
7. 
   
8. 
   
9. def func(msg):
   
10.     print("*msg: ", msg)
    
11.     time.sleep(3)
    
12.     print("*end")
    
13. 
    
14. 
    
15. @app.task
    
16. def add():
    
17.     p = Pool(5)
    
18.     for i in range(10):
    
19.         msg = f'hello str({i})'
    
20.         p.apply_async(func, (msg,))
    
21.     p.close()
    
22.     p.join()
    
23.     print('all done')

复制代码

以下是运行信息，直接调用成功，使用 celery 失败：

1. [root[url=home.php?mod=space&uid=152075]@localhost[/url] ce_test]# python
   
2. Python 3.7.8 (default, Jan 26 2021, 15:45:27)
   
3. [GCC 4.8.5 20150623 (Red Hat 4.8.5-44)] on linux
   
4. Type "help", "copyright", "credits" or "license" for more information.
   
5. >>> from tasks import add
   
6. >>> add()
   
7. *msg:  hello str(0)
   
8. *msg:  hello str(1)
   
9. *msg:  hello str(2)
   
10. *msg:  hello str(3)
    
11. *msg:  hello str(4)
    
12. *end
    
13. *end
    
14. *end
    
15. *msg:  hello str(5)
    
16. *end
    
17. *msg:  hello str(6)
    
18. *msg:  hello str(7)
    
19. *end
    
20. *msg:  hello str(8)
    
21. *msg:  hello str(9)
    
22. *end
    
23. *end
    
24. *end
    
25. *end
    
26. *end
    
27. all done
    
28. >>> add.delay()
    
29. <AsyncResult: f0bbd061-53b6-44a9-8087-ad08fed4401c>

复制代码

以下是报错信息：

1. [root@localhost ce_test]# celery -A tasks worker -l info
   
2. /usr/local/python3/lib/python3.7/site-packages/celery/platforms.py:801: RuntimeWarning: You're running the worker with superuser privileges: this is
   
3. absolutely not recommended!
   
4. 
   
5. Please specify a different user using the --uid option.
   
6. 
   
7. User information: uid=0 euid=0 gid=0 egid=0
   
8. 
   
9.   uid=uid, euid=euid, gid=gid, egid=egid,
   
10. 
    
11. -------------- celery@localhost.localdomain v4.4.7 (cliffs)
    
12. --- ***** -----
    
13. -- ******* ---- Linux-3.10.0-1160.11.1.el7.x86_64-x86_64-with-centos-7.9.2009-Core 2021-03-18 15:45:41
    
14. - *** --- * ---
    
15. - ** ---------- [config]
    
16. - ** ---------- .> app:         tasks:0x7fae4a6f1590
    
17. - ** ---------- .> transport:   redis://:**@172.17.10.175:6379/3
    
18. - ** ---------- .> results:     disabled://
    
19. - *** --- * --- .> concurrency: 4 (prefork)
    
20. -- ******* ---- .> task events: OFF (enable -E to monitor tasks in this worker)
    
21. --- ***** -----
    
22. -------------- [queues]
    
23.                 .> celery           exchange=celery(direct) key=celery
    
24. 
    
25. 
    
26. [tasks]
    
27.   . tasks.add
    
28. 
    
29. [2021-03-18 15:45:41,779: INFO/MainProcess] Connected to redis://:**@172.17.10.175:6379/3
    
30. [2021-03-18 15:45:41,797: INFO/MainProcess] mingle: searching for neighbors
    
31. [2021-03-18 15:45:42,895: INFO/MainProcess] mingle: all alone
    
32. [2021-03-18 15:45:42,920: INFO/MainProcess] celery@localhost.localdomain ready.
    
33. [2021-03-18 15:49:02,260: INFO/MainProcess] Received task: tasks.add[f0bbd061-53b6-44a9-8087-ad08fed4401c]
    
34. [2021-03-18 15:49:02,271: ERROR/ForkPoolWorker-1] Task tasks.add[f0bbd061-53b6-44a9-8087-ad08fed4401c] raised unexpected: AssertionError('daemonic processes are not allowed to have children')
    
35. Traceback (most recent call last):
    
36.   File "/usr/local/python3/lib/python3.7/site-packages/celery/app/trace.py", line 412, in trace_task
    
37.     R = retval = fun(*args, **kwargs)
    
38.   File "/usr/local/python3/lib/python3.7/site-packages/celery/app/trace.py", line 704, in __protected_call__
    
39.     return self.run(*args, **kwargs)
    
40.   File "/home/ce_test/tasks.py", line 17, in add
    
41.     p = Pool(5)
    
42.   File "/usr/local/python3/lib/python3.7/multiprocessing/context.py", line 119, in Pool
    
43.     context=self.get_context())
    
44.   File "/usr/local/python3/lib/python3.7/multiprocessing/pool.py", line 176, in __init__
    
45.     self._repopulate_pool()
    
46.   File "/usr/local/python3/lib/python3.7/multiprocessing/pool.py", line 241, in _repopulate_pool
    
47.     w.start()
    
48.   File "/usr/local/python3/lib/python3.7/multiprocessing/process.py", line 110, in start
    
49.     'daemonic processes are not allowed to have children'
    
50. AssertionError: daemonic processes are not allowed to have children

复制代码

有哪位大哥能帮忙解答下吗？

qqq911

可以试试集群模式

jingzizx

不清楚