前段时间需要使用rabbitmq做写缓存，一直使用pika+rabbitmq的组合，pika这个模块虽然可以很直观地操作rabbitmq，但是官方给的例子太简单，对其底层原理了解又不是很深，遇到很多坑，尤其是需要自己写连接池管理和channel池管理。虽然也有用过celery，一直也是celery+redis的组合，涉及很浅；目前打算深研一下celery+redis+rabbitmq的使用。
我们先不在集成框架如flask或Django中使用celery，而仅仅单独使用。
Celery是一个异步任务队列，一个Celery有三个核心组件：
Celery客户端:用于发布后台作业；当与Flask一起工作的时候，客户端与Flask应用一起运行。 Celeryworkers:运行后台作业的进程。Celery支持本地和远程的workers，可以在本地服务器上启动一个单独的worker，也可以在远程服务器上启动worker，需要拷贝代码； 消息代理:客户端通过消息队列和workers进行通信，Celery支持多种方式来实现这些队列。最常用的代理就是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做任务队列；
发生了什么事
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类型，可以看看添加进去的任务数据。
在项目目录下执行命令：
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"}}我们可以看到body里面有我们需要执行的函数的一切信息，celery的worker接收到消息后就会反序列化body数据，执行相应的方法。
常见的数据序列化方式
binary:二进制序列化方式；python的pickle默认的序列化方法； json:json支持多种语言,可用于跨语言方案，但好像不支持自定义的类对象； XML:类似标签语言； msgpack:二进制的类json序列化方案,但比json的数据结构更小,更快； yaml:yaml表达能力更强,支持的数据类型较json多,但是python客户端的性能不如json
经过比较，为了保持跨语言的兼容性和速度，采用msgpack或json方式；
celery的性能和许多因素有关，比如序列化的方式，连接rabbitmq的方式，多进程、单线程等等，我们可以指定配置；
也可以直接加载配置
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.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使用详解二
本文内容总结：前言，celery+rabbitmq初步，简单介绍，安装rabbitmq和redis，初步使用，第一步：创建并发送一个异步任务，第二步:开启worker执行任务，分析序列化的消息，将序列化消息反序列化，celery配置，基本配置项，加载配置，一份比较常用的配置文件，总结，参考，
原文链接：https://www.cnblogs.com/cwp-bg/p/8759638.html相关推荐
数学中有许多类型的级数，可以在C编程中轻松解决。该程序是在C程序中找到系列后继的和。Tn = n2 - (n-1 2...
要从全部四列中找到最低分，请使用MySQLLEAST( 函数。让我们首先创建一个表-create table DemoTable( Score1 int,...
