python之celery使用详解一
本文内容纲要：
-前言
-celery+rabbitmq初步
-简单介绍
-安装rabbitmq和redis
-初步使用
-第一步：创建并发送一个异步任务
-第二步:开启worker执行任务
-分析序列化的消息
-将序列化消息反序列化
-celery配置
-基本配置项
-加载配置
-一份比较常用的配置文件
-总结
-参考
前言
前段时间需要使用rabbitmq做写缓存，一直使用pika+rabbitmq的组合，pika这个模块虽然可以很直观地操作rabbitmq，但是官方给的例子太简单，对其底层原理了解又不是很深，遇到很多坑，尤其是需要自己写连接池管理和channel池管理。虽然也有用过celery，一直也是celery+redis的组合，涉及很浅；目前打算深研一下celery+redis+rabbitmq的使用。
celery+rabbitmq初步
我们先不在集成框架如flask或Django中使用celery，而仅仅单独使用。
简单介绍
Celery是一个异步任务队列，一个Celery有三个核心组件：
- Celery客户端:用于发布后台作业；当与Flask一起工作的时候，客户端与Flask应用一起运行。
- Celeryworkers:运行后台作业的进程。Celery支持本地和远程的workers，可以在本地服务器上启动一个单独的worker，也可以在远程服务器上启动worker，需要拷贝代码；
- 消息代理:客户端通过消息队列和workers进行通信，Celery支持多种方式来实现这些队列。最常用的代理就是RabbitMQ和Redis。
安装rabbitmq和redis
-
rabbitmq安装和配置参考：rabbitmq安装和配置
-
redis的安装和配置参考：redis的安装和配置
-
redis-py安装：
sudopipinstallredis
-
redis-py操作redis参考：python操作redis
为了提高性能，官方推荐使用librabbitmq，这是一个连接rabbitmq的C++的库；
#选择broker客户端、序列化和并发
sudopipinstallcelery[librabbitmq,redis,msgpack,gevent]
初步使用
一般我们使用redis做结果存储，使用rabbitmq做任务队列；
第一步：创建并发送一个异步任务
#初始化
#tasks.py
fromceleryimportCelery
app=Celery('tasks',broker='amqp://username:passwd@ip:port/varhost',backend='redis://username:passwd@ip:6390/db')
@app.task
defadd(x,y):
returnx+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-Aapp.celery_tasks.celeryworker-Qqueue--loglevel=info
#-A参数指定创建的celery对象的位置，该app.celery_tasks.celery指的是app包下面的celery_tasks.py模块的celery实例，注意一定是初始化后的实例，后面加worker表示该实例就是任务执行者；
#-Q参数指的是该worker接收指定的队列的任务，这是为了当多个队列有不同的任务时可以独立；如果不设会接收所有的队列的任务；
#-l参数指定worker输出的日志级别；
任务执行完毕后结果存储在redis中，查看redis中的数据，发现存在一个string类型的键值对：
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"}}
将序列化消息反序列化
importpickle
importbase64
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
fromceleryimportCelery
importceleryconfig
app=Celery(__name__,include=["task"])
#引入配置文件
app.config_from_object(celeryconfig)
if__name__=='__main__':
result=add.delay(30,42)
#task.py
frommainimportapp
@app.task
defadd(x,y):
returnx+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"]#指定任务接受的内容序列化的类型.
也可以直接加载配置
fromceleryimportCelery
importceleryconfig
app=Celery(__name__,include=["task"])
app.conf.update(
task_serializer='json',
accept_content=['json'],
result_serializer='json',
timezone='Europe/Oslo',
enable_utc=True,
)
此外还有两个方法可以加载配置，但开发不会直接调用：
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将被杀死，任务移交给父进程
#celeryworker的并发数，默认是服务器的内核数目,也是命令行-c参数指定的数目
CELERYD_CONCURRENCY=4
#celeryworker每次去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开头的routingkey都会被放到这个队列
"routing_key":"topic.#",
"exchange":"topic_exchange",
"exchange_type":"topic",
},
"task_eeg":{#设置扇形交换机
"exchange":"tasks",
"exchange_type":"fanout",
"binding_key":"tasks",
},
}
在celery4.0以后配置参数改成了小写，对于4.0以后的版本替代参数：
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使用详解二
参考
- http://docs.celeryproject.org/en/latest/userguide/tasks.html#task-options
- http://docs.jinkan.org/docs/celery/getting-started/first-steps-with-celery.html
- http://www.pythondoc.com/flask-celery/first.html
- https://blog.csdn.net/kk123a/article/details/74549117
- https://blog.csdn.net/preyta/article/details/54288870
本文内容总结：前言，celery+rabbitmq初步，简单介绍，安装rabbitmq和redis，初步使用，第一步：创建并发送一个异步任务，第二步:开启worker执行任务，分析序列化的消息，将序列化消息反序列化，celery配置，基本配置项，加载配置，一份比较常用的配置文件，总结，参考，
原文链接：https://www.cnblogs.com/cwp-bg/p/8759638.html