初识Celery

Posted 2020-09-17 茁壮的小草

本系列文章的开发环境：

window 7 + python2.7 + pycharm5 + celery3.1.25 + django1.9.4

在我们日常的开发工作中，经常会遇到这几种情况：

1、在web应用中，用户触发一个操作，执行后台处理程序，这个程序需要执行很长时间才能返回结果。怎样才能不阻塞http请求，不让用户等待从而提高用户体验呢？
2、定时任务脚本：生产环境经常会跑一些定时任务脚本，假如你有上千台的服务器、上千种任务，定时任务的管理很困难，如何对job进行有效的管理？
3、异步需求：比如发送短信/邮件、推送消息、清理/设置缓存？

如果你有以上的需求，那么Celery可能对你很有用。

Celery - 分布式任务队列系统

Celery是一个可以处理大量消息的分布式任务系统，它凭借简单、灵活、可靠的特性被广泛使用。Celery聚焦于实时处理任务，同时也支持定时的任务调度。

1、特性：

• 查看定时任务的执行情况，比如执行是否成功、当前状态、执行任务花费的时间等。
• 易于其他框架集成，如使用django管理后台添加、更新、删除任务。
• 方便把任务和配置管理相关联。
• 可选多进程、Eventlet和Gevent三种模式并发执行。
• 提供错误处理机制。

2、架构图

 

从上图中可以知道Celery包含如下组件：

• Producer：凡是调用了Celery API、函数或装饰器而产生任务并交给任务队列处理的都是任务生产者。
• 任务调度组件：Beat进程会读取配置文件的内容，周期性地将配置中到期需要执行的任务发送给任务队列。
• Celery Worker：负责执行任务的线程，可以在多台服务器运行提高执行效率。
• Broker：消息中间件，负责接受任务生产者的任务，并且转发work进行执行。Celery目前支持RabbitMQ、Redis、MongoDB、Beanstalk、SQLAlchemy、Zookeeper等作为消息中间件，官方推荐使用RabbitMQ。
• Result Backend：任务处理完后保存状态信息和结果，以供查询。Celery默认已支持Redis、RabbitMQ、MongoDB、Django ORM、SQLAlchemy等方式。

开始使用Celery 

这里Broker和Result Backend都选择RabbitMQ。

1、安装

     1） 安装RabbitMQ

     2)  安装Celery 3.1.25

         为什么选择这个低版本？请见最下面的问题列表。

pip install celery==3.1.25

2、一个简单例子

　　2.1） 在一个目录中创建tasks.py文件，内容如下：

from celery import Celery

