python celery多worker、多队列、定时任务

多worker、多队列
celery是一个分布式的任务调度模块，那么怎么实现它的分布式功能呢，celery可以支持多台不同的计算机执行不同的任务或者相同的任务。

如果要说celery的分布式应用的话，就要提到celery的消息路由机制，提到AMQP协议。

简单理解：

可以有多个”消息队列”（message Queue），不同的消息可以指定发送给不同的Message Queue，

而这是通过Exchange来实现的，发送消息到”消息队列”中时，可以指定routiing_key，Exchange通过routing_key来吧消息路由（routes）到不同的”消息队列”中去。

celery多worker、多队列、定时任务 - 图1

exchange 对应 一个消息队列(queue)，即：通过”消息路由”的机制使exchange对应queue，每个queue对应每个worker。

下面我们来看一个列子：

  1. '''
  3. '''
  4. vi tasks.py
  6. #!/usr/bin/env python
  7. #-*- coding:utf-8 -*-
  8. from celery import Celery
  10. app = Celery()
  11. app.config_from_object("celeryconfig")  # 指定配置文件
  13. @app.task
  14. def taskA(x,y):
  15. return x + y
  17. @app.task
  18. def taskB(x,y,z):
  19. return x + y + z
  21. @app.task
  22. def add(x,y):
  23. return x + y

编写配置文件，配置文件一般单独写在一个文件中。

  1. vi celeryconfig.py
  3. #!/usr/bin/env python
  4. #-*- coding:utf-8 -*-
  6. from kombu import Exchange,Queue
  8. BROKER_URL = "redis://47.106.106.220:5000/1" 
  9. CELERY_RESULT_BACKEND = "redis://47.106.106.220:5000/2"
  11. CELERY_QUEUES = (
  12. Queue("default",Exchange("default"),routing_key="default"),
  13. Queue("for_task_A",Exchange("for_task_A"),routing_key="for_task_A"),
  14. Queue("for_task_B",Exchange("for_task_B"),routing_key="for_task_B") 
  15. )
  16. # 路由
  17. CELERY_ROUTES = {
  18. 'tasks.taskA':{"queue":"for_task_A","routing_key":"for_task_A"},
  19. 'tasks.taskB':{"queue":"for_task_B","routing_key":"for_task_B"}
  20. }

远程客户端上编写测试脚本

  1. vi test.py
  3. from tasks import *
  4. re1 = taskA.delay(100, 200)
  5. print(re1.result)
  6. re2 = taskB.delay(1, 2, 3)
  7. print(re2.result)
  8. re3 = add.delay(1, 2)
  9. print(re3.status)

启动两个worker来分别指定taskA、taskB，开两个窗口分别执行下面语句。

  1. celery -A tasks worker -l info -n workerA.%h -Q for_task_A
  3. celery -A tasks worker -l info -n workerB.%h -Q for_task_B

远程客户端上执行脚本可以看到如下输出：

  1. python test.py 
  2. 300
  3. 6
  4. PENDING

在taskA所在窗口可以看到如下输出：

  1. .......
  2. .......
  3. .......
  4. task_A
  6. [tasks]
  7.   . tasks.add
  8.   . tasks.taskA
  9.   . tasks.taskB
  11. [2018-05-27 19:23:49,235: INFO/MainProcess] Connected to redis://47.106.106.220:5000/1
  12. [2018-05-27 19:23:49,253: INFO/MainProcess] mingle: searching for neighbors
  13. [2018-05-27 19:23:50,293: INFO/MainProcess] mingle: all alone
  14. [2018-05-27 19:23:50,339: INFO/MainProcess] celery@workerA.izwz920j4zsv1q15yhii1qz ready.
  15. [2018-05-27 19:23:56,051: INFO/MainProcess] sync with celery@workerB.izwz920j4zsv1q15yhii1qz
  16. [2018-05-27 19:24:28,855: INFO/MainProcess] Received task: tasks.taskA[8860e78a-b82b-4715-980c-ae125dcab2f9]  
  17. [2018-05-27 19:24:28,872: INFO/ForkPoolWorker-1] Task tasks.taskA[8860e78a-b82b-4715-980c-ae125dcab2f9] succeeded in 0.0162177120219s: 300

在taskB所在窗口可以看到如下输出：

  1. .......
  2. .......
  3. .......
  4. task_B
  5. [tasks]
  6.   . tasks.add
  7.   . tasks.taskA
  8.   . tasks.taskB
  10. [2018-05-27 19:23:56,012: INFO/MainProcess] Connected to redis://47.106.106.220:5000/1
  11. [2018-05-27 19:23:56,022: INFO/MainProcess] mingle: searching for neighbors
  12. [2018-05-27 19:23:57,064: INFO/MainProcess] mingle: sync with 1 nodes
  13. [2018-05-27 19:23:57,064: INFO/MainProcess] mingle: sync complete
  14. [2018-05-27 19:23:57,112: INFO/MainProcess] celery@workerB.izwz920j4zsv1q15yhii1qz ready.
  15. [2018-05-27 19:24:33,885: INFO/MainProcess] Received task: tasks.taskB[5646d0b7-3dd5-4b7f-8994-252c5ef03973]  
  16. [2018-05-27 19:24:33,910: INFO/ForkPoolWorker-1] Task tasks.taskB[5646d0b7-3dd5-4b7f-8994-252c5ef03973] succeeded in 0.0235358460341s: 6

我们看到状态是PENDING，表示没有执行，这个是因为没有celeryconfig.py文件中指定改route到哪一个Queue中，所以会被发动到默认的名字celery的Queue中，但是我们还没有启动worker执行celery中的任务。下面，我们来启动一个worker来执行celery队列中的任务。

