目錄

• • 目錄
    • 使用場景
    • 程式碼編寫
    • 底層實現

專案的開發經常用到celery後臺任務，下面記錄下我在專案中用到celery的地方以及自己的一些思考。

使用場景

1. 在web開發時有一些耗時的操作需要執行，但是你又不想一直阻塞前端，那麼可以嘗試使用celery的後臺任務，將請求傳送到celery後臺，然後前端不再阻塞，最後celery後臺將任務完成之後將結果返回。使用者就可以知道任務是否執行成功。
2. 做資料庫非同步更新，可能要在某個時間點與遠端資料庫進行同步，更新的時間也比較久，不希望阻塞前端的程序。

程式碼編寫

首先來看一下一個簡單的celery後臺任務。專案裡的部署任務是一個耗時的函式，使用者在前端點選了一個部署按鈕之後可以執行其他操作，最後會將部署的結果返回給使用者，更新任務狀態。這裡的修飾器是傳入了celery任務的上下文。

@celery_ext.celery.task(base=RequestContextTask, ignore_result=True)
def deploy(**kwargs):
    """執行部署任務.

    :param task_id: 任務 id
    :param update_task_state_url: 更新任務狀態通知的 URL
    """
    # 更新任務狀態
    # 執行作業系統安裝

在全域性flask框架入口init，一般在這個檔案裡我們建立全域性的flask物件，然後進行外掛初始化，並且註冊檢視和藍圖。我們初始化了celery_ext,進行外掛的初始化

celery_ext = FlaskCeleryExt()
...
def create_app(config)
    ...
    celery_ext.init_app(app)
    ...

再來看下FlaskCeleryExt這個類，主要做的是物件一些初始化工作

class FlaskCeleryExt(object):
    """Flask-Celery extension."""

    def __init__(self, app=None, create_celery_app=None):
        """Initialize the Flask-CeleryExt."""
 

        self.create_celery_app = create_celery_app or pkg_create_celery_app
        self.celery = None
        if app is not None:
            self.init_app(app)

    def init_app(self, app):
        """Initialize a Flask application."""
        # Follow the Flask guidelines on usage of app.extensions
        if not hasattr(app, 'extensions'):
            app.extensions = {}  # pragma: no cover
        app.extensions['flask-celeryext'] = self
        self.celery = self.create_celery_app(app)

當然啦，還有一些配置的工作。我們一般將全域性的配置檔案寫在config.py裡面。這裡設定了celery broker url 以及一些後臺任務的屬性配置。也可以在設定BACKEND的URL。我們在setup app的時候會用到這些配置的引數。

class Config(object):
    """框架配置
    配置 Flask 所必需的引數，其它系統引數應儲存到資料庫
    """
    # celery
    CELERY_BROKER_URL = 'amqp://guest:[email protected]:5672//'
    CELERY_TASK_ALWAYS_EAGER = False
    CELERY_TASK_EAGER_PROPAGATES_EXCEPTIONS = True

在manage.py中新增啟動命令列

@manager.option('-l', '--loglevel', dest='loglevel', default='info')
@manager.option('-c', '--concurrency', dest='concurrency', type=int, default=2)
def worker(loglevel='info', concurrency=2):
    """啟動後臺的 celery worker"""
    from celery.apps.worker import Worker

    worker_entry = Worker(app=celery, loglevel=loglevel, concurrency=concurrency)
    worker_entry.start()

最後在執行的時候，命令列需要啟動rabbitmq以及celery worker。

# 執行rabbitmq
    sudo service rabbitmq start

# 執行celery worker
    celery worker -A manage.celery --loglevel=info

底層實現

1. python修飾器
   修飾器本身其實是一個python函式，他可以讓其他函式在不需要做任何程式碼改動的前提下增加額外的功能。引數可以是一個函式，返回值也可以是一個函式，好處在於不需要修改原來函式的程式碼並且為他新增新的功能，也能使一些常用的函式程式碼模組得到重用，減少冗餘度。另外修飾器可以進行疊加。celery用到修飾器的地方是設定celery後臺任務的上下文。

2. celery實現
   celery是python多工庫來編寫的任務佇列工具，可以並行地執行一些任務。
   python與celery之間使用了訊息佇列作為中間人來進行任務佇列的管理。
   中介軟體的選擇可以是rabbitmq也可以是redis，看開發的需求吧。我們專案用的是rabbitmq。
   大致的一個邏輯圖是這樣子的：
   
   worker是執行單元，其實也就是多個執行緒。
   broker是訊息中介軟體，用於接收並暫存訊息佇列。
   backend是儲存每次任務的結果。