石家庄征婚,新标准小学英语教案,焦灵花
由于从事区块链钱包相关开发，对于区块链链上资源需要频繁的进行检查同步，在flask项目中，对于celery这个异步任务执行工具，使用的频率算是相当的高，今天，我就来简单总结一下celery的概念和使用方法。
celery是一个异步任务的调度工具，是distributed task queue，分布式任务队列，分布式决定了可以有多个worker的存在，队列表示其是异步操作，即存在一个产生任务提出需求的工头，和一群等着被分配工作的码农。
在python中定义celery的时候，我们要引入broker，中文翻译过来就是"中间人"的意思，在这里broker起到一个中间人的角色，在工头提出任务的时候，把所有的任务放到broker里面，在broker的另一头，一群码农等着取出一个个任务准备着手做。
这种模式注定了整个系统会是个开环系统，工头对于码农们把任务做的怎样是不知情的，所以我们要引入backend来保存每次任务的结果。这个backend有点像我们的broker，也是存储信息用的，只不过这里存的是那些任务的返回结果。我们可以选择只让错误执行的任务返回结果到backend，这样我们取回结果，便可以知道有多少任务执行失败了。
broker是一个消息传输的中间件，它是用来存储生产出来的各种待执行任务的。每当应用程序调用celery的异步任务的时候，会向broker传递消息，而后celery的worker将会取到消息，进行程序执行，broker可以看成是一个消息队列，其中broker的中文意思是经纪人，用来发送和接受信息。这个broker有几个方案可供选择：rabbitmq（消息队列），redis（缓存数据库），数据库（不推荐），等等。
通常程序发送的消息，发完就完了，可能都不知道对方什么时候接受了，为此，celery实现了一个backend，用于存储这些消息以及celery执行的一些消息和结果，backend是在celery的配置中的一个配置项celery_result_backend，作用是保存结果和状态，如果你需要跟踪任务的状态，那么需要设置这一项，可以是database backend，也可以是cache backend。
对于brokers，官方推荐是rabbitmq和redis，至于backend，就是数据库，为了简单可以都使用redis。在我的项目中，都是使用redis。
celery的架构由消息中间件(message broker)，任务执行单元(worker)和任务执行结果存储(task result store)三部分组成。
celery本身不提供消息服务，但是可以方便的和第三方提供的消息中间件集成，包括rabbitmq，redis，mongodb等
worker是celery提供的任务执行的单元，worker并发的运行在分布式的系统节点中。
task result store用来存储worker执行的任务的结果，celery支持以不同方式存储任务的结果，包括amqp，redis，memcached，mongodb，sqlalchemy，django等
安装redis,它的安装比较简单：
~$ pip install redis
然后进行配置，一般都在项目的config.py文件里配置：
celery_broker_url = "redis://localhost:6379/0"
url的格式为：redis://:password@hostname:port/db_number
url scheme后的所有字段都是可选的，并且默认为localhost的6379端口，也就是redis的默认端口，使用数据库0。
安装celery：
~$ pip install celery
使用celery包含三个方面：1，定义任务函数 2，运行celery服务 3，客户应用程序的调用
创建一个文件tasks.py输入下列代码：
我要评论from celery import celery
broker = "redis://localhost:6379/0"
backend = "redis://localhost:6379/1"
app = celery("tasks", broker=broker, backend=backend)
@app.task
def add(x, y)
return x + y
上述代码导入了celery，然后创建了celery实例app,实例化的过程中指定了任务名tasks（和文件名一致），传入了broker和backend。然后创建了一个任务函数add。下面启动
celery服务，在当前命令行终端运行：
~$ celery -a tasks worker
目录结构（celery -a tasks worker --loglevel=info这条命令当前工作目录必须和tasks.py所在的目录相同，即进入tasks.py所在目录执行这条命令）
调用delay函数即可启动add这个任务，这个函数的效果是发送一条消息到broker中去，这个消息包括要执行的函数，函数的参数以及其他消息，具体的可以看celery官方文档。这个时候worker会等待broker中的消息，一旦收到消息就会立刻执行消息。
注意：如果把返回值赋值给一个变量，那么原来的应用程序也会被阻塞，需要等待异步任务返回的结果，因此，实际使用中，不需要把结果赋值。
使用配置文件
celery的配置比较多，可以在官方配置文档：http://docs.celeryproject.org/en/latest/userguide/configuration.html 查询每个配置项的含义。
上述的使用是简单的配置，下面介绍一个更健壮的方式来使用celery。首先创建一个python包，celery服务，姑且命名为proj。目录文件如下：
|- proj |-- __init__.py |-- celery.py # 创建celery实例 |-- config.py # 配置文件 |-- tasks.py # 任务函数
首先是 celery.py
from __future__ import absolute_import
from celery import celery
app = celery("proj", include=["proj.tasks"])
app.config_from_object("proj.config")
if __name__ == "__main__":
app.start()
这一次创建app，并没有直接指定broker和backend。而是在配置文件中。
然后是 config.py
from __future__ import absolute_import broker_url = "redis://localhost:6379/0" celery_backend_url = "redis://localhost:6379/1"
最后是 tasks.py
from __future__ import absolute_import
from proj.celery import app
@app.task
def add(x, y):
return x + y
使用方法也很简单，在proj的同一级目录执行celery:
celery -a proj worker -l info
现在使用任务也很简单，直接在客户端代码调用proj.tasks里的函数即可。
scheduler(定时任务，周期性任务)
一种常见的需求是每隔一段时间执行一个任务
在celery中执行定时任务非常简单，只需要设置celery对象的celerybeat_schedule属性即可。
配置如下 config.py
from __future__ import absolute_import
broker_url = "redis://localhost:6379/0"
celery_backend_url = "redis://localhost:6379/1"
celery_timezone = "asia/shanghai"
from datetime import timedelta
celerybeat_schedule = {
'add-every-30-seconds':{
'task':'proj.tasks.add',
'schedule':timedelta(seconds=30),
'args':(16, 16)
},
}
注意配置文件需要指定时区，这段代码表示每隔30秒执行add函数，一旦使用了scheduler，启动celery需要加上-b参数。
celery -a proj worker -b -l info
对于celery的介绍就到这里了，欢迎交流技术难点。
如对本文有疑问，请在下面进行留言讨论，广大热心网友会与你互动！！ 点击进行留言回复
Python爬虫:Request Payload和Form Data的简单区别说明
浅谈Python中threading join和setDaemon用法及区别说明
Python3-异步进程回调函数(callback())介绍
python继承threading.Thread实现有返回值的子类实例
Python中使用threading.Event协调线程的运行详解
网友评论