celery -A tasks worker -l info -n worker.%h -Q celery
再次在远程客户端执行test.py，可以看到结果执行成功，并且刚新启动的worker窗口有如下输出：

  1. .......
  2. .......
  3. .......
  4. [tasks]
  5.   . tasks.add
  6.   . tasks.taskA
  7.   . tasks.taskB
  9. [2018-05-27 19:25:44,596: INFO/MainProcess] Connected to redis://47.106.106.220:5000/1
  10. [2018-05-27 19:25:44,611: INFO/MainProcess] mingle: searching for neighbors
  11. [2018-05-27 19:25:45,660: INFO/MainProcess] mingle: sync with 2 nodes
  12. [2018-05-27 19:25:45,660: INFO/MainProcess] mingle: sync complete
  13. [2018-05-27 19:25:45,711: INFO/MainProcess] celery@worker.izwz920j4zsv1q15yhii1qz ready.
  14. [2018-05-27 19:25:45,868: INFO/MainProcess] Received task: tasks.add[f9c5ca2b-623e-4c0a-9c45-a99fb0b79ed5]  
  15. [2018-05-27 19:25:45,880: INFO/ForkPoolWorker-1] Task tasks.add[f9c5ca2b-623e-4c0a-9c45-a99fb0b79ed5] succeeded in 0.0107084610499s: 3

Celery与定时任务

在celery中执行定时任务非常简单，只需要设置celery对象中的CELERYBEAT_SCHEDULE属性即可。
下面我们接着在celeryconfig.py中添加CELERYBEAT_SCHEDULE变量：

  1. '''
  2. 遇到问题没人解答？小编创建了一个Python学习交流QQ群：857662006 
  3. 寻找有志同道合的小伙伴，互帮互助,群里还有不错的视频学习教程和PDF电子书！
  4. '''
  5. cat celeryconfig.py
  7. #!/usr/bin/env python
  8. #-*- coding:utf-8 -*-
  10. from kombu import Exchange,Queue
  12. BROKER_URL = "redis://47.106.106.220:5000/1" 
  13. CELERY_RESULT_BACKEND = "redis://47.106.106.220:5000/2"
  15. CELERY_QUEUES = (
  16. Queue("default",Exchange("default"),routing_key="default"),
  17. Queue("for_task_A",Exchange("for_task_A"),routing_key="for_task_A"),
  18. Queue("for_task_B",Exchange("for_task_B"),routing_key="for_task_B") 
  19. )
  21. CELERY_ROUTES = {
  22. 'tasks.taskA':{"queue":"for_task_A","routing_key":"for_task_A"},
  23. 'tasks.taskB':{"queue":"for_task_B","routing_key":"for_task_B"}
  24. }
  26. # 新增加的定时任务部分
  27. CELERY_TIMEZONE = 'UTC'
  28. CELERYBEAT_SCHEDULE = {
  29.     'taskA_schedule' : {
  30.         'task':'tasks.taskA',
  31.         'schedule':2,
  32.         'args':(5,6)
  33.     },
  34.     'taskB_scheduler' : {
  35.         'task':"tasks.taskB",
  36.         "schedule":10,
  37.         "args":(10,20,30)
  38.     },
  39.     'add_schedule': {
  40.         "task":"tasks.add",
  41.         "schedule":5,
  42.         "args":(1,2)
  43.     }
  44. }

还是按之前启动三个worker

  1. celery -A tasks worker -l info -n workerA.%h -Q for_task_A
  3. celery -A tasks worker -l info -n workerB.%h -Q for_task_B
  5. celery -A tasks worker -l info -n worker.%h -Q celery

启动定时任务

  1. [root@izwz920j4zsv1q15yhii1qz scripts]# celery -A tasks beat
  2. celery beat v4.1.1 (latentcall) is starting.
  3. __    -    ... __   -        _
  4. LocalTime -> 2018-05-27 19:39:29
  5. Configuration ->
  6.     . broker -> redis://47.106.106.220:5000/1
  7.     . loader -> celery.loaders.app.AppLoader
  8.     . scheduler -> celery.beat.PersistentScheduler
  9.     . db -> celerybeat-schedule
  10.     . logfile -> [stderr]@%WARNING
  11.     . maxinterval -> 5.00 minutes (300s)

在之前启动worker的三个窗口分别可以看到定时任务正在运行：

  1. celery -A tasks worker -l info -n workerA.%h -Q for_task_A
  3. [2018-05-27 19:41:27,432: INFO/ForkPoolWorker-1] Task tasks.taskA[60f41780-c9a2-477b-be46-6620ef07631f] succeeded in 0.00289130600868s: 11
  4. [2018-05-27 19:41:29,428: INFO/MainProcess] Received task: tasks.taskA[27220f52-dde2-471a-a87c-3f533d67217c]
  5. ......
  6. ......
  8. celery -A tasks worker -l info -n workerB.%h -Q for_task_B
  10. [2018-05-27 19:41:18,420: INFO/ForkPoolWorker-1] Task tasks.taskB[b6f9aee3-e6b4-4f10-9428-457d9bb844cf] succeeded in 0.00282042898471s: 60
  11. [2018-05-27 19:41:28,416: INFO/MainProcess] Received task: tasks.taskB[44dfea0b-b725-4874-bea2-9b66e8da573b]
  12. ......
  13. ......
  16. celery -A tasks worker -l info -n worker.%h -Q celery 
  18. [2018-05-27 19:41:23,428: INFO/ForkPoolWorker-1] Task tasks.add[315a9cca-3c95-4517-9289-2ece15cd46a4] succeeded in 0.00355823297286s: 3
  19. [2018-05-27 19:41:28,423: INFO/MainProcess] Received task: tasks.add[c4a1b2c7-ecb7-4af4-85c1-a341b3ec6726]
  20. ......
  21. ......