不正经

这个屌丝很懒，什么也没留下！

热门标签

Celery异步任务

作者：不正经 | 2024-04-13 04:06:45

踩

celery异步任务

1. 简介

Celery 是一个异步任务队列。可以使用它在应用上下文之外执行任务。总的想法就是应用程序可能需要执行任何消耗资源的任务都可以交给任务队列，让应用程序自由和快速地响应客户端请求。

使用 Celery 运行后台任务并不像在线程中这样做那么简单。但是好处多多，Celery 具有分布式架构，使你的应用易于扩展。

1.1 定义

Celery是一个基于Python开发的分布式异步消息任务队列，它非常易于使用。通过它可以轻松的实现任务的异步处理，如果你的业务场景中需要用到异步任务，就可以考虑使用Celery。

1.2 组成

一个 Celery 安装有三个核心组件：

Celery 客户端: 用于发布后台作业。当与 Flask 一起工作的时候，客户端与 Flask 应用一起运行。
Celery workers: 这些是运行后台作业的进程。Celery 支持本地和远程的 workers，因此你就可以在 Flask 服务器上启动一个单独的 worker，随后随着你的应用需求的增加而新增更多的 workers。
Brokers（消息代理）: 客户端通过消息队列和 workers 进行通信，Celery 支持多种方式来实现这些队列。

支持的消息中间件总览：

消息中间件	支持适配状态	支持监控	支持远程控制
RabbitMQ	稳定	是	是
Redis	稳定	是	是
Amazon SQS	稳定	否	否
Zookeeper	实验	否	否

RabbitMQ是默认的Broker，它不需要其他额外的依赖和初始化配置。

参考文章
- RabbitMQ的Python客户端pika使用调研： RabbitMQ的Python客户端pika使用调研
- 官方教程Using RabbitMQ：http://docs.celeryproject.org/en/latest/getting-started/brokers/rabbitmq.html
最常用的代理
- RabbitMQ: http://www.rabbitmq.com/
- Redis: http://redis.io/。

可以直接到 Github repository 获取本文用到的代码。 README 文件将会给你快速和直接的方式去运行示例应用。
Celery支持多种消息中间件作为Broker，即中间人。来在应用程序和Worker之间传递消息。

1.3 优点

1. Simple(简单)
Celery 使用和维护都非常简单，并且不需要配置文件，包括一个mailing-list 和一个IRC channel.

这有一个最简单的应用示例，可以参照：

from celery import Celery

app = Celery('hello', broker='amqp://guest@localhost//')

@app.task
def hello():
  return 'hello world'
1
2
3
4
5
6
7

2. Highly Available（高可用）
woker和client会在网络连接丢失或者失败时，自动进行重试。并且有的brokers 也支持“双主”或者“主／从”的方式实现高可用。
3. Fast（快速）
单个的Celery进程每分钟可以处理百万级的任务，并且只需要毫秒级的往返延迟（使用 RabbitMQ, librabbitmq, 和优化设置时）
4. Flexible（灵活）
Celery几乎每个部分都可以扩展使用，自定义池实现、序列化、压缩方案、日志记录、调度器、消费者、生产者、broker传输等等。

1.4 应用场景

异步：
有的任务执行时间较长，你不想让程序一直等待结果返回，可以先将其任务放入celery任务队列中，并从Celery获取一个任务ID。后续通过任务ID询问Celery来得知任务执行状态和进度。
定时：
需要定时执行同样的任务，Celery任务队列支持定时触发，可以按照时间间隔或者crontab表达式来触发任务。

1.5 文档

管理获取任务状态：
- 官方文档:http://docs.celeryproject.org/en/latest/userguide/tasks.html#states
backend
- 官方文档:http://docs.celeryproject.org/en/latest/userguide/tasks.html#task-result-backends
Celery 文档： http://docs.celeryproject.org/en/latest/index.html

2. 基本工作流程

基本工作流程

3. 安装及环境搭建

3.1 celery安装

通过Python包管理平台（PyPI:https://pypi.org/Flasksearch/?c=Framework+%3A%3A+）
源码安装,使用pip安装:
```
$ pip install -U Celery
1
```

3.2 celery环境搭建

