目錄

前文列表

擴充套件閱讀

Celery

Celery 是使用 Python 多工庫來編寫的任務佇列工具, 可以 並行 的執行任務. 我們會將執行時間較長但又不那麼追求實時的功能以非同步任務的形式完成, EG. 上傳檔案, 傳送郵件…, Python 和 Celery 之間需要一箇中間人(訊息佇列)來進行任務佇列的管理, Celery 官方推薦使用 RabbirMQ 或 Redis 來充當這個角色. 當然也可以同時使用兩者, 其中 MQ 作為中間人, Redis 傳遞 Celery 執行的結果給應用端. 這樣做的優勢在於, 返回給應用的結果是持久化儲存在資料庫中的.

訊息佇列: 是一種專門設計的系統, 用於在生產者(往佇列傳送訊息的程式)和消費者(從佇列中取出訊息的佇列)之間傳遞訊息.

• 安裝 Celery

pip install Celery

• 安裝 Flask-Celery-Helper
  Flask-Celery-Helper 是一個 Flask 擴充套件, 用於輔助使用 app 來初始化 Celery 物件, 使其得以註冊到 app 物件中.

pip install Flask-Celery-Helper
pip freeze > requirments.txt

• 安裝 RabbitMQ: 官方安裝文件
  啟動 RabbitMQ

/etc/init.d/rabbitmq-server start

將 Celery 加入到應用中

• 配置 Celery 連線 RabbitMQ 的 URL
  vim jmilkfansblog/config.py

class DevConfig(Config):
    """Development config class."""
...
    # Celery <--> RabbitMQ connection
    CELERY_RESULT_BACKEND = "amqp://guest:[email protected]:5672//"
    CELERY_BROKER_URL = "amqp://guest:
 
[email protected]:5672//"

NOTE: RabbitMQ 使用預設的 guest 使用者, 埠為 5672

• 建立 celery 物件
  vim jmilkfansblog/extensions.py

from flask.ext.celery import Celery

...

# Create the Flask-Celery-Helper's instance
flask_celery = Celery()

• 將 celery 物件註冊到 app 物件
  vim jmilkfansblog/__init__.py

from jmilkfansblog.extensions import flask_celery

def create_app(object_name):
    """Create the app instance via `Factory Method`"""
...
    # Init the Flask-Celery-Helper via app object
    # Register the celery object into app object
    flask_celery.init_app(app)

NOTE 1: Celery 的程序必須在 Flask app 的上下文中執行, 這樣 Celery 才能夠跟其他的 Flask 擴充套件協同工作。所以必須將 flask_celery 物件註冊到 app 物件中, 並且每建立一個 Celery 程序都需要建立一個新的 Flask app 物件, 這裡我們使用工廠模式來建立 celery 物件。這一點是非常重要的，實際上 Flask application 和 Celery application 是兩個不同的程序，在 Celery 沒有加入 Flask 上下文的情況下，Celery 的程式邏輯就不能輕易的訪問 Flask 相關資源，比如不能載入 Flask 的環境配置資訊，無法通過 Flask 來訪問資料庫，不能使用 Flask 的擴充套件功能等。如果想做到這些，Celery 都需要自己再實現一套相同的邏輯，這樣做顯然是沒有必要的。所以 Flask application 原生支援將自己的 Context 嵌入到別的 application 中，當然有些情況也需要相應擴充套件的輔助，例如 Flask-Celery-Helper 在這裡就充當著這個角色。

NOTE 2: flask_celery 物件是 Flask-Celery-Helper 擴充套件的物件, 用於輔助處理 Celery 的初始化, 所以實際上我們是可以不使用這個擴充套件, 而直接使用 Celery 的. celery 物件才是真正的 Celery 的物件.

• 使用工廠模式來建立 celery 物件
  ./celery_runner.py

mport os

from celery import Celery

from jmilkfansblog import create_app


def make_celery(app):
    """Create the celery process."""

    # Init the celery object via app's configuration.
    celery = Celery(
        app.import_name,
        backend=app.config['CELERY_RESULT_BACKEND'],
        broker=app.config['CELERY_BROKER_URL'])

    # Flask-Celery-Helpwe to auto-setup the config.
    celery.conf.update(app.config)
    TaskBase = celery.Task

    class ContextTask(TaskBase):

        abstract = True

        def __call__(self, *args, **kwargs):
            """Will be execute when create the instance object of ContextTesk."""

            # Will context(Flask's Extends) of app object(Producer Sit) 
            # be included in celery object(Consumer Site).
            with app.app_context():
                return TaskBase.__call__(self, *args, **kwargs)

    # Include the app_context into celery.Task.
    # Let other Flask extensions can be normal calls.
    celery.Task = ContextTask
    return celery

env = os.environ.get('BLOG_ENV', 'dev')
flask_app = create_app('jmilkfansblog.config.%sConfig' % env.capitalize())
# 1. Each celery process needs to create an instance of the Flask application.
# 2. Register the celery object into the app object.
celery = make_celery(flask_app)

