如何在 Ubuntu VPS 上使用 Celery 与 RabbitMQ 来做队列

简介

---

异步或非阻塞处理是一种将某些任务的执行与程序的主要流程分离的方法。这为您提供了几个优势，包括允许用户界面代码在没有中断的情况下运行。

消息传递是程序组件用来通信和交换信息的一种方法。它可以同步或异步实现，并且可以允许离散进程进行无问题的通信。消息传递通常作为传统数据库的替代实现，因为消息队列通常实现了额外的功能，提供了增加的性能，并且可以完全驻留在内存中。

Celery 是建立在异步消息传递系统上的任务队列。它可以用作编程任务可以被倾倒的桶。传递任务的程序可以继续执行和响应功能，然后稍后它可以轮询 celery 来查看计算是否完成并检索数据。

虽然 celery 是用 Python 编写的，但它的协议可以在任何语言中实现。它甚至可以通过 webhooks 与其他语言一起使用。

通过在程序环境中实现作业队列，您可以轻松卸载任务并继续处理用户的交互。这是增加应用程序响应性的简单方法，并且在执行长时间运行的计算时不会被锁定。

在本指南中，我们将在 Ubuntu 12.04 VPS 上安装和实现使用 RabbitMQ 作为消息系统的 celery 作业队列。

安装组件

---

安装 Celery

---

Celery 是用 Python 编写的，因此可以像处理常规 Python 包一样轻松安装。

我们将按照处理 Python 包的推荐程序，通过创建虚拟环境来安装我们的消息系统。这有助于我们保持环境稳定，不会影响更大的系统。

从 Ubuntu 的默认存储库安装 Python 虚拟环境包：

sudo apt-get update
sudo apt-get install python-virtualenv
1
2

我们将创建一个消息目录，在这里我们将实现我们的系统：

mkdir ~/messaging
cd ~/messaging
1
2

现在我们可以创建一个虚拟环境，通过以下命令安装 celery：

virtualenv --no-site-packages venv
1

配置好虚拟环境后，可以通过输入以下命令激活它：

source venv/bin/activate
1

您的提示符将更改以反映您现在正在使用我们上面创建的虚拟环境。这将确保我们的 Python 包安装在本地而不是全局。

如果在任何时候我们需要停用环境（现在不需要），可以输入：

deactivate
1

现在我们已经激活了环境，可以使用 pip 安装 celery：

pip install celery
1

安装 RabbitMQ

---

Celery 需要一个消息代理来处理来自外部源的请求。这个代理被称为“broker”。

有很多可供选择的代理选项，包括关系型数据库、NoSQL 数据库、键值存储和实际消息系统。

我们将配置 celery 使用 RabbitMQ 消息系统，因为它提供了强大、稳定的性能，并且与 celery 交互良好。这是一个很好的解决方案，因为它包含了与我们预期使用的功能很好契合的特性。

我们可以通过 Ubuntu 的存储库安装 RabbitMQ：

sudo apt-get install rabbitmq-server
1

RabbitMQ 服务在安装后会自动启动在我们的服务器上。

创建 Celery 实例

---

为了使用 celery 的任务排队功能，安装后的第一步是创建一个 celery 实例。这是一个简单的过程，导入包，创建一个“app”，然后设置 celery 能够在后台执行的任务。

让我们在我们的消息目录内创建一个名为 tasks.py 的 Python 脚本，我们可以在其中定义我们的工作人员可以执行的任务。

sudo nano ~/messaging/tasks.py
1

我们应该做的第一件事是从 celery 包中导入 Celery 函数：

from celery import Celery
1

之后，我们可以创建一个连接到默认 RabbitMQ 服务的 celery 应用程序实例：

from celery import Celery

app = Celery('tasks', backend='amqp', broker='amqp://')
1
2
3

Celery 函数的第一个参数是将用于标识任务的前缀名称。

backend 参数是一个可选参数，如果您希望查询后台任务的状态或检索其结果，则是必需的。

如果您的任务只是执行一些工作然后退出，而不返回在程序中使用的有用值，您可以将此参数省略。如果只有一些任务需要此功能，请在此启用它，我们可以在后面逐个案例地禁用它。

broker 参数指定连接到我们代理所需的 URL。在我们的情况下，这是运行在我们服务器上的 RabbitMQ 服务。RabbitMQ 使用一种称为“amqp”的协议运行。如果 RabbitMQ 在其默认配置下运行，celery 可以连接而无需其他信息，只需 amqp:// 方案。

构建 Celery 任务

---

在这个文件中，我们现在需要添加我们的任务。

每个 Celery 任务都必须使用装饰器 @app.task 来引入。这允许 Celery 识别可以添加其排队功能的函数。在每个装饰器之后，我们只需创建一个我们的工作进程可以运行的函数。

我们的第一个任务将是一个简单的函数，它将一个字符串打印到控制台。

from celery import Celery

app = Celery('tasks', backend='amqp', broker='amqp://')

@app.task
def print_hello():
    print 'hello there'
1
2
3
4
5
6
7

因为这个函数不返回任何有用的信息（它将其打印到控制台），我们可以告诉 Celery 不使用后端来存储关于此任务的状态信息。这在内部更简单，需要更少的资源。