使用pip命令安装与Celery捆绑、依赖的组件：

$ pip install "celery[librabbitmq,redis,auth,msgpack]"
1

4. Celery使用

4.1 与Flask框架集成

Flask 与 Celery 整合是十分简单，不需要任何插件。一个 Flask 应用需要使用 Celery 的话只需要初始化 Celery 客户端
Celery 通过创建一个 Celery 类对象来初始化，传入应用名称以及消息代理的连接 URL

from flask import Flask
from celery import Celery

app = Flask(__name__)
# 配置代理服务器url
app.config['CELERY_BROKER_URL'] = 'redis://localhost:6379/0'
# celery存储状态和运行结果
app.config['CELERY_RESULT_BACKEND'] = 'redis://localhost:6379/0'

celery = Celery(app.name, broker=app.config['CELERY_BROKER_URL'])
# 其他配置
celery.conf.update(app.config)
1
2
3
4
5
6
7
8
9
10
11
12

任何需要作为后台任务的函数需要用 celery.task 装饰器装饰。例如:

@celery.task
def my_background_task(arg1, arg2):
    # some long running task here
    return result
1
2
3
4

接着 Flask 应用能够请求这个后台任务的执行，像这样:

task = my_background_task.delay(10, 20)
1

delay() 方法是强大的 apply_async() 调用的快捷方式。这样相当于使用 apply_async():

task = my_background_task.apply_async(args=[10, 20])
1

当使用 apply_async()，可以通过countdown参数执行定时操作

task = my_background_task.apply_async(args=[10, 20], countdown=60)
1

delay() 和 apply_async() 的返回值是一个表示任务的对象，这个对象可以用于获取任务状态。

4.2 简单例子：异步发送邮件

需求：能够发送邮件但是不阻塞主应用。

安装Flask-Mail扩展： https://pythonhosted.org/Flask-Mail/

示例应用是一个只有一个输入文本框的简单表单。
要求用户在此文本框中输入一个电子邮件地址，并在提交，服务器会发送一个测试电子邮件到这个邮件地址。
表单中包含两个提交按钮，一个立即发送邮件，一个是一分钟后发送邮件。

<html>
  <head>
    <title>Flask + Celery Examples</title>
  </head>
  <body>
    <h1>Flask + Celery Examples</h1>
    <h2>Example 1: Send Asynchronous Email</h2>
    {% for message in get_flashed_messages() %}
    <p style="color: red;">{{ message }}</p>
    {% endfor %}
    <form method="POST">
      <p>Send test email to: <input type="text" name="email" value="{{ email }}"></p>
      <input type="submit" name="submit" value="Send">
      <input type="submit" name="submit" value="Send in 1 minute">
    </form>
  </body>
</html>
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

只是一个普通的 HTML 表单，再加上 Flask 闪现消息。
Flask-Mail 扩展需要一些配置，尤其是电子邮件服务器发送邮件的时候会用到一些细节。为了简单使用 Gmail 账号作为邮件服务器:

# Flask-Mail configuration
app.config['MAIL_SERVER'] = 'smtp.googlemail.com'
app.config['MAIL_PORT'] = 587
app.config['MAIL_USE_TLS'] = True
app.config['MAIL_USERNAME'] = os.environ.get('MAIL_USERNAME')
app.config['MAIL_PASSWORD'] = os.environ.get('MAIL_PASSWORD')
app.config['MAIL_DEFAULT_SENDER'] = 'flask@example.com'
1
2
3
4
5
6
7

将Gmial账号设置在系统的环境变量，这是从应用中导入的。

有一个单一的路由来支持这个示例:

@app.route('/', methods=['GET', 'POST'])
def index():
    if request.method == 'GET':
        return render_template('index.html', email=session.get('email', ''))
    email = request.form['email']
    session['email'] = email

    # send the email
    msg = Message('Hello from Flask',
                  recipients=[request.form['email']])
    msg.body = 'This is a test email sent from a background Celery task.'
    if request.form['submit'] == 'Send':
        # send right away
        send_async_email.delay(msg)
        flash('Sending email to {0}'.format(email))
    else:
        # send in one minute
        send_async_email.apply_async(args=[msg], countdown=60)
        flash('An email will be sent to {0} in one minute'.format(email))

    return redirect(url_for('index'))
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