NOTE 1: 我們以後會以 CLI 的形式來管理和控制 Celery 的 worker, 所以我們將 celery 物件的實現模組放置在 ./celery_runner.py 中, 而不是 jmilkfansblog/celery_runner.py. Flask app 內部的 tasks 任何的定義和實現就交由 Flask-Celery-Helper 來支援就好了, 這也是 Flask-Celery-Helper 存在的意義.

NOTE 2: make_celery()最重要的作用就是讓每個 Celery 的程序中(celery物件)都包含有 app 物件的上下文, 至於為什麼這麼做呢? 上述已經給出了答案.

NOTE 3: 這裡通過 create_app() 建立的物件不能夠命名為 app, 而是命名為 flask_app, 這是因為 Celery 會預設將命名為 app 或 celery 的物件都作為一個 Celery 物件來處理.

NOTE 4: celery.conf.update(app.config) 會將 app 物件的 config 更新到 celery 物件中, 當然也包括了剛剛定義的 RabbitMQ 連線 URL 配置.

• 啟動 Celery 服務

(env) [email protected]:/opt/JmilkFan-s-Blog$ celery worker -A celery_runner --loglevel=info
...

 -------------- celery@JmilkFan-Devstack v4.0.1 (latentcall)
---- **** ----- 
--- * ***  * -- Linux-4.4.0-53-generic-x86_64-with-Ubuntu-16.04-xenial 2016-12-15 19:12:33
-- * - **** --- 
- ** ---------- [config]
- ** ---------- .> app:         jmilkfansblog:0x7f5b24345990
- ** ---------- .> transport:   amqp://guest:**@localhost:5672//
- ** ---------- .> results:     amqp://
- *** --- * --- .> concurrency: 4 (prefork)
-- ******* ---- .> task events: OFF (enable -E to monitor tasks in this worker)
--- ***** ----- 
 -------------- [queues]
                .> celery           exchange=celery(direct) key=celery


[tasks]
  . jmilkfansblog.tasks.digest
  . jmilkfansblog.tasks.log
  . jmilkfansblog.tasks.multiply
  . jmilkfansblog.tasks.remind

NOTE: 在啟動 Celery 服務的時候, 能夠用 Output 看見其自身的配置資訊和現在所處理的 tasks.

實現向新使用者傳送歡迎郵件

下面使用 Celery 實現在使用者建立賬戶之後, 在指定的時間內非同步的向新使用者傳送歡迎郵件.

• 新增 Reminder Model 使用者存放使用者的 email 地址和歡迎內容.
  vim jmilkfansblog/models.py

class Reminder(db.Model):
    """Represents Proected reminders."""

    __tablename__ = 'reminders'
    id = db.Column(db.String(45), primary_key=True)
    date = db.Column(db.DateTime())
    email = db.Column(db.String(255))
    text = db.Column(db.Text())

    def __init__(self, id, text):
        self.id = id
        self.email = text

    def __repr__(self):
        return '<Model Reminder `{}`>'.format(self.text[:20])

• 建立一個 task
  vim jmilkfansblog/tasks.py

import smtplib
import datetime
from email.mime.text import MIMEText

from flask_mail import Message

from jmilkfansblog.extensions import flask_celery, mail

@flask_celery.task(
    bind=True,
    igonre_result=True,
    default_retry_delay=300,
    max_retries=5)
def remind(self, primary_key):
    """Send the remind email to user when registered.
       Using Flask-Mail.
    """

    reminder = Reminder.query.get(primary_key)

    msg = MIMEText(reminder.text)
    msg['Subject'] = 'Welcome!'
    msg['FROM'] = <your_email>
    msg['To'] = reminder.email

    try:
        smtp_server = smtplib.SMTP('localhost')
        smtp_server.starttls()
        smtp_server.login(<user>, <password>)
        smtp_server.sendmail(<your_email>,
                             [reminder.email],
                             msg.as_string())
        smtp_server.close()
        return

    except Exception as err:
        self.retry(exc=err)

def on_reminder_save(mapper, connect, self):
    """Callbask for task remind."""
    remind.apply_async(args=(self.id), eta=self.date)

NOTE 1: Celery Task 本質上就是一個被 celery.task()裝飾過的函式,

NOTE 2: 使用主鍵 primary_key 來獲取 reminder 物件是為了避免資料競態的發生, 因為從生成 reminder 物件到 task 被執行的過程並不能保證資料是最新的, 這也是處理非同步呼叫時, 需要時刻注意的地方.

NOTE 3: on_reminder_save() 是一個回撥函式, 當我們在一個特定的情景下呼叫這個函式的時候就觸發了一個 Celery task.

• 應用 SQLAlchemy 的 event 特性來觸發 Celery task
  vim jmilkfansblog/__init__.py

from sqlalchemy import event


def create_app(object_name):
    """Create the app instance via `Factory Method`"""
...
    # Will be callback on_reminder_save when insert recond into table `reminder`.
    event.listen(Reminder, 'after_insert', on_reminder_save)

• NOTE 1: SQLAlchemy 允許在 Model 上註冊回撥函式, 當 Model 物件發生特定的情景時, 就會執行這個回撥函式, 這就是所謂的 event, 這裡我們使用 after_insert 來指定當建立一個新的 Reminder 物件(插入一條記錄)時就觸發這個回撥函式. 而是回撥函式中的形參, 會由 event 來負責傳入.