from celery import Celery

app = Celery('tasks', backend='amqp', broker='amqp://')

@app.task(ignore_result=True)
def print_hello():
    print 'hello there'
1
2
3
4
5
6
7

接下来，我们将添加另一个函数，它将生成素数（取自 RosettaCode）。这可能是一个长时间运行的过程，因此这是一个很好的例子，说明我们在等待结果时如何处理异步工作进程。

from celery import Celery

app = Celery('tasks', backend='amqp', broker='amqp://')

@app.task(ignore_result=True)
def print_hello():
    print 'hello there'

@app.task
def gen_prime(x):
    multiples = []
    results = []
    for i in xrange(2, x+1):
        if i not in multiples:
            results.append(i)
            for j in xrange(i*i, x+1, i):
                multiples.append(j)
    return results
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

因为我们关心这个函数的返回值，并且我们想知道它何时完成（以便我们可以使用结果等），所以我们不会向这个第二个任务添加 ignore_result 参数。

保存并关闭文件。

启动 Celery 工作进程

---

我们现在可以启动一个工作进程，它将能够接受来自应用程序的连接。它将使用我们刚刚创建的文件来了解它可以执行的任务。

启动一个工作实例就像使用 celery 命令调用应用程序名称一样简单。我们将在字符串末尾包含一个 “&” 字符，将我们的工作进程放入后台：

celery worker -A tasks &
1

这将启动一个应用程序，然后将其从终端分离，允许您继续使用它进行其他任务。

如果您想要启动多个工作进程，可以使用 -n 参数为每个工作进程命名：

celery worker -A tasks -n one.%h &
celery worker -A tasks -n two.%h &
1
2

当工作进程被命名时，%h 将被主机名替换。

要停止工作进程，您可以使用 kill 命令。我们可以查询进程 ID，然后根据这些信息消除工作进程。

ps auxww | grep 'celery worker' | awk '{print $2}' | xargs kill
1

这将允许工作进程在退出之前完成其当前任务。

如果您希望关闭所有工作进程而不等待它们完成任务，可以执行：

ps auxww | grep 'celery worker' | awk '{print $2}' | xargs kill -9
1

使用队列处理工作

---

我们可以使用我们生成的工作进程在后台为我们的程序完成工作。

我们将在 Python 解释器中探索不同的选项，而不是创建一个完整的程序来演示它是如何工作的：

python
1

在提示符下，我们可以将我们的函数导入到环境中：

from tasks import print_hello
from tasks import gen_prime
1
2

如果您测试这些函数，它们似乎没有任何特殊功能。第一个函数按预期打印一行：

print_hello()
1

---

hello there
1

第二个函数返回一个素数列表：

primes = gen_prime(1000)
print primes
1
2

如果我们给第二个函数一个更大的数字范围来检查，执行将挂起，因为它在计算中：

primes = gen_prime(50000)
1

通过输入 “CTRL-C” 来停止执行。这个过程显然没有在后台计算。

要访问后台工作进程，我们需要使用 .delay 方法。Celery 为我们的函数添加了额外的功能。这个方法用于将函数传递给工作进程执行。它应该立即返回：

primes = gen_prime.delay(50000)
1

这个任务现在正在由我们之前启动的工作进程执行。因为我们为应用程序配置了 backend 参数，我们可以检查计算的状态并访问结果。

要检查任务是否完成，我们可以使用 .ready 方法：

primes.ready()
1

---

False
1

“False” 的值意味着任务仍在运行，结果尚不可用。当我们得到 “True” 的值时，我们可以对答案做些什么。

primes.ready()
1

---

True
1

我们可以使用 .get 方法获取值。

如果我们已经使用 .ready 方法验证了值是否计算出来，那么我们可以像这样使用该方法：

print primes.get()
1

---

[2, 3, 5, 7, 11, 13, 17, 19, 23, 29, 31, 37, 41, 43, 47, 53, 59, 61, 67, 71, 73, 79, 83, 89, 97, 101, 103, 107, 109, 113, 127, 131, 137, 139, 149, 151, 157, 163, 167, 173, 179, 181, 191, 193, 197, 199, 211, 223, 227, 229, 233, 239, 241, 251, 257, 263, 269, 271, 277, 281, 283, 293, 307, 311, 313, 317, 331, 337, 347, 349, 353, 359, 367, 373, 379, 383, 389, 397, 401, 409, 419, 421, 431, 433, 439, 443, 449, 457, 461, 463, 467, 479, 487, 491, 499, 503, 509, 521, 523,
. . .
1
2

然而，如果您在调用 .get 之前没有使用 .ready 方法，您很可能希望添加一个 “timeout” 选项，以便您的程序不必等待结果，这将违反我们的实现目的：

print primes.get(timeout=2)
1

如果超时，这将引发异常，您可以在程序中处理它。

结论

---

虽然这些信息足以让你开始在程序中使用 celery，但这只是揭开了这个库的全部功能的一角。Celery允许你将后台任务串联在一起，对任务进行分组，并以有趣的方式组合函数。

虽然 celery 是用 Python 编写的，但可以通过 Webhooks 与其他语言一起使用。这使得它非常灵活，可以将任务移到后台，而不受所选择的语言的限制。