这是一个很标准的 Flask 应用。
由于这是一个非常简单的表单，我决定在没有扩展的帮助下处理它，因此我用 request.method 和 request.form 来完成所有的管理。
我保存用户在文本框中输入的值在 session 中，这样在页面重新加载后就能记住它。
在这个函数中让人有兴趣的是发送邮件的时候是通过调用一个叫做 send_async_email 的 Celery 任务，该任务调用 delay() 或者 apply_async() 方法。
这个应用的最后一部分就是能够完成作业的异步任务:

@celery.task
def send_async_email(msg):
    """Background task to send an email with Flask-Mail."""
    with app.app_context():
        mail.send(msg)
1
2
3
4
5

这个任务使用 celery.task 装饰使得成为一个后台作业。这个函数唯一值得注意的就是 Flask-Mail 需要在应用的上下文中运行，因此需要在调用 send() 之前创建一个应用上下文。

重点注意在这个示例中从异步调用返回值并不保留，因此应用不能知道调用成功或者失败。当你运行这个示例的时候，需要检查 Celery worker 的输出来排查发送邮件的问题。

4.3 复杂例子

显示状态更新和结果

后台作业启动应用,许多应用程序有必要监控它的后台任务并且获取运行结果。

这个示例展示一个虚构的长时间运行的任务。用户点击按钮启动一个或者更多的长时间运行的任务，在浏览器上的页面使用 ajax 轮询服务器更新所有任务的状态。每一个任务，页面都会显示一个图形的状态栏，进度条，一个状态消息，并且当任务完成的时候，也会显示任务的执行结果。示例的截图在本文的最开始。

状态更新的后台任务

让我向你们展示我在第二个示例中使用的后台任务:

@celery.task(bind=True)
def long_task(self):
    """Background task that runs a long function with progress reports."""
    verb = ['Starting up', 'Booting', 'Repairing', 'Loading', 'Checking']
    adjective = ['master', 'radiant', 'silent', 'harmonic', 'fast']
    noun = ['solar array', 'particle reshaper', 'cosmic ray', 'orbiter', 'bit']
    message = ''
    total = random.randint(10, 50)
    for i in range(total):
        if not message or random.random() < 0.25:
            message = '{0} {1} {2}...'.format(random.choice(verb),
                                              random.choice(adjective),
                                              random.choice(noun))
        self.update_state(state='PROGRESS',
                          meta={'current': i, 'total': total,
                                'status': message})
        time.sleep(1)
    return {'current': 100, 'total': 100, 'status': 'Task completed!',
            'result': 42}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

在 Celery 装饰器中添加了 bind=True 参数:告诉 Celery 发送一个 self 参数到我的函数，我能够使用它(self)来记录状态更新。

因为这个任务真没有干什么有用的事情，我决定使用随机的动词，形容词和名词组合的幽默状态信息。你可以在代码上看到我用来生成上述信息的毫无意义的列表。

self.update_state() 调用是 Celery 如何接受这些任务更新。有一些内置的状态，比如 STARTED, SUCCESS 等等，但是 Celery 也支持自定义状态。这里我使用一个叫做 PROGRESS 的自定义状态。连同状态，还有一个附件的元数据，该元数据是 Python 字典形式，包含目前和总的迭代数以及随机生成的状态消息。客户端可以使用这些元素来显示一个漂亮的进度条。每迭代一次休眠一秒，以模拟正在做一些工作。

当循环退出，一个 Python 字典作为函数结果返回。这个字典包含了更新迭代计数器，最后的状态消息和幽默的结果。

上面的 long_task() 函数在一个 Celery worker 进程中运行。下面你能看到启动这个后台作业的 Flask 应用路由:

@app.route('/longtask', methods=['POST'])
def longtask():
    task = long_task.apply_async()
    return jsonify({}), 202, {'Location': url_for('taskstatus',
                                                  task_id=task.id)}
1
2
3
4
5

