Celery 使用详解

前言

前段时间需要使用rabbitmq做写缓存，一直使用pika+rabbitmq的组合，pika这个模块虽然可以很直观地操作rabbitmq，但是官方给的例子太简单，对其底层原理了解又不是很深，遇到很多坑，尤其是需要自己写连接池管理和channel池管理。虽然也有用过celery，一直也是celery+redis的组合，涉及很浅；目前打算深研一下celery+redis+rabbitmq的使用。

celery + rabbitmq初步

我们先不在集成框架如flask或Django中使用celery，而仅仅单独使用。

简单介绍

Celery 是一个异步任务队列，一个Celery有三个核心组件：

Celery 客户端: 用于发布后台作业；当与 Flask 一起工作的时候，客户端与 Flask 应用一起运行。
Celery workers: 运行后台作业的进程。Celery 支持本地和远程的 workers，可以在本地服务器上启动一个单独的 worker，也可以在远程服务器上启动worker，需要拷贝代码；
消息代理: 客户端通过消息队列和 workers 进行通信，Celery 支持多种方式来实现这些队列。最常用的代理就是 RabbitMQ 和 Redis。

安装rabbitmq和redis

rabbitmq安装和配置参考：rabbitmq安装和配置
redis的安装和配置参考：redis的安装和配置
redis-py安装：

1	sudo pip install redis

redis-py操作redis参考：python操作redis

为了提高性能，官方推荐使用librabbitmq，这是一个连接rabbitmq的C++的库；

1 2	# 选择broker客户端、序列化和并发 sudo pip install celery[librabbitmq,redis,msgpack,gevent]

初步使用

一般我们使用redis做结果存储，使用rabbitmq做任务队列；

第一步：创建并发送一个异步任务

# 初始化

# tasks.py

from celery import Celery

app = Celery('tasks', broker='amqp://username:passwd@ip:port/varhost',backend='redis://username:passwd@ip:6390/db')

@app.task

def add(x, y):

return x + y

if __name__ == '__main__':

result = add.delay(30, 42)

# broker:任务队列的中间人；

# backend:任务执行结果的存储；

发生了什么事

app.task装饰add函数成一个Task实例，add.delay函数将task实例序列化后，通过librabbitmq库的方法将任务发送到rabbitmq；
该过程创建一个名字为celery的exchange交换机，类型为direct（直连交换机）;创建一个名为celery的queue，队列和交换机使用路由键celery绑定；
打开rabbitmq管理后台，可以看到有一条消息已经在celery队列中；

记住：当有多个装饰器的时候，app.task一定要在最外层；

扩展

如果使用redis作为任务队列中间人，在redis中存在两个键 celery和_kombu.binding.celery， _kombu.binding.celery表示有一名为 celery 的任务队列（Celery 默认），而键celery为默认队列中的任务列表，使用list类型，可以看看添加进去的任务数据。

第二步:开启worker执行任务

在项目目录下执行命令：

celery -A app.celery_tasks.celery worker -Q queue --loglevel=info

# -A参数指定创建的celery对象的位置，该app.celery_tasks.celery指的是app包下面的celery_tasks.py模块的celery实例，注意一定是初始化后的实例，后面加worker表示该实例就是任务执行者；

# -Q参数指的是该worker接收指定的队列的任务，这是为了当多个队列有不同的任务时可以独立；如果不设会接收所有的队列的任务；

# -l参数指定worker输出的日志级别；

任务执行完毕后结果存储在redis中，查看redis中的数据，发现存在一个string类型的键值对：

1	celery-task-meta-064e4262-e1ba-4e87-b4a1-52dd1418188f:data

该键值对的失效时间默认为24小时。

分析序列化的消息

add.delay将Task实例序列化后发送到rabbitmq，那么序列化的过程是怎样的呢？

下面是添加到rabbitmq任务队列中的消息数据，使用的是pickle模块对body部分的数据进行序列化：