#创建celery实例，其中backend表示采用rpc瞬态信息，不保存到数据库；broker表示连接RabbitMQ URL
app = Celery(\'tasks\',backend=\'rpc://\',broker=\'pyamqp://guest@localhost//\')

@app.task
def hello():
    return "hello celery"

     2.2） 启动celery worker

              到tasks.py文件那层目录，执行以下命令：

celery -A tasks worker --loglevel=info

启动输出信息如下：

E:\\workdir\\test_pro>celery -A tasks worker --loglevel=info
[2017-05-10 18:26:18,298: WARNING/MainProcess] c:\\python27\\lib\\site-packages\\celery\\apps\\worker.py:161: CDeprecationWarnin
Starting from version 3.2 Celery will refuse to accept pickle by default.

The pickle serializer is a security concern as it may give attackers
the ability to execute any command.  It\'s important to secure
your broker from unauthorized access when using pickle, so we think
that enabling pickle should require a deliberate action and not be
the default choice.

If you depend on pickle then you should set a setting to disable this
warning and to be sure that everything will continue working
when you upgrade to Celery 3.2::

    CELERY_ACCEPT_CONTENT = [\'pickle\', \'json\', \'msgpack\', \'yaml\']

You must only enable the serializers that you will actually use.


  warnings.warn(CDeprecationWarning(W_PICKLE_DEPRECATED))

 -------------- celery@507B9D97E083 v3.1.25 (Cipater)
---- **** -----
--- * ***  * -- Windows-7-6.1.7601-SP1
-- * - **** ---
- ** ---------- [config]
- ** ---------- .> app:         tasks:0x35fc240
- ** ---------- .> transport:   amqp://guest:**@localhost:5672//
- ** ---------- .> results:     rpc://
- *** --- * --- .> concurrency: 4 (prefork)
-- ******* ----
--- ***** ----- [queues]
 -------------- .> celery           exchange=celery(direct) key=celery


[tasks]
  . tasks.hello

[2017-05-10 18:26:18,433: INFO/MainProcess] Connected to amqp://guest:**@127.0.0.1:5672//
[2017-05-10 18:26:18,483: INFO/MainProcess] mingle: searching for neighbors
[2017-05-10 18:26:19,497: INFO/MainProcess] mingle: all alone
[2017-05-10 18:26:19,523: WARNING/MainProcess] celery@507B9D97E083 ready.

View Code

     2.3） 测试结果

           另起一个终端，还是到tasks.py那层目录，进入python命令行：

 

这时celery worker会有提示信息：

 

 到此为止，celery入门就介绍到这里了，下一节介绍如何在django中使用celery。

 

 

遇到的问题列表： 

 1、启动celery worker时报错

ERROR:
`[2016-11-05 20:23:25,759: CRITICAL/MainProcess] Unrecoverable error: TypeError(\'must be integer, not _subprocess_handle\',)
Traceback (most recent call last):
File "c:\\development\\env\\zillow2\\lib\\site-packages\\celery\\worker\\worker.py", line 203, in start
self.blueprint.start(self)
File "c:\\development\\env\\zillow2\\lib\\site-packages\\celery\\bootsteps.py", line 119, in start
step.start(parent)
File "c:\\development\\env\\zillow2\\lib\\site-packages\\celery\\bootsteps.py", line 370, in start
return self.obj.start()
File "c:\\development\\env\\zillow2\\lib\\site-packages\\celery\\concurrency\\base.py", line 131, in start
self.on_start()
File "c:\\development\\env\\zillow2\\lib\\site-packages\\celery\\concurrency\\prefork.py", line 112, in on_start
**self.options)
File "c:\\development\\env\\zillow2\\lib\\site-packages\\billiard\\pool.py", line 1008, in init
self._create_worker_process(i)
File "c:\\development\\env\\zillow2\\lib\\site-packages\\billiard\\pool.py", line 1117, in _create_worker_process
w.start()
File "c:\\development\\env\\zillow2\\lib\\site-packages\\billiard\\process.py", line 122, in start
self._popen = self._Popen(self)
File "c:\\development\\env\\zillow2\\lib\\site-packages\\billiard\\context.py", line 383, in _Popen
return Popen(process_obj)
File "c:\\development\\env\\zillow2\\lib\\site-packages\\billiard\\popen_spawn_win32.py", line 64, in init
_winapi.CloseHandle(ht)
TypeError: must be integer, not _subprocess_handle

(zillow2) C:\\DEVELOPMENT\\zillow2\\project>Traceback (most recent call last):
File "", line 1, in
File "c:\\development\\env\\zillow2\\lib\\site-packages\\billiard\\spawn.py", line 159, in spawn_main
new_handle = steal_handle(parent_pid, pipe_handle)
File "c:\\development\\env\\zillow2\\lib\\site-packages\\billiard\\reduction.py", line 121, in steal_handle
_winapi.PROCESS_DUP_HANDLE, False, source_pid)
WindowsError: [Error 87] The parameter is incorrect

解决方案：这是因为windows系统不支持celery4以上版本，请降级到3.xx即可解决该问题

 

 

 

 

 

 参考文档：https://zhuanlan.zhihu.com/p/22304455