1.什么是任务队列?
任务队列是一个将工作分布到多线程或多台机器上的机制。
一个任务队列的输入是一个工作单元也被称为任务。专用的工作者进程将会持续监控任务队列并执行它们。
Celery通过消息进行通信，通常通过一个中间人在客户端和工作者之间协调。客户端初始化一个任务并将它加到任务队列，中间人将其发送到工作者。
一个Celery系统可以由多个工作者和中间人构成，这为高可用性和横向扩展提供了便利。
Celery是用python语言实现的，但是可以使用任何语言实现其协议。除了python以外，还有node.js实现的node-celery和一个php实现的客户端。
Node.js, and a&.
不同语言之间可以相互操作，通过使用webhooks..
2.你需要知道的
Celery需要一个消息通道来发送和接收消息。RabbitMQ和Redis中间人通道是目前支持的特性。但是Celery也支持很多其它正在实验中的解决方案，如使用SQLite做本地开发。
Celery可以运行在一台或多台机器上，甚至可以运行在不同的数据中心上。
如果你是第一次使用Celery或者刚刚从老版本转到Celery3.0上，你需要阅读我们的开始教程。
使用Celery的第一步
Celery包含一系列的任务队列。你无需学习Celery要解决地所有复杂问题就可以很容易地使用它。它是围绕最佳实践设计的，所以你可以扩展和集成其它语言。
它还提供了一些工具来帮助你在生产环境中更好地使用它。
通过本教程你将会学习到使用Celery的所有基本知识，包括以下内容:
a.选择并安装一个消息通道（中间人）
b.安装Celery并创建第一个任务
c.开始工作者并执行任务
d.跟踪任务不同阶段的状态，并监视它们的返回值。
Celery初看起来有些复杂，但是不用担心－－－本教程将会让你很快上手。我们将会尽量保持简单，以使你不必困惑于它的复杂特性。
在学完本教程后你最好浏览文档的剩余部分，比如“下一步”教程，这里将会展示Celery的更多能力。
选择一个中间人
Celery需要一个解决方案来发送和接收消息，通常这是通过一个被称为“中间人”的独立服务来实现的。
有以下几种选择：
RabiitMQ功能齐全，稳定，耐用而且易于安装。它是生产环境的完美选择。
更详细地关于RabiitMQ的介绍请关注后续教程：使用RabbitMQ&
如果你是在Ubuntu或者Debian上安装，执行以下命令：
$ sudo apt-get install rabbitmq-server
执行完上述命令后，中间人rabbitMQ已经在后台运行起来了。将会有下面的提示信息:
Starting&rabbitmq-server:&SUCCESS.
如果是在其它系统上使用RabbitMQ，也不用担心。下面的网页链接有类似的安装说明，包括如何在Windows上安装和使用:
redis功能也很齐全，但是数据丢失会对程序异常或者断电的情况更加敏感。形影相随详细的介绍请关注后续教程：使用Redis&
不推荐使用数据库作为消息队列。但是对于一些比较小的装置也足够了。你可以的选择包括：
使用SQLAlchemy&
使用Django Database
如果已经在使用一个Django数据库为例，在开发的过程中使用它作为你的消息中间人将会十分方便，同时你在生产环境中可以使用其它更健壮的系统。
其它中间人
除了上面介绍的以外，还有一些实验中的消息通道实现可以选择，包括Amazon SQS,MongoDB和IronMQ.
安装Celery
Celery是一个python package Index(PyPI),所以可以通过标准的python工具如pip或easy_install来安装：
$ pip install celery
首先你需要一个Celery实例，也可以简称为Celery应用程序或者&app&。
这个app会作为Celery中所有工作的入口，比如创建任务，管理工作者。它必须能够被其它模块导入。
下面的教程中将所有的代码放在一个模块中，但是对于更大的工程你可能需要创建更专业的模块划分。
下面我们创建一个tasks.py:
from celery import Celery
app = Celery('tasks', broker='amqp://guest@localhost//')
def add(x, y):
return x + y
Celery的第一个参数是本模块的名称，这对于自动生成名称是需要的；第二个参数是一个broker关键字参数，用于指定你想使用的中间人的URL，这里以默认配置使用RabiitMQ。关于消息中间人你可以有更多选择，比如对于RabiitMQ使用amqp://localhost，或者对于Redis使用redis://localhost
然后你定义了一个名为add的任务，并返回x,y之和。
下面通过使用worker参数执行我们的程序
$ celery -A tasks worker --loglevel=info
在生产环境中，你可能希望以守护进程的方式运行worker。要实现这种方式，你需要使用对应平台所提供的工具，或者一些像的工具。
更详细的介绍请关注后续教程（以守护进程方式运行worker ）。
要查看完整的可用的命令行选项，请执行：
celery worker --help
还有一些其它可用的命令，也可以使用help命令:
$ celery help
你可以通过使用delay()方法来调用我们的任务。
这是一个调用apply_async()方法的便捷方法。它能够给我们对任务执行有更多的控制。
&&& from tasks import add
&&& add.delay(4, 4)
这个任务会被先前运行的worker执行，你可以通过控制台的输出来确认。
调用任务会返回一个AsyncResult实例，可以通过它来查看任务的状态，等待任务执行结束或者获得它的返回值（如果任务执行失败，获取异常信息或者调用栈）。
但是这些不是默认开启的，你必须配置Celery来使用一个结果后台，下一节将会详细地介绍。
如果你想保存任务状态轨迹，Celery需要在某处保存或者发送这些状态。有一些内置的结果后台可以使用：SQLAlchemy/Django ORM,Memcached,Redis,AMQP(RabbitMQ),MongoDB,或者你可以自已实现。
下面的例子使用rpc结果后台，它发送状态作为临时消息。结果后台通过backend关键字参数指定（如果你使用了配置模块，也可以通过task_result_settings来配置）。
app = Celery('tasks', backend='rpc://', broker='amqp://')
如果你想使用Redis作为结果后台，但是仍使用RabbitMQ作为中间人(一种十分流行的组合)，你可以这样:
app = Celery('tasks', backend='redis://localhost', broker='amqp://')
更多关于result backend的教程请关注教程Result Backends。
现在我们配置了结果后台，我们再次运行我们的任务。这样在你调用了任务之后，将能够获得一个AsyncResult实例。
&&& result = add.delay(4, 4)
ready()方法将会返回这个实例，不管任务是正在运行还是已经运行结束。
&&& result.ready()
你也可以等待结果完成，但是很少这么做，因为这会把异步调用转换为同步调用。
&&& result.get(timeout=1)
如果任务抛出一个异常，get()方法将会再次将其抛出。但是你可以通过指定propagate参数来覆盖这种行为。
&&& result.get(propagate=False)
如果任务抛出一个异常，你也可以获得原始的调用栈：
&&& result.traceback
通过&查看完整地结果对象说明。
Celery,是一个消费者模型，本身不需要过多的操作。它有一个输入和一个输出，你需要将输入连接到一个中间人，如果需要还需要把输出连接到一个结果后台。
默认的配置对大部分用户已经足够好了，但是你也可以通过微调来使Celery按照你的要求工作。
通过阅读可用的选项来熟悉可以配置的选项是一个好办法。你可以通过&来了解这些选项。
可以通过configuration模块直接对app进行配置。
下面是一个配置任务序列化的例子：
app.conf.task_serializer = 'json'
你可以通过update方法来一次性地配置多个选项：
app.conf.update(
task_serializer='json',
accept_content=['json'],
# Ignore other content
result_serializer='json',
timezone='Europe/Oslo',
enable_utc=True,
对于大型工程来说，使用专业的配置模块十分有用。在实际使用中，硬编码任务周期间隔和任务路由选项将会十分麻烦，所以最好将这些集中配置，尤其是提供一个库让用户 控制它们期望的任务行为。你可以想象一下，系统管理可以在系统故障时简单地改变配置。
你可以通过app.config_from_object()方法告诉Celery实例使用一个配置模块。
app.config_from_object('celeryconfig')
这个配置模块通常起名为'celeryconfig,但是你可以使用任何模块名称。
一个名为'celeryconfig.py'模块必须能够被加载，位于当前目录或者PYTHONPATH。典型的celeryconfig.py像下面这样：
celeryconfig.py:
broker_url = 'amqp://'
result_backend = 'rpc://'
task_serializer = 'json'
result_serializer = 'json'
accept_content = ['json']
timezone = 'Europe/Oslo'
enable_utc = True
要确保配置模块没有任何语法错误，且能够正确的工作，你可以通过下面的命令尝试导入它：
$ python -m celeryconfig
要查看全部的配置选项说明，可以查看:&.
下面演示了配置文件的强大功能，你可以将一个任务路由到一个特定的队列。
celeryconfig.py:
task_routes = {
'tasks.add': 'low-priority',
除了配置任务路由你还可以限制任务的频率，10/m代表这个任务在一分钟内最多只能有10个被执行。
celeryconfig.py:
task_annotations = {
'tasks.add': {'rate_limit': '10/m'}
如果你正在使用RabiitMQ或者Redis作为中间人，你也可以在运行时为任务设置一个新的频率限制
$ celery -A tasks control rate_limit tasks.add 10/m
new rate limit set successfully
下面是一些常用的文档链接
任务路由：
远程控制命令：&,来监控你的工作者在干些什么。
如果你想了解更多，你可以继续学习“下一步&”教程。然后你可以开始学习用户指引.。
下面的问题章节列出了使用中的常见问题：.
a.如果你正在使用Debian,Ubuntu或者其它基于Debian的系统：
Debian最近将/dev/shm文件重命名为了/run/shm.
一种常见的解决方案是创建一个符号链接：
# ln -s /run/shm /dev/shm
如果你指定了--pidfile,--logfile,--statedb参数，你必须确保指定的目录对启动工作者的用户有读写权限。
后台结果不能工作或者任务总是处于挂起状态：
所有的任务默认是“挂起”状态，所以这个状态最好被称为“unkown”。当一个任务被发送后Celery不会更新任何状态，任何没有历史的任务被认为是pending状态（毕竟你知道task id）。
1.确保任务没有开启ignore_enabled
开启这个选项将会强制工作者忽略更新任务状态。
2.确保task_ignore_result配置是关闭的
3.确保没有任何老的工作者还在运行
很容易偶然地运行多个工作者，所以运行新的工作者之前确保先前的工作者已经正常的关闭了。
老的没有配置result backend的工作者可能劫持任务。
可以设置--pidfile为绝对路径来避免这种情况的发生。
4.确保客户端配置了正确的result backend.
如果由于某些原因，客户端配置使用了不同于工作者的result backend，你可能就获取不到结果，所以通过监视它来确保采用了正确的结果后台.
&&& result = task.delay()
&&& print(result.backend)
本文已收录于以下专栏：
相关文章推荐
celery不能以守护进程方式运行自己，需要使用下面的工具：
通用的初始化脚本 
脚本代码:extra/generic-init.d/ 
这个目录包含了celery worker程序通用的初始化脚本，...
使用RabbitMQ结合CeleryRabbitMQ 是默认的中间人，首先安装RabiitMQapt-get install rabbitmq
# 添加用户
rabbitmqctl add_user ...
我用的setting配置：#!/usr/bin/env python
import random
from kombu import serialization
from kombu import E...
最近同事项目想使用celery与rabbitmq来做任务调度，让我做一次这方面的使用分享。工作之余大致整理了一下。
一、先介绍一下rabbitmq消息队列
oAMQP，即AdvancedMess...
网上有很多《跟巴菲特学看上市公司财务报表》诸如此类的文章，仁者见仁智者见智。本文重点不在于，如何分析财务报表，而是如何获得财务报表，为后续的方便分析做准备！
Application
Celery库在使用前必须先实例化， 这个实例被称为application(或者简称为'app')。
主要是通过scrapy爬取二手房相关信息，只关心ershoufang相关链接，源码地址:
/happyAnger6/scrapy_tutorials
简介Celery是一个开源的分布式系统，支持任务队列实时处理，也支持定时任务。
Celery4.0是支持Python2.7的最后一个版本,后续版本需要使用Python3.5+.
Celery不支持...
/entry/65138
Celery是一个分布式任务队列工具，是一个异步的任务队列基于分布式消息传递。更多介绍可以参考 官网
这篇文章只是简单的...
python利用百度API进行地理编码（将地名转换为经纬度信息）本文章通过讲解如何在百度地图API申请密钥，然后在python中调用API接口将自有数据中的地名转换为经纬度坐标。
他的最新文章
讲师：姜飞俊
讲师：汪木铃
您举报文章：
举报原因：
原文地址：
原因补充：
(最多只允许输入30个字)This document describes the current stable version of Celery (4.1).
For development docs,
Using RabbitMQ
RabbitMQ is the default broker so it doesn’t require any additional
dependencies or initial configuration, other than the URL location of
the broker instance you want to use:
broker_url = 'amqp://myuser:mypassword@localhost:5672/myvhost'
For a description of broker URLs and a full list of the
various broker configuration options available to Celery,
see , and see below for setting up the
username, password and vhost.
over at RabbitMQ’s website. For macOS
If you’re getting nodedown errors after installing and using
rabbitmqctl then this blog post can help you identify
the source of the problem:
To use Celery we need to create a RabbitMQ user, a virtual host and
allow that user access to that virtual host:
$ sudo rabbitmqctl add_user myuser mypassword
$ sudo rabbitmqctl add_vhost myvhost
$ sudo rabbitmqctl set_user_tags myuser mytag
$ sudo rabbitmqctl set_permissions -p myvhost myuser &.*& &.*& &.*&
Substitute in appropriate values for myuser, mypassword and myvhost above.
See the RabbitMQ
for more information about .
The easiest way to install RabbitMQ on macOS is using
the new and
shiny package management system for macOS.
First, install Homebrew using the one-line command provided by the :
ruby -e &$(curl -fsSL /Homebrew/homebrew/go/install)&
Finally, we can install RabbitMQ using brew:
$ brew install rabbitmq
After you’ve installed RabbitMQ with brew you need to add the following to
your path to be able to start and stop the broker: add it to the start-up file for your
shell (e.g., .bash_profile or .profile).
PATH=$PATH:/usr/local/sbin
If you’re using a DHCP server that’s giving you a random host name, you need
to permanently configure the host name. This is because RabbitMQ uses the host name
to communicate with nodes.
Use the scutil command to permanently set your host name:
$ sudo scutil --set HostName myhost.local
Then add that host name to /etc/hosts so it’s possible to resolve it
back into an IP address:
localhost myhost myhost.local
If you start the rabbitmq-server, your rabbit node should now
be rabbit@myhost, as verified by rabbitmqctl:
$ sudo rabbitmqctl status
Status of node rabbit@myhost ...
[{running_applications,[{rabbit,&RabbitMQ&,&1.7.1&},
{mnesia,&MNESIA
CXC 138 12&,&4.4.12&},
{os_mon,&CPO
CXC 138 46&,&2.2.4&},
{sasl,&SASL
CXC 138 11&,&2.1.8&},
{stdlib,&ERTS
CXC 138 10&,&1.16.4&},
{kernel,&ERTS
CXC 138 10&,&2.13.4&}]},
{nodes,[rabbit@myhost]},
{running_nodes,[rabbit@myhost]}]
This is especially important if your DHCP server gives you a host name
starting with an IP address, (e.g., 23.10.cast.net).
case RabbitMQ will try to use rabbit@23: an illegal host name.
To start the server:
$ sudo rabbitmq-server
you can also run it in the background by adding the -detached option
(note: only one dash):
$ sudo rabbitmq-server -detached
Never use kill (kill(1)) to stop the RabbitMQ server,
but rather use the rabbitmqctl command:
$ sudo rabbitmqctl stop
When the server is running, you can continue reading .
Previous topic
Next topiccelery的官方文档其实相对还是写的很不错的.但是在一些深层次的使用上面却显得杂乱甚至就没有某些方面的介绍, 通过我的一个测试环境的settings.py来说明一些使用celery的技巧和解决办法
amqp交换类型
其实一共有4种交换类型，还有默认类型和自定义类型. 但是对我们配置队列只会用到其中之三，我来一个个说明，英语好的话可以直接去看英文文档
首先思考一下流程:
celerybeat生成任务消息，然后发送消息到一个exchange(交换机)
交换机决定那个(些)队列会接收这个消息，这个其实就是根据下面的exchange的类型和绑定到这个交换机所用的bindingkey
我们这里要说的其实就是怎么样决定第二步谁接收的问题
Direct Exchange
如其名，直接交换，也就是指定一个消息被那个队列接收， 这个消息被celerybeat定义个一个routing key，如果你发送给交换机并且那个队列绑定的bindingkey 那么就会直接转给这个队列
Topic Exchange
你设想一下这样的环境(我举例个小型的应该用场景): 你有三个队列和三个消息, A消息可能希望被X,Y处理,B消息你希望被,X,Z处理,C消息你希望被Y,Z处理.并且这个不是队列的不同而是消息希望被相关的队列都去执行,看一张图可能更好理解:
对，Topic可以根据同类的属性进程通配, 你只需要routing key有&.&分割:比如上图中的usa.news, usa.weather, europe.news, europe.weather
Fanout Exchange
先想一下广播的概念, 在设想你有某个任务，相当耗费时间，但是却要求很高的实时性,那么你可以需要多台服务器的多个workers一起工作，每个服务器负担其中的一部分,但是celerybeat只会生成一个任务,被某个worker取走就没了, 所以你需要让每个服务器的队列都要收到这个消息.这里很需要注意的是:你的fanout类型的消息在生成的时候为多份,每个队列一份，而不是一个消息发送给单一队列的次数
我的settings.py
这里只是相关于celery的部分:
import djcelery
djcelery.setup_loader()
INSTALLED_APPS = (
'django.contrib.auth',
'django.contrib.contenttypes',
'django.contrib.sessions',
'django.contrib.sites',
#'django.contrib.staticfiles',
'django.contrib.messages',
# Uncomment the next line to enable the admin:
'django.contrib.admin',
'django.contrib.staticfiles',
# Uncomment the next line to enable admin documentation:
# 'django.contrib.admindocs',
'dongwm.smhome',
'dongwm.apply',
'djcelery', # 这里增加了djcelery 也就是为了在django admin里面可一直接配置和查看celery
'django_extensions',
'djsupervisor',
'django.contrib.humanize',
'django_jenkins'
BROKER_URL = 'amqp://username:password@localhost:5672/yourvhost'
CELERY_IMPORTS = (
'dongwm.smhome.tasks',
'dongwm.smdata.tasks',
CELERY_RESULT_BACKEND = "amqp" # 官网优化的地方也推荐使用c的librabbitmq
CELERY_TASK_RESULT_EXPIRES = 1200 # celery任务执行结果的超时时间，我的任务都不需要返回结果,只需要正确执行就行
CELERYD_CONCURRENCY = 50 # celery worker的并发数 也是命令行-c指定的数目,事实上实践发现并不是worker也多越好,保证任务不堆积,加上一定新增任务的预留就可以
CELERYD_PREFETCH_MULTIPLIER = 4 # celery worker 每次去rabbitmq取任务的数量，我这里预取了4个慢慢执行,因为任务有长有短没有预取太多
CELERYD_MAX_TASKS_PER_CHILD = 40 # 每个worker执行了多少任务就会死掉，我建议数量可以大一些，比如200
CELERYBEAT_SCHEDULER = 'djcelery.schedulers.DatabaseScheduler' # 这是使用了django-celery默认的数据库调度模型,任务执行周期都被存在你指定的orm数据库中
CELERY_DEFAULT_QUEUE = "default_dongwm" # 默认的队列，如果一个消息不符合其他的队列就会放在默认队列里面
CELERY_QUEUES = {
"default_dongwm": { # 这是上面指定的默认队列
"exchange": "default_dongwm",
"exchange_type": "direct",
"routing_key": "default_dongwm"
"topicqueue": { # 这是一个topic队列 凡是topictest开头的routing key都会被放到这个队列
"routing_key": "topictest.#",
"exchange": "topic_exchange",
"exchange_type": "topic",
"test2": { # test和test2是2个fanout队列,注意他们的exchange相同
"exchange": "broadcast_tasks",
"exchange_type": "fanout",
"binding_key": "broadcast_tasks",
"exchange": "broadcast_tasks",
"exchange_type": "fanout",
"binding_key": "broadcast_tasks2",
class MyRouter(object):
def route_for_task(self, task, args=None, kwargs=None):
if task.startswith('topictest'):
'queue': 'topicqueue',
# 我的dongwm.tasks文件里面有2个任务都是test开头
elif task.startswith('dongwm.tasks.test'):
"exchange": "broadcast_tasks",
# 剩下的其实就会被放到默认队列
return None
# CELERY_ROUTES本来也可以用一个大的含有多个字典的字典,但是不如直接对它做一个名称统配
CELERY_ROUTES = (MyRouter(), )
Views(...) Comments()python（68）
Celery官方帮助 &
本文主要介绍Celery+RabbitMQ的入门知识
Celery 是一个异步任务队列/基于分布式消息传递作业队列，它侧重于实时操作，同样也支持调度
RabbitMQ为应用程序提供了强大的消息服务。它很容易使用，适合在云规模应用，并支持所有主流的操作系统和开发平台。RabbitMQ在Mozilla公共许可下开源
安装Celery
使用easy_install安装
sudo easy_install celery
安装RabbitMQ
在ubuntu下使用apt-get方式安装rabbitmq-server
sudo apt-get install rabbitmq-server
安装完毕之后可以使用如下命令查看MQ当前服务状态
alex@alex-pc:~/test$ sudo rabbitmqctl status
[sudo] password for alex:
Status of node 'rabbit@alex-pc' ...
[{running_applications,[{rabbit,&RabbitMQ&,&1.7.2&},
{mnesia,&MNESIA
CXC 138 12&,&4.4.12&},
{os_mon,&CPO
CXC 138 46&,&2.2.4&},
{sasl,&SASL
CXC 138 11&,&2.1.8&},
{stdlib,&ERTS
CXC 138 10&,&1.16.4&},
{kernel,&ERTS
CXC 138 10&,&2.13.4&}]},
{nodes,['rabbit@alex-pc']},
{running_nodes,['rabbit@alex-pc']}]
配置celery
选择一个测试目录，在当前路径下新建配置文件celeryconfig.py,如下：
import sys
sys.path.insert(0, os.getcwd())
CELERY_IMPORTS = (&tasks&, )
CELERY_RESULT_BACKEND = &amqp&
BROKER_HOST = &localhost&
BROKER_PORT = 5672
BROKER_USER = &guest&
BROKER_PASSWORD = &guest&
BROKER_VHOST = &/
新建tasks.py,如下:
from celery.task import task
def add(x, y):
return x + y
启动celery服务
在终端中使用如下命令：
alex@alex-pc:~/test$ celeryd --loglevel=INFO
[ 23:06:27,226: WARNING/MainProcess]
-------------- celery@alex-pc v2.2.7
---- **** -----
* -- [Configuration]
-- * - **** ---
amqplib://guest@localhost:5672/
- ** ----------
celery.loaders.default.Loader
- ** ----------
. logfile:
[stderr]@INFO
- ** ----------
. concurrency: 2
- ** ----------
- *** --- * ---
-- ******* ----
--- ***** ----- [Queues]
--------------
exchange:celery (direct) binding:celery
. tasks.add
[ 23:06:27,233: INFO/PoolWorker-1] child process calling self.run()
[ 23:06:27,235: INFO/PoolWorker-2] child process calling self.run()
[ 23:06:27,236: WARNING/MainProcess] celery@alex-pc has started.
打开ipython或者任何python shell即可
In [8]: from celery.task import task
In [9]: import tasks
In [10]: res = tasks.add.delay(2,2)
In [11]: res.ready()
Out[11]: True
In [12]: res.result
Out[12]: 4
从celeryd的loginfo的输出信息中可以看到调用成功:
[ 23:06:39,081: INFO/MainProcess] Got task from broker: tasks.add[82d8c609-2cac-47e8-be20-eb6a2dc502d6]
[ 23:06:39,120: INFO/MainProcess] Task tasks.add[82d8c609-2cac-47e8-be20-eb6a2dc502d6] succeeded in 0.5s: 4
&&相关文章推荐
* 以上用户言论只代表其个人观点，不代表CSDN网站的观点或立场
访问：1318474次
积分：16886
积分：16886
排名：第612名
原创：372篇
转载：805篇
评论：97条
(2)(1)(4)(6)(9)(6)(2)(3)(14)(6)(13)(12)(28)(85)(7)(17)(6)(44)(36)(15)(7)(82)(51)(7)(7)(40)(30)(44)(97)(122)(96)(56)(10)(84)(59)(2)(10)(43)(6)(8)
(window.slotbydup = window.slotbydup || []).push({
id: '4740881',
container: s,
size: '200,200',
display: 'inlay-fix'}