python之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安裝：
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