Tips:在学习Celery过程中，使用的系统为Windows 10、Celery版本为3.1.18①、中间人使用RabbitMQ。
C:\Users\foolf>celery --version
3.1.18 (Cipater)
什么是任务队列
任务队列是一种在线程或者机器之间分发任务的机制。
消息队列的输入是工作的一个单元，称为任务，独立的职程(Worker)进程持续监视队列中是否有需要处理的新任务。
Celery使用消息通信，通信一般使用中间人(Broker)在客户端和职程之间斡旋。这个过程从客户端想队列中添加消息开始，之后中间人将消息派送给职程。
Celery是Python编写的，但协议可以使用任何语言实现。
需要什么
Celery需要一个发送和接受消息的传述者。RabbbitMQ和Redis中间人的消息支持所有的特性，我们主要是使用RabbitMQ作中间人(关于中间人RabbitMQ的安装可以网上搜索，有很多详细的教程)。
Celery优势
在程序运行过程中，我们经常会遇到一些耗时耗资源的操作，为了避免阻塞主程序，我们会采用异步或者多线程来处理任务。比如在主程序中调用一个函数，并从该函数中获取函数返回值。如果这个函数不能很快执行完成并返回，那么主程序就会阻塞，知直到函数返回。
Celery是一个强大的分布式任务队列，它可以让人物的执行完全脱离主程序，甚至可以被分配到其他的主机上运行。
Celery架构：
架构
从图上可以看出Celery包含几个模块：
- 任务模块
主要包异步任务和定时任务，异步任务通常在业务逻辑中被触发并发送到任务队列中，而定时任务是由Celery Beat进程周期性的将任务发往任务队列。 - 消息中间件Broker
Broker就是任务调度队列，接受任务生产者发送来的消息，将任务存入队列，之所以需要中间人的原因是Celrey本身是不提供消息队列的服务，所以需要第三方组件实现。 - 任务执行单元Worker
Worker是执行任务的单元，它实时监控消息队列，如果有任务就获取任务并执行它。 - 任务存储Backend
Backend用于存储任务只想的结果，存储可以使用RabbitMQ或者Redis或者数据库等。
安装Celery
Celery已经提交到Pypi上，所以我们可是使用Python的工具pip来安装。
pip install celery==3.1.18
上面的安装命令，如果没有指定版本，系统会默认安装最新版本的Celery，但是这里可能在后面的学习中遇到问题。关于问题放到最后解释，暂且我们先安装3.1.18版本的Celery。
创建Celery实例
如果你已经安装好了Celery，那么现在就可以创建Celery实例了：
tasks.py
# coding:utf-8
from celery import Celery
app = Celery('tasks', broker='amqp://guest@localhost//')
@app.task
def add(x, y):
return x + y
Celery的第一个参数是当前模块名称，这个参数是必须的，第二个参数是中间人关键字参数，指定我们所使用的的消息中间人的URL，这里我们使用的是RabbitMQ。我们定义了一个单一的任务，称为add,返回两个数字的和。
启动Celery职程服务器(Worker)
celery -A tasks worker --loglevel=info
参数-A指定了Celery实例的位置，这个实例是在tasks.py文件中，Celery会自动在该文件中查找Celery对象实例。
--loglevel指定日志的级别，默认是warning。
如果启动正常，就会看到下面的输出。
running
调用任务
现在我们已经开启了一个Worker了，这样我们可以在应用程序中使用 delay()或者 apply_async()方法来调用任务。
在tasks.py文件所在的目录打开终端。
>>> from tasks import add
>>> add.delay(2, 8)
<AsyncResult: 1b50f449-8fa2-478a-9eea-561a3c29fd43>
>>>
我们先从tasks.py文件中导入add任务对象，然后使用delay()方法将任务发送到消息中间件，我们之前开启的那个Worker会一直监控任务队列，知道有任务到来，就会执行。
我们到Worker中可以看到多了几条日志信息：
[2017-03-09 19:45:35,351: INFO/MainProcess] Received task: tasks.add[1b50f449-8fa2-478a-9eea-561a3c29fd43]
[2017-03-09 19:45:40,920: INFO/MainProcess] Task tasks.add[1b50f449-8fa2-478a-9eea-561a3c29fd43] succeeded in 5.56299996376s: 10
说明我们的任务被调度并执行成功了。
获得结果
刚我们在命令行中调用任务，很明显任务执行完成，但是我们并不知道任务执行后得到的结果是什么。如果我们想获得执行后的结果可以这样：
>>> result = add.delay(2, 8)
>>> result.ready() # 查看任务执行的状态，此刻任务没有执行完成，显示False
False
>>> result.ready()
True # 表示任务已经执行完成
>>> result.get() # 获取任务的执行结果
10
>>>
注解：
①：之前我是在windows下学习的Celery，安装的Celery版本是4.0.2；在运行Worker过程中遇到如下ed错误：
I:\Celery\celery-examples>celery -A tasks worker --loglevel=info
-------------- celery@DESKTOP-N53SFFK v4.0.2 (latentcall)
---- **** -----
--- * *** * -- Windows-10-10.0.14393 2017-02-28 00:32:22
-- * - **** ---
- ** ---------- [config]
- ** ---------- .> app: tasks:0x4700908
- ** ---------- .> transport: amqp://guest:**@localhost:5672//
- ** ---------- .> results: disabled://
- *** --- * --- .> concurrency: 4 (prefork)
-- ******* ---- .> task events: OFF (enable -E to monitor tasks in this worker)
--- ***** -----
-------------- [queues]
.> celery exchange=celery(direct) key=celery
[tasks]
. tasks.add
[2017-02-28 00:32:22,619: CRITICAL/MainProcess] Unrecoverable error: TypeError('argument 1 must be an integer, not _subprocess_handle',)
Traceback (most recent call last):
File "c:\python27\lib\site-packages\celery\worker\worker.py", line 203, in start
self.blueprint.start(self)
File "c:\python27\lib\site-packages\celery\bootsteps.py", line 119, in start
step.start(parent)
File "c:\python27\lib\site-packages\celery\bootsteps.py", line 370, in start
return self.obj.start()
File "c:\python27\lib\site-packages\celery\concurrency\base.py", line 131, in start
self.on_start()
File "c:\python27\lib\site-packages\celery\concurrency\prefork.py", line 112, in on_start
**self.options)
File "c:\python27\lib\site-packages\billiard\pool.py", line 1008, in __init__
self._create_worker_process(i)
File "c:\python27\lib\site-packages\billiard\pool.py", line 1117, in _create_worker_process
w.start()
File "c:\python27\lib\site-packages\billiard\process.py", line 122, in start
self._popen = self._Popen(self)
File "c:\python27\lib\site-packages\billiard\context.py", line 383, in _Popen
return Popen(process_obj)
File "c:\python27\lib\site-packages\billiard\popen_spawn_win32.py", line 64, in __init__
_winapi.CloseHandle(ht)
TypeError: argument 1 must be an integer, not _subprocess_handle
I:\Celery\celery-examples>Traceback (most recent call last):
File "<string>", line 1, in <module>
File "c:\python27\lib\site-packages\billiard\spawn.py", line 159, in spawn_main
new_handle = steal_handle(parent_pid, pipe_handle)
File "c:\python27\lib\site-packages\billiard\reduction.py", line 121, in steal_handle
_winapi.PROCESS_DUP_HANDLE, False, source_pid)
WindowsError: [Error 87]
经过搜索发现是因为winsows是不支持celery4的。参照的回答在这https://github.com/celery/celery/issues/3551
所以我机制的将版本降低到3，运行正常。记录下来仅仅是避免其他人在学习中不会再这个小问题上浪费时间。
END
由于学习的还是celery的基础，所以后面的更复杂的内容等学了再更。
参考文章：
http://docs.jinkan.org/docs/celery/index.html
http://www.guodongkeji.com/newsshow-24-2135-1.html
网友评论