{"body": "gAJ9cQAoWAQAAAB0YXNrcQFYGAAAAHRlc3RfY2VsZXJ5LmFkZF90b2dldGhlcnECWAIAAABpZHEDWCQAAAA2NmQ1YTg2Yi0xZDM5LTRjODgtYmM5OC0yYzE4YjJjOThhMjFxBFgEAAAAYXJnc3EFSwlLKoZxBlgGAAAAa3dhcmdzcQd9cQhYBwAAAHJldHJpZXNxCUsAWAMAAABldGFxCk5YBwAAAGV4cGlyZXNxC05YAwAAAHV0Y3EMiFgJAAAAY2FsbGJhY2tzcQ1OWAgAAABlcnJiYWNrc3EOTlgJAAAAdGltZWxpbWl0cQ9OToZxEFgHAAAAdGFza3NldHERTlgFAAAAY2hvcmRxEk51Lg==",

# body是序列化后使用base64编码的信息，包括具体的任务参数，其中包括了需要执行的方法、参数和一些任务基本信息

"content-encoding": "binary", # 序列化数据的编码方式

"content-type": "application/x-python-serialize", # 任务数据的序列化方式，默认使用python内置的序列化模块pickle

"headers": {},

"properties":

{"reply_to": "b7580727-07e5-307b-b1d0-4b731a796652", # 结果的唯一id

"correlation_id": "66d5a86b-1d39-4c88-bc98-2c18b2c98a21", # 任务的唯一id

"delivery_mode": 2,

"delivery_info": {"priority": 0, "exchange": "celery", "routing_key": "celery"}, # 指定交换机名称，路由键，属性

"body_encoding": "base64", # body的编码方式

"delivery_tag": "bfcfe35d-b65b-4088-bcb5-7a1bb8c9afd9"}}

将序列化消息反序列化

import pickle

import base64

result = base64.b64decode('gAJ9cQAoWAQAAAB0YXNrcQFYGAAAAHRlc3RfY2VsZXJ5LmFkZF90b2dldGhlcnECWAIAAABpZHEDWCQAAAA2NmQ1YTg2Yi0xZDM5LTRjODgtYmM5OC0yYzE4YjJjOThhMjFxBFgEAAAAYXJnc3EFSwlLKoZxBlgGAAAAa3dhcmdzcQd9cQhYBwAAAHJldHJpZXNxCUsAWAMAAABldGFxCk5YBwAAAGV4cGlyZXNxC05YAwAAAHV0Y3EMiFgJAAAAY2FsbGJhY2tzcQ1OWAgAAABlcnJiYWNrc3EOTlgJAAAAdGltZWxpbWl0cQ9OToZxEFgHAAAAdGFza3NldHERTlgFAAAAY2hvcmRxEk51Lg==')

print(pickle.loads(result))

# 结果

{

'task': 'test_celery.add_together', # 需要执行的任务

'id': '66d5a86b-1d39-4c88-bc98-2c18b2c98a21', # 任务的唯一id

'args': (9, 42), # 任务的参数

'kwargs': {},

'retries': 0,

'eta': None,

'expires': None, # 任务失效时间

'utc': True,

'callbacks': None, # 完成后的回调

'errbacks': None, # 任务失败后的回调

'timelimit': (None, None), # 超时时间

'taskset': None,

'chord': None

}

我们可以看到body里面有我们需要执行的函数的一切信息，celery的worker接收到消息后就会反序列化body数据，执行相应的方法。

常见的数据序列化方式

binary: 二进制序列化方式；python的pickle默认的序列化方法；

json:json 支持多种语言, 可用于跨语言方案，但好像不支持自定义的类对象；

XML:类似标签语言；

msgpack:二进制的类 json 序列化方案, 但比 json 的数据结构更小, 更快；

yaml:yaml 表达能力更强, 支持的数据类型较 json 多, 但是 python 客户端的性能不如 json

经过比较，为了保持跨语言的兼容性和速度，采用msgpack或json方式；

celery配置

celery的性能和许多因素有关，比如序列化的方式，连接rabbitmq的方式，多进程、单线程等等，我们可以指定配置；

基本配置项

CELERY_DEFAULT_QUEUE：默认队列

BROKER_URL : 代理人即rabbitmq的网址

CELERY_RESULT_BACKEND：结果存储地址

