前言
前段时间需要使用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做任务队列;
第一步:创建并发送一个异步任务
1 2 3 4 5 6 7 8 9 10 11 12 13 14 | # 初始化 # 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执行任务
在项目目录下执行命令:
1 2 3 4 5 | 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部分的数据进行序列化:
1 2 3 4 5 6 7 8 9 10 11 12 | {"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"}} |
将序列化消息反序列化
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 | 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数据,执行相应的方法。
- 常见的数据序列化方式
1 2 3 4 5 | binary: 二进制序列化方式;python的pickle默认的序列化方法; json:json 支持多种语言, 可用于跨语言方案,但好像不支持自定义的类对象; XML:类似标签语言; msgpack:二进制的类 json 序列化方案, 但比 json 的数据结构更小, 更快; yaml:yaml 表达能力更强, 支持的数据类型较 json 多, 但是 python 客户端的性能不如 json |
经过比较,为了保持跨语言的兼容性和速度,采用msgpack或json方式;
celery配置
celery的性能和许多因素有关,比如序列化的方式,连接rabbitmq的方式,多进程、单线程等等,我们可以指定配置;
基本配置项
1 2 3 4 5 6 7 | CELERY_DEFAULT_QUEUE:默认队列 BROKER_URL : 代理人即rabbitmq的网址 CELERY_RESULT_BACKEND:结果存储地址 CELERY_TASK_SERIALIZER:任务序列化方式 CELERY_RESULT_SERIALIZER:任务执行结果序列化方式 CELERY_TASK_RESULT_EXPIRES:任务过期时间 CELERY_ACCEPT_CONTENT:指定任务接受的内容序列化类型(序列化),一个列表; |
加载配置
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 | # 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"] # 指定任务接受的内容序列化的类型. |
也可以直接加载配置
1 2 3 4 5 6 7 8 9 10 | 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() # 从命令行加载 |
一份比较常用的配置文件
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 | # 注意,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以后的版本替代参数:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 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了,初始化方法:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 | 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对象,变成一个可序列化并发送到远程服务器的任务;它有多种修饰方式:
- 使用默认的参数
1 2 3 | @celery.task def function_name(): pass |
- 指定相关参数
1 2 3 4 5 6 7 8 9 10 11 12 13 14 | @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,即引入抖动,避免重试任务集中执行; |
1 2 3 4 5 6 7 | # 当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基类
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 | 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的一般属性
1 2 3 4 5 6 7 8 9 10 11 12 13 | 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; |
调用异步任务
调用异步任务有三个方法,如下:
1 2 3 | task.delay():这是apply_async方法的别名,但接受的参数较为简单; task.apply_async(args=[arg1, arg2], kwargs={key:value, key:value}):可以接受复杂的参数 send_task():可以发送未被注册的异步任务,即没有被celery.task装饰的任务; |
app.send_task
1 2 3 4 5 6 7 8 | # 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方法的简化版,不支持执行选项,只能传递任务的参数。
1 2 3 4 5 | @app.task def add(x, y, z=0): return x + y add.delay(30,40,z=5) # 包括位置参数和关键字参数 |
Task.apply_async
apply_async支持执行选项,它会覆盖全局的默认参数和定义该任务时指定的执行选项,本质上还是调用了send_task方法;
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 | 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, }) |
- 自定义发布者,交换机,路由键, 队列, 优先级,序列方案和压缩方法:
1 2 3 4 5 6 7 | task.apply_async((2,2), compression='zlib', serialize='json', queue='priority.high', routing_key='web.add', priority=0, exchange='web_exchange') |
获取任务结果和状态
由于celery发送的都是去其他进程执行的任务,如果需要在客户端监控任务的状态,有如下方法:
1 2 3 4 5 6 7 8 9 | 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
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 | # 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