关于python: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.pyfrom celery import Celeryapp = Celery('tasks', broker='amqp://username:passwd@ip:port/varhost',backend='redis://username:passwd@ip:6390/db')@app.taskdef add(x, y):    return x + yif __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 pickleimport base64result = 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.pyfrom celery import Celeryimport celeryconfigapp = Celery(__name__, include=["task"])# 引入配置文件app.config_from_object(celeryconfig)if __name__ == '__main__':    result = add.delay(30, 42)# task.pyfrom main import app@app.taskdef add(x, y):    return x + y  # celeryconfig.pyBROKER_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 Celeryimport celeryconfigapp = 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_URLBROKER_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_contentCELERY_ENABLE_UTC    enable_utcCELERY_IMPORTS    importsCELERY_INCLUDE    includeCELERY_TIMEZONE    timezoneCELERYBEAT_MAX_LOOP_INTERVAL    beat_max_loop_intervalCELERYBEAT_SCHEDULE    beat_scheduleCELERYBEAT_SCHEDULER    beat_schedulerCELERYBEAT_SCHEDULE_FILENAME    beat_schedule_filenameCELERYBEAT_SYNC_EVERY    beat_sync_everyBROKER_URL    broker_urlBROKER_TRANSPORT    broker_transportBROKER_TRANSPORT_OPTIONS    broker_transport_optionsBROKER_CONNECTION_TIMEOUT    broker_connection_timeoutBROKER_CONNECTION_RETRY    broker_connection_retryBROKER_CONNECTION_MAX_RETRIES    broker_connection_max_retriesBROKER_FAILOVER_STRATEGY    broker_failover_strategyBROKER_HEARTBEAT    broker_heartbeatBROKER_LOGIN_METHOD    broker_login_methodBROKER_POOL_LIMIT    broker_pool_limitBROKER_USE_SSL    broker_use_sslCELERY_CACHE_BACKEND    cache_backendCELERY_CACHE_BACKEND_OPTIONS    cache_backend_optionsCASSANDRA_COLUMN_FAMILY    cassandra_tableCASSANDRA_ENTRY_TTL    cassandra_entry_ttlCASSANDRA_KEYSPACE    cassandra_keyspaceCASSANDRA_PORT    cassandra_portCASSANDRA_READ_CONSISTENCY    cassandra_read_consistencyCASSANDRA_SERVERS    cassandra_serversCASSANDRA_WRITE_CONSISTENCY    cassandra_write_consistencyCASSANDRA_OPTIONS    cassandra_optionsCELERY_COUCHBASE_BACKEND_SETTINGS    couchbase_backend_settingsCELERY_MONGODB_BACKEND_SETTINGS    mongodb_backend_settingsCELERY_EVENT_QUEUE_EXPIRES    event_queue_expiresCELERY_EVENT_QUEUE_TTL    event_queue_ttlCELERY_EVENT_QUEUE_PREFIX    event_queue_prefixCELERY_EVENT_SERIALIZER    event_serializerCELERY_REDIS_DB    redis_dbCELERY_REDIS_HOST    redis_hostCELERY_REDIS_MAX_CONNECTIONS    redis_max_connectionsCELERY_REDIS_PASSWORD    redis_passwordCELERY_REDIS_PORT    redis_portCELERY_RESULT_BACKEND    result_backendCELERY_MAX_CACHED_RESULTS    result_cache_maxCELERY_MESSAGE_COMPRESSION    result_compressionCELERY_RESULT_EXCHANGE    result_exchangeCELERY_RESULT_EXCHANGE_TYPE    result_exchange_typeCELERY_TASK_RESULT_EXPIRES    result_expiresCELERY_RESULT_PERSISTENT    result_persistentCELERY_RESULT_SERIALIZER    result_serializerCELERY_RESULT_DBURI    请result_backend改用。CELERY_RESULT_ENGINE_OPTIONS    database_engine_options[...]_DB_SHORT_LIVED_SESSIONS    database_short_lived_sessionsCELERY_RESULT_DB_TABLE_NAMES    database_db_namesCELERY_SECURITY_CERTIFICATE    security_certificateCELERY_SECURITY_CERT_STORE    security_cert_storeCELERY_SECURITY_KEY    security_keyCELERY_ACKS_LATE    task_acks_lateCELERY_TASK_ALWAYS_EAGER    task_always_eagerCELERY_TASK_ANNOTATIONS    task_annotationsCELERY_TASK_COMPRESSION    task_compressionCELERY_TASK_CREATE_MISSING_QUEUES    task_create_missing_queuesCELERY_TASK_DEFAULT_DELIVERY_MODE    task_default_delivery_modeCELERY_TASK_DEFAULT_EXCHANGE    task_default_exchangeCELERY_TASK_DEFAULT_EXCHANGE_TYPE    task_default_exchange_typeCELERY_TASK_DEFAULT_QUEUE    task_default_queueCELERY_TASK_DEFAULT_RATE_LIMIT    task_default_rate_limitCELERY_TASK_DEFAULT_ROUTING_KEY    task_default_routing_keyCELERY_TASK_EAGER_PROPAGATES    task_eager_propagatesCELERY_TASK_IGNORE_RESULT    task_ignore_resultCELERY_TASK_PUBLISH_RETRY    task_publish_retryCELERY_TASK_PUBLISH_RETRY_POLICY    task_publish_retry_policyCELERY_QUEUES    task_queuesCELERY_ROUTES    task_routesCELERY_TASK_SEND_SENT_EVENT    task_send_sent_eventCELERY_TASK_SERIALIZER    task_serializerCELERYD_TASK_SOFT_TIME_LIMIT    task_soft_time_limitCELERYD_TASK_TIME_LIMIT    task_time_limitCELERY_TRACK_STARTED    task_track_startedCELERYD_AGENT    worker_agentCELERYD_AUTOSCALER    worker_autoscalerCELERYD_CONCURRENCY    worker_concurrencyCELERYD_CONSUMER    worker_consumerCELERY_WORKER_DIRECT    worker_directCELERY_DISABLE_RATE_LIMITS    worker_disable_rate_limitsCELERY_ENABLE_REMOTE_CONTROL    worker_enable_remote_controlCELERYD_HIJACK_ROOT_LOGGER    worker_hijack_root_loggerCELERYD_LOG_COLOR    worker_log_colorCELERYD_LOG_FORMAT    worker_log_formatCELERYD_WORKER_LOST_WAIT    worker_lost_waitCELERYD_MAX_TASKS_PER_CHILD    worker_max_tasks_per_childCELERYD_POOL    worker_poolCELERYD_POOL_PUTLOCKS    worker_pool_putlocksCELERYD_POOL_RESTARTS    worker_pool_restartsCELERYD_PREFETCH_MULTIPLIER    worker_prefetch_multiplierCELERYD_REDIRECT_STDOUTS    worker_redirect_stdoutsCELERYD_REDIRECT_STDOUTS_LEVEL    worker_redirect_stdouts_levelCELERYD_SEND_EVENTS    worker_send_task_eventsCELERYD_STATE_DB    worker_state_dbCELERYD_TASK_LOG_FORMAT    worker_task_log_formatCELERYD_TIMER    worker_timerCELERYD_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