CELERY_TASK_SERIALIZER：任务序列化方式

CELERY_RESULT_SERIALIZER：任务执行结果序列化方式

CELERY_TASK_RESULT_EXPIRES：任务过期时间

CELERY_ACCEPT_CONTENT：指定任务接受的内容序列化类型(序列化)，一个列表；

加载配置

# main.py

from celery import Celery

import celeryconfig

app = Celery(__name__, include=["task"])

# 引入配置文件

app.config_from_object(celeryconfig)

if __name__ == '__main__':

result = add.delay(30, 42)

# task.py

from main import app

@app.task

def add(x, y):

return x + y

# celeryconfig.py

BROKER_URL = 'amqp://username:password@localhost:5672/yourvhost'

CELERY_RESULT_BACKEND = 'redis://localhost:6379/0'

CELERY_TASK_SERIALIZER = 'msgpack'

CELERY_RESULT_SERIALIZER = 'msgpack'

CELERY_TASK_RESULT_EXPIRES = 60 * 60 * 24 # 任务过期时间

CELERY_ACCEPT_CONTENT = ["msgpack"] # 指定任务接受的内容序列化的类型.

也可以直接加载配置

from celery import Celery

import celeryconfig

app = Celery(__name__, include=["task"])

app.conf.update(

task_serializer='json',

accept_content=['json'],

result_serializer='json',

timezone='Europe/Oslo',

enable_utc=True,

)

此外还有两个方法可以加载配置，但开发不会直接调用：

1 2	app.config_from_envvar() # 从环境变量加载 app.config_from_cmdline() # 从命令行加载

一份比较常用的配置文件

# 注意，celery4版本后，CELERY_BROKER_URL改为BROKER_URL

BROKER_URL = 'amqp://username:passwd@host:port/虚拟主机名'

# 指定结果的接受地址

CELERY_RESULT_BACKEND = 'redis://username:passwd@host:port/db'

# 指定任务序列化方式

CELERY_TASK_SERIALIZER = 'msgpack'

# 指定结果序列化方式

CELERY_RESULT_SERIALIZER = 'msgpack'

# 任务过期时间,celery任务执行结果的超时时间

CELERY_TASK_RESULT_EXPIRES = 60 * 20

# 指定任务接受的序列化类型.

CELERY_ACCEPT_CONTENT = ["msgpack"]

# 任务发送完成是否需要确认，这一项对性能有一点影响

CELERY_ACKS_LATE = True

# 压缩方案选择，可以是zlib, bzip2，默认是发送没有压缩的数据

CELERY_MESSAGE_COMPRESSION = 'zlib'

# 规定完成任务的时间

CELERYD_TASK_TIME_LIMIT = 5 # 在5s内完成任务，否则执行该任务的worker将被杀死，任务移交给父进程

# celery worker的并发数，默认是服务器的内核数目,也是命令行-c参数指定的数目

CELERYD_CONCURRENCY = 4

# celery worker 每次去rabbitmq预取任务的数量

CELERYD_PREFETCH_MULTIPLIER = 4

# 每个worker执行了多少任务就会死掉，默认是无限的

CELERYD_MAX_TASKS_PER_CHILD = 40

# 设置默认的队列名称，如果一个消息不符合其他的队列就会放在默认队列里面，如果什么都不设置的话，数据都会发送到默认的队列中

CELERY_DEFAULT_QUEUE = "default"

# 设置详细的队列