正如你所见，客户端需要发起一个 POST 请求到 /longtask 来掀开这些任务中的一个的序幕。服务器启动任务，并且存储返回值。对于响应我使用状态码 202，这个状态码通常是在 REST APIs 中使用用来表明一个请求正在进行中。我也添加了 Location 头，值为一个客户端用来获取状态信息的 URL。这个 URL 指向另一个叫做 taskstatus 的 Flask 路由，并且有 task.id 作为动态的要素。

从 Flask 应用中访问任务状态

上面提及到 taskstatus 路由负责报告有后台任务提供的状态更新。这里就是这个路由的实现:

@app.route('/status/<task_id>')
def taskstatus(task_id):
    task = long_task.AsyncResult(task_id)
    if task.state == 'PENDING':
        // job did not start yet
        response = {
            'state': task.state,
            'current': 0,
            'total': 1,
            'status': 'Pending...'
        }
    elif task.state != 'FAILURE':
        response = {
            'state': task.state,
            'current': task.info.get('current', 0),
            'total': task.info.get('total', 1),
            'status': task.info.get('status', '')
        }
        if 'result' in task.info:
            response['result'] = task.info['result']
    else:
        # something went wrong in the background job
        response = {
            'state': task.state,
            'current': 1,
            'total': 1,
            'status': str(task.info),  # this is the exception raised
        }
    return jsonify(response)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

这个路由生成一个 JSON 响应，该响应包含任务的状态以及设置在 update_state() 调用中作为 meta 的参数的所有值，客户端可以使用这些构建一个进度条。遗憾地是这个函数需要检查一些条件，因此代码有些长。为了能够访问任务的数据，我重新创建了任务对象，该对象是 AsyncResult 类的实例，使用了 URL 中给的任务 id。

第一个 if 代码块是当任务还没有开始的时候(PENDING 状态)。在这种情况下暂时没有状态信息，因此我人为地制造了些数据。接下来的 elif 代码块返回后台的任务的状态信息。任务提供的信息可以通过访问 task.info 获得。如果数据中包含键 result ，这就意味着这是最终的结果并且任务已经结束，因此我把这些信息也加到响应中。最后的 else 代码块是任务执行失败的情况，这种情况下 task.info 中会包含异常的信息。

不管你是否相信，服务器所有要做的事情已经完成了。剩下的部分就是需要客户端需要实现的，在这里也就是用 JavaScript 脚本的网页来实现。

客户端的 Javascript

对于图形进度条使用 nanobar.js，从 CDN 上引用它。

# 图形进度条，可从CDN上引用它
<script src="//cdnjs.cloudflare.com/ajax/libs/nanobar/0.2.1/nanobar.min.js"></script>
# 引入jQuery，jianhuaajax的调用
<script src="//cdnjs.cloudflare.com/ajax/libs/jquery/2.1.3/jquery.min.js"></script>
1
2
3
4

启动连接后台作业的按钮的 Javascript 处理程序如下:

