共计 10816 个字符，预计需要花费 28 分钟才能阅读完成。
前言
前段时间须要应用 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 装置：
sudo pip install redis - redis-py 操作 redis 参考：python 操作 redis
为了进步性能，官网举荐应用 librabbitmq，这是一个连贯 rabbitmq 的 C ++ 的库；
# 抉择 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 类型的键值对：
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,
) 此外还有两个办法能够加载配置，但开发不会间接调用：
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 当前的版本代替参数：
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参考
- 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
本文为转载文章，贵在分享，版权归原作者及原出处所有，如波及版权等问题，请及时与我分割。原文出处：天宇之游
原文链接：https://www.cnblogs.com/cwp-bg/p/8759638.html
正文完