CELERY_QUEUES = {

"default": { # 这是上面指定的默认队列

"exchange": "default",

"exchange_type": "direct",

"routing_key": "default"

"topicqueue": { # 这是一个topic队列凡是topictest开头的routing key都会被放到这个队列

"routing_key": "topic.#",

"exchange": "topic_exchange",

"exchange_type": "topic",

"task_eeg": { # 设置扇形交换机

"exchange": "tasks",

"exchange_type": "fanout",

"binding_key": "tasks",

}

在celery4.0以后配置参数改成了小写，对于4.0以后的版本替代参数：

100

101

102

103

4.0版本以下参数 4.0版本以上配置参数

CELERY_ACCEPT_CONTENT accept_content

CELERY_ENABLE_UTC enable_utc

CELERY_IMPORTS imports

CELERY_INCLUDE include

CELERY_TIMEZONE timezone

CELERYBEAT_MAX_LOOP_INTERVAL beat_max_loop_interval

CELERYBEAT_SCHEDULE beat_schedule

CELERYBEAT_SCHEDULER beat_scheduler

CELERYBEAT_SCHEDULE_FILENAME beat_schedule_filename

CELERYBEAT_SYNC_EVERY beat_sync_every

BROKER_URL broker_url

BROKER_TRANSPORT broker_transport

BROKER_TRANSPORT_OPTIONS broker_transport_options

BROKER_CONNECTION_TIMEOUT broker_connection_timeout

BROKER_CONNECTION_RETRY broker_connection_retry

BROKER_CONNECTION_MAX_RETRIES broker_connection_max_retries

BROKER_FAILOVER_STRATEGY broker_failover_strategy

BROKER_HEARTBEAT broker_heartbeat

BROKER_LOGIN_METHOD broker_login_method

BROKER_POOL_LIMIT broker_pool_limit

BROKER_USE_SSL broker_use_ssl

CELERY_CACHE_BACKEND cache_backend

CELERY_CACHE_BACKEND_OPTIONS cache_backend_options

CASSANDRA_COLUMN_FAMILY cassandra_table

CASSANDRA_ENTRY_TTL cassandra_entry_ttl

CASSANDRA_KEYSPACE cassandra_keyspace

CASSANDRA_PORT cassandra_port

CASSANDRA_READ_CONSISTENCY cassandra_read_consistency

CASSANDRA_SERVERS cassandra_servers

CASSANDRA_WRITE_CONSISTENCY cassandra_write_consistency

CASSANDRA_OPTIONS cassandra_options

CELERY_COUCHBASE_BACKEND_SETTINGS couchbase_backend_settings

CELERY_MONGODB_BACKEND_SETTINGS mongodb_backend_settings

CELERY_EVENT_QUEUE_EXPIRES event_queue_expires

CELERY_EVENT_QUEUE_TTL event_queue_ttl

CELERY_EVENT_QUEUE_PREFIX event_queue_prefix

CELERY_EVENT_SERIALIZER event_serializer

CELERY_REDIS_DB redis_db

CELERY_REDIS_HOST redis_host

CELERY_REDIS_MAX_CONNECTIONS redis_max_connections

CELERY_REDIS_PASSWORD redis_password

CELERY_REDIS_PORT redis_port

CELERY_RESULT_BACKEND result_backend

CELERY_MAX_CACHED_RESULTS result_cache_max

CELERY_MESSAGE_COMPRESSION result_compression

CELERY_RESULT_EXCHANGE result_exchange

CELERY_RESULT_EXCHANGE_TYPE result_exchange_type

CELERY_TASK_RESULT_EXPIRES result_expires

CELERY_RESULT_PERSISTENT result_persistent

CELERY_RESULT_SERIALIZER result_serializer

CELERY_RESULT_DBURI 请result_backend改用。

CELERY_RESULT_ENGINE_OPTIONS database_engine_options

[...]_DB_SHORT_LIVED_SESSIONS database_short_lived_sessions

CELERY_RESULT_DB_TABLE_NAMES database_db_names

CELERY_SECURITY_CERTIFICATE security_certificate

CELERY_SECURITY_CERT_STORE security_cert_store

CELERY_SECURITY_KEY security_key

CELERY_ACKS_LATE task_acks_late

CELERY_TASK_ALWAYS_EAGER task_always_eager

CELERY_TASK_ANNOTATIONS task_annotations

CELERY_TASK_COMPRESSION task_compression

CELERY_TASK_CREATE_MISSING_QUEUES task_create_missing_queues

CELERY_TASK_DEFAULT_DELIVERY_MODE task_default_delivery_mode

CELERY_TASK_DEFAULT_EXCHANGE task_default_exchange

CELERY_TASK_DEFAULT_EXCHANGE_TYPE task_default_exchange_type

CELERY_TASK_DEFAULT_QUEUE task_default_queue

CELERY_TASK_DEFAULT_RATE_LIMIT task_default_rate_limit

CELERY_TASK_DEFAULT_ROUTING_KEY task_default_routing_key

CELERY_TASK_EAGER_PROPAGATES task_eager_propagates

CELERY_TASK_IGNORE_RESULT task_ignore_result

CELERY_TASK_PUBLISH_RETRY task_publish_retry

CELERY_TASK_PUBLISH_RETRY_POLICY task_publish_retry_policy

CELERY_QUEUES task_queues

CELERY_ROUTES task_routes

CELERY_TASK_SEND_SENT_EVENT task_send_sent_event

CELERY_TASK_SERIALIZER task_serializer

CELERYD_TASK_SOFT_TIME_LIMIT task_soft_time_limit

CELERYD_TASK_TIME_LIMIT task_time_limit

CELERY_TRACK_STARTED task_track_started

CELERYD_AGENT worker_agent

CELERYD_AUTOSCALER worker_autoscaler

CELERYD_CONCURRENCY worker_concurrency

CELERYD_CONSUMER worker_consumer

CELERY_WORKER_DIRECT worker_direct

CELERY_DISABLE_RATE_LIMITS worker_disable_rate_limits

CELERY_ENABLE_REMOTE_CONTROL worker_enable_remote_control

CELERYD_HIJACK_ROOT_LOGGER worker_hijack_root_logger

CELERYD_LOG_COLOR worker_log_color

CELERYD_LOG_FORMAT worker_log_format

CELERYD_WORKER_LOST_WAIT worker_lost_wait

CELERYD_MAX_TASKS_PER_CHILD worker_max_tasks_per_child

CELERYD_POOL worker_pool

CELERYD_POOL_PUTLOCKS worker_pool_putlocks

CELERYD_POOL_RESTARTS worker_pool_restarts

CELERYD_PREFETCH_MULTIPLIER worker_prefetch_multiplier

CELERYD_REDIRECT_STDOUTS worker_redirect_stdouts

CELERYD_REDIRECT_STDOUTS_LEVEL worker_redirect_stdouts_level

CELERYD_SEND_EVENTS worker_send_task_events

CELERYD_STATE_DB worker_state_db

CELERYD_TASK_LOG_FORMAT worker_task_log_format

CELERYD_TIMER worker_timer

CELERYD_TIMER_PRECISION worker_timer_precision

Celery对象

核心的对象就是Celery了，初始化方法：

class Celery(object):

def __init__(self, main=None, loader=None, backend=None,

amqp=None, events=None, log=None, control=None,

set_as_current=True, accept_magic_kwargs=False,

tasks=None, broker=None, include=None, changes=None,

config_source=None, fixups=None, task_cls=None,

autofinalize=True, **kwargs):

# 常用的需要配置的参数

main:如果作为__main__运行，则为主模块的名称。用作自动生成的任务名称的前缀

loader:当前加载器实例。

backend:任务结果url；

amqp:AMQP对象或类名，一般不管；

log:日志对象或类名；

set_as_current:将本实例设为全局当前应用

tasks:任务注册表。

broker:使用的默认代理的URL,任务队列；

include:每个worker应该导入的模块列表，以实例创建的模块的目录作为起始路径；

这些参数都是celery实例化的配置，我们也可以不写，然后使用config_from_object方法加载配置；

创建异步任务的方法task

任何被task修饰的方法都会被创建一个Task对象，变成一个可序列化并发送到远程服务器的任务；它有多种修饰方式：

使用默认的参数

@celery.task

def function_name():

pass

指定相关参数

@celery.task(bind=True, name='name')

def function_name():

pass

# task方法参数

name:可以显式指定任务的名字；默认是模块的命名空间中本函数的名字。

serializer：指定本任务的序列化的方法；

bind:一个bool值，设置是否绑定一个task的实例，如果绑定，task实例会作为参数传递到任务方法中，可以访问task实例的所有的属性，即前面反序列化中那些属性

base:定义任务的基类，可以以此来定义回调函数，默认是Task类，我们也可以定义自己的Task类

default_retry_delay：设置该任务重试的延迟时间，当任务执行失败后，会自动重试，单位是秒，默认3分钟；

autoretry_for：设置在特定异常时重试任务，默认False即不重试；

retry_backoff：默认False，设置重试时的延迟时间间隔策略；

retry_backoff_max：设置最大延迟重试时间，默认10分钟，如果失败则不再重试；

retry_jitter：默认True，即引入抖动，避免重试任务集中执行；

# 当bind=True时，add函数第一个参数是self，指的是task实例

@task(bind=True) # 第一个参数是self，使用self.request访问相关的属性

def add(self, x, y):

try:

logger.info(self.request.id)

except:

self.retry() # 当任务失败则进行重试

自定义Task基类

import celery

class MyTask(celery.Task):

# 任务失败时执行

def on_failure(self, exc, task_id, args, kwargs, einfo):

print('{0!r} failed: {1!r}'.format(task_id, exc))

# 任务成功时执行

def on_success(self, retval, task_id, args, kwargs):

pass

# 任务重试时执行

def on_retry(self, exc, task_id, args, kwargs, einfo):

pass

@task(base=MyTask)

def add(x, y):

raise KeyError()

#方法相关的参数

exc:失败时的错误的类型；

task_id:任务的id；

args:任务函数的参数；

kwargs:键值对参数；

einfo:失败或重试时的异常详细信息；

retval:任务成功执行的返回值；

Task的一般属性

Task.name:任务名称；

Task.request：当前任务的信息；

Task.max_retries：设置重试的最大次数

Task.throws：预期错误类的可选元组，不应被视为实际错误，而是结果失败；

Task.rate_limit：设置此任务类型的速率限制

Task.time_limit：此任务的硬限时（以秒为单位）。

Task.ignore_result：不存储任务状态。默认False；

Task.store_errors_even_if_ignored：如果True，即使任务配置为忽略结果，也会存储错误。

Task.serializer：标识要使用的默认序列化方法的字符串。

Task.compression：标识要使用的默认压缩方案的字符串。默认为task_compression设置。

Task.backend：指定该任务的结果存储后端用于此任务。

Task.acks_late：如果设置True为此任务的消息将在任务执行后确认，而不是在执行任务之前（默认行为），即默认任务执行之前就会发送确认；

Task.track_started：如果True任务在工作人员执行任务时将其状态报告为“已启动”。默认是False；

调用异步任务

调用异步任务有三个方法，如下：

task.delay():这是apply_async方法的别名,但接受的参数较为简单；

task.apply_async(args=[arg1, arg2], kwargs={key:value, key:value})：可以接受复杂的参数

send_task():可以发送未被注册的异步任务，即没有被celery.task装饰的任务；

app.send_task

# tasks.py

from celery import Celery

app = Celery()

def add(x,y):

return x+y

app.send_task('tasks.add',args=[3,4]) # 参数基本和apply_async函数一样

# 但是send_task在发送的时候是不会检查tasks.add函数是否存在的，即使为空也会发送成功，所以celery执行是可能找不到该函数报错；

Task.delay

delay方法是apply_async方法的简化版，不支持执行选项，只能传递任务的参数。

@app.task

def add(x, y, z=0):

return x + y

add.delay(30,40,z=5) # 包括位置参数和关键字参数

Task.apply_async

apply_async支持执行选项，它会覆盖全局的默认参数和定义该任务时指定的执行选项，本质上还是调用了send_task方法；

add.apply_async(args=[30,40], kwargs={'z':5})

# 其他参数

task_id:为任务分配唯一id，默认是uuid;

countdown : 设置该任务等待一段时间再执行，单位为s；

eta : 定义任务的开始时间；eta=time.time()+10;

expires : 设置任务时间，任务在过期时间后还没有执行则被丢弃；

retry : 如果任务失败后, 是否重试;使用true或false，默认为true

shadow：重新指定任务的名字str，覆盖其在日志中使用的任务名称；

retry_policy : {},重试策略.如下：

max_retries : 最大重试次数, 默认为 3 次.

interval_start : 重试等待的时间间隔秒数, 默认为 0 , 表示直接重试不等待.

interval_step : 每次重试让重试间隔增加的秒数, 可以是数字或浮点数, 默认为 0.2

interval_max : 重试间隔最大的秒数, 即通过 interval_step 增大到多少秒之后, 就不在增加了, 可以是数字或者浮点数, 默认为 0.2 .

routing_key:自定义路由键；

queue：指定发送到哪个队列；

exchange：指定发送到哪个交换机；

priority：任务队列的优先级，0到255之间，对于rabbitmq来说0是最高优先级；

serializer：任务序列化方法；通常不设置；

compression：压缩方案，通常有zlib, bzip2

headers：为任务添加额外的消息；

link：任务成功执行后的回调方法；是一个signature对象；可以用作关联任务；

link_error: 任务失败后的回调方法，是一个signature对象；

# 如下

add.apply_async((2, 2), retry=True, retry_policy={

'max_retries': 3,

'interval_start': 0,

'interval_step': 0.2,

'interval_max': 0.2,

})

自定义发布者,交换机,路由键, 队列, 优先级,序列方案和压缩方法:

task.apply_async((2,2),

compression='zlib',

serialize='json',

queue='priority.high',

routing_key='web.add',

priority=0,

exchange='web_exchange')

获取任务结果和状态

由于celery发送的都是去其他进程执行的任务，如果需要在客户端监控任务的状态，有如下方法：

r = task.apply_async()

r.ready() # 查看任务状态，返回布尔值, 任务执行完成, 返回 True, 否则返回 False.

r.wait() # 会阻塞等待任务完成, 返回任务执行结果，很少使用；

r.get(timeout=1) # 获取任务执行结果，可以设置等待时间，如果超时但任务未完成返回None；

r.result # 任务执行结果，未完成返回None；

r.state # PENDING, START, SUCCESS，任务当前的状态

r.status # PENDING, START, SUCCESS，任务当前的状态

r.successful # 任务成功返回true

r.traceback # 如果任务抛出了一个异常，可以获取原始的回溯信息

但是一般业务中很少用到，因为获取任务执行的结果需要阻塞，celery使用场景一般是不关心结果的。

使用celery

# seting.py

# 设置配置

BROKER_URL = 'amqp://username:password@localhost:5672/yourvhost'

CELERY_RESULT_BACKEND = 'redis://localhost:6379/0'

CELERY_TASK_SERIALIZER = 'msgpack'

CELERY_RESULT_SERIALIZER = 'msgpack'

CELERY_TASK_RESULT_EXPIRES = 60 * 60 * 24

CELERY_ACCEPT_CONTENT = ["msgpack"]

CELERY_DEFAULT_QUEUE = "default"

CELERY_QUEUES = {

"default": { # 这是上面指定的默认队列

"exchange": "default",

"exchange_type": "direct",

"routing_key": "default"

}

# app.py --- 初始化celery对象

from celery import Celery

import seting

from task import test_one, test_two

celery = Celery(__name__, include=["task"]) # 设置需要导入的模块

# 引入配置文件

celery.config_from_object(seting)

if __name__ == '__main__':

test_one.apply_async((2,2),

routing_key='default',

priority=0,

exchange='default')

# task.py --- 定义需要执行的任务

from app import celery

@celery.task

def test_one(x, y):

return x + y

@celery.task(name="one_name")

def test_two(x, y):

return x * y

本文为转载文章，贵在分享，版权归原作者及原出处所有，如涉及版权等问题，请及时与我联系。
原文出处：天宇之游
原文链接：https://www.cnblogs.com/cwp-bg/p/8759638.html

前言

celery + rabbitmq初步

简单介绍

安装rabbitmq和redis

初步使用

第一步：创建并发送一个异步任务

第二步:开启worker执行任务

分析序列化的消息

将序列化消息反序列化

celery配置

基本配置项

加载配置

一份比较常用的配置文件

Celery对象

创建异步任务的方法task

Task的一般属性

调用异步任务

app.send_task

Task.delay

Task.apply_async

获取任务结果和状态

使用celery

相关推荐

发表评论