function start_long_task() {
    // add task status elements
    div = $('<div class="progress"><div></div><div>0%</div><div>...</div><div>&nbsp;</div></div><hr>');
    $('#progress').append(div);

    // create a progress bar
    var nanobar = new Nanobar({
        bg: '#44f',
        target: div[0].childNodes[0]
    });

    // send ajax POST request to start background job
    $.ajax({
        type: 'POST',
        url: '/longtask',
        success: function(data, status, request) {
            status_url = request.getResponseHeader('Location');
            update_progress(status_url, nanobar, div[0]);
        },
        error: function() {
            alert('Unexpected error');
        }
    });
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

div 的代码:

<div class="progress">
    <div></div>         <-- Progress bar
    <div>0%</div>       <-- Percentage
    <div>...</div>      <-- Status message
    <div>&nbsp;</div>   <-- Result
</div>
<hr>
1
2
3
4
5
6
7

最后 Javascript 的 update_progress 函数代码如下:

function update_progress(status_url, nanobar, status_div) {
      // send GET request to status URL
      $.getJSON(status_url, function(data) {
          // update UI
          percent = parseInt(data['current'] * 100 / data['total']);
          nanobar.go(percent);
          $(status_div.childNodes[1]).text(percent + '%');
          $(status_div.childNodes[2]).text(data['status']);
          if (data['state'] != 'PENDING' && data['state'] != 'PROGRESS') {
              if ('result' in data) {
                  // show result
                  $(status_div.childNodes[3]).text('Result: ' + data['result']);
              }
              else {
                  // something unexpected happened
                  $(status_div.childNodes[3]).text('Result: ' + data['state']);
              }
          }
          else {
              // rerun in 2 seconds
              setTimeout(function() {
                  update_progress(status_url, nanobar, status_div);
              }, 2000);
          }
      });
  }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

5. 运行示例

首先下载代码，代码的位于 Github repository:https://github.com/miguelgrinberg/flask-celery-example，接着执行以下的命令:

$ git clone https://github.com/miguelgrinberg/flask-celery-example.git
$ cd flask-celery-example
$ virtualenv venv
$ source venv/bin/activate
(venv) $ pip install -r requirements.txt
1
2
3
4
5

接着，启动 redis，关于 redis 的安装，启动以及配置，请参阅Redis 文档: http://redis.io/documentation/。

最后，执行如下命令运行示例:

$ export MAIL_USERNAME=<your-gmail-username>
$ export MAIL_PASSWORD=<your-gmail-password>
$ source venv/bin/activate
(venv) $ celery worker -A app.celery --loglevel=info
1
2
3
4

运行你的 Flask 应用来感受 Flask 和 Celery 一起工作的快乐:

$ source venv/bin/activate
(venv) $ python app.py
1
2

Previous:http://www.pythondoc.com/flask-celery/index.html

6 触发器

6.1 定时触发器

6.2 调用触发器

7. 踩过的一些坑

任务重复执行

使用 Celery Once 来防止 Celery 重复执行同一个任务：http://einverne.github.io/post/2018/01/use-celery-once-to-prevent-multiple-execution.html

在使用 Celery 的时候发现有的时候 Celery 会将同一个任务执行两遍，我遇到的情况是相同的任务在不同的 worker 中被分别执行，并且时间只相差几毫秒。这问题我一直以为是自己哪里处理的逻辑有问题，后来发现其他人 也有类似的问题，然后基本上出问题的都是使用 Redis 作为 Broker 的，而我这边一方面不想将 Redis 替换掉，就只能在 task 执行的时候加分布式锁了。

不过在 Celery 的 issue 中搜索了一下，有人使用 Redis 实现了分布式锁，然后也有人使用了 Celery Once。 大致看了一下 Celery Once ，发现非常符合现在的情况，就用了下。

Celery Once 也是利用 Redis 加锁来实现，他的使用非常简单，参照 GitHub 的使用很快就能够用上。Celery Once 在 Task 类基础上实现了 QueueOnce 类，该类提供了任务去重的功能，所以在使用时，我们自己实现的方法需要将 QueueOnce 设置为 base

@task(base=QueueOnce, once={'graceful': True})
后面的 once 参数表示，在遇到重复方法时的处理方式，默认 graceful 为 False，那样 Celery 会抛出 AlreadyQueued 异常，手动设置为 True，则静默处理。

另外如果要手动设置任务的 key，可以指定 keys 参数

@celery.task(base=QueueOnce, once={'keys': ['a']})
def slow_add(a, b):
    sleep(30)
    return a + b
总得来说，分为几步

第一步，安装

pip install -U celery_once
第二步，增加配置

from celery import Celery
from celery_once import QueueOnce
from time import sleep

celery = Celery('tasks', broker='amqp://guest@localhost//')
celery.conf.ONCE = {
  'backend': 'celery_once.backends.Redis',
  'settings': {
    'url': 'redis://localhost:6379/0',
    'default_timeout': 60 * 60
  }
}
第三步，修改 delay 方法

example.delay(10)
# 修改为
result = example.apply_async(args=(10))
第四步，修改 task 参数

@celery.task(base=QueueOnce, once={'graceful': True, keys': ['a']})
def slow_add(a, b):
    sleep(30)
    return a + b
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45

更多详细的参数可以参考 GitHub，或者直接阅读源码。

reference
https://github.com/cameronmaske/celery-once

celery 在执行task时有个机制,就是任务时长超过了 visibility_timeout 时还没执行完,就会指定其他worker重新开始task,默认的时长是一小时.

声明：本文内容由网友自发贡献，转载请注明出处：【wpsshop】