Celery詳解

1、背景

由于從事區塊鏈錢包相關開發，對于區塊鏈鏈上資源需要頻繁的進行檢查同步，在flask項目中，對于celery這個異步任務執行工具，使用的頻率算是相當的高，今天，我就來簡單總結一下celery的概念和使用方法。

2、形象比喻

Celery是一個異步任務的調度工具，是Distributed Task Queue，分布式任務隊列，分布式決定了可以有多個worker的存在，隊列表示其是異步操作，即存在一個產生任務提出需求的工頭，和一群等著被分配工作的碼農。

在python中定義Celery的時候，我們要引入Broker，中文翻譯過來就是"中間人"的意思，在這里Broker起到一個中間人的角色，在工頭提出任務的時候，把所有的任務放到Broker里面，在Broker的另一頭，一群碼農等著取出一個個任務準備著手做。

這種模式注定了整個系統會是個開環系統，工頭對于碼農們把任務做的怎樣是不知情的，所以我們要引入Backend來保存每次任務的結果。這個Backend有點像我們的Broker，也是存儲信息用的，只不過這里存的是那些任務的返回結果。我們可以選擇只讓錯誤執行的任務返回結果到Backend，這樣我們取回結果，便可以知道有多少任務執行失敗了。

3、celery具體介紹

3.1 Broker

broker是一個消息傳輸的中間件，它是用來存儲生產出來的各種待執行任務的。每當應用程序調用celery的異步任務的時候，會向broker傳遞消息，而后celery的worker將會取到消息，進行程序執行，broker可以看成是一個消息隊列，其中broker的中文意思是經紀人，用來發送和接受信息。這個broker有幾個方案可供選擇：RabbitMQ(消息隊列)，Redis(緩存數據庫)，數據庫(不推薦)，等等。

3.2 Backend

通常程序發送的消息，發完就完了，可能都不知道對方什么時候接受了，為此，celery實現了一個backend，用于存儲這些消息以及celery執行的一些消息和結果，Backend是在Celery的配置中的一個配置項CELERY_RESULT_BACKEND，作用是保存結果和狀態，如果你需要跟蹤任務的狀態，那么需要設置這一項，可以是Database backend，也可以是Cache backend。

對于brokers，官方推薦是rabbitmq和redis，至于backend，就是數據庫，為了簡單可以都使用redis。在我的項目中，都是使用redis。

4、使用

4.1 celery架構

Celery的架構由消息中間件(message broker)，任務執行單元(worker)和任務執行結果存儲(task result store)三部分組成。消息中間件

Celery本身不提供消息服務，但是可以方便的和第三方提供的消息中間件集成，包括RabbitMQ，Redis，MongoDB等任務執行單元

Worker是celery提供的任務執行的單元，worker并發的運行在分布式的系統節點中。任務結果存儲

Task result store用來存儲Worker執行的任務的結果，Celery支持以不同方式存儲任務的結果，包括AMQP，redis，memcached，mongodb，SQLAlchemy，Django等

4.2 安裝redis+celery

安裝Redis,它的安裝比較簡單：

~$ pip install redis

然后進行配置，一般都在項目的config.py文件里配置：

CELERY_BROKER_URL = "redis://localhost:6379/0"

URL的格式為：redis://:password@hostname:port/db_number

URL Scheme后的所有字段都是可選的，并且默認為localhost的6379端口，也就是redis的默認端口，使用數據庫0。

安裝Celery：

~$ pip install celery

4.3 使用Celery

使用celery包含三個方面：1，定義任務函數 2，運行celery服務 3，客戶應用程序的調用

創建一個文件tasks.py輸入下列代碼：

from celery import Celery

broker = "redis://localhost:6379/0"

backend = "redis://localhost:6379/1"

app = Celery("tasks", broker=broker, backend=backend)

@app.task

def add(x, y)

return x + y

上述代碼導入了celery，然后創建了celery實例app,實例化的過程中指定了任務名tasks(和文件名一致)，傳入了broker和backend。然后創建了一個任務函數add。下面啟動

celery服務，在當前命令行終端運行：

~$ celery -A tasks worker

目錄結構(celery -A tasks worker --loglevel=info這條命令當前工作目錄必須和tasks.py所在的目錄相同，即進入tasks.py所在目錄執行這條命令)

調用delay函數即可啟動add這個任務，這個函數的效果是發送一條消息到broker中去，這個消息包括要執行的函數，函數的參數以及其他消息，具體的可以看Celery官方文檔。這個時候worker會等待broker中的消息，一旦收到消息就會立刻執行消息。

注意：如果把返回值賦值給一個變量，那么原來的應用程序也會被阻塞，需要等待異步任務返回的結果，因此，實際使用中，不需要把結果賦值。

使用配置文件

4.4 健壯性

上述的使用是簡單的配置，下面介紹一個更健壯的方式來使用celery。首先創建一個python包，celery服務，姑且命名為proj。目錄文件如下：

|- proj

|-- __init__.py

|-- celery.py # 創建celery實例

|-- config.py # 配置文件

|-- tasks.py # 任務函數

首先是 celery.py

from __future__ import absolute_import

from celery import Celery

app = Celery("proj", include=["proj.tasks"])

app.config_from_object("proj.config")

if __name__ == "__main__":

app.start()

這一次創建app，并沒有直接指定broker和backend。而是在配置文件中。

然后是 config.py

from __future__ import absolute_import

BROKER_URL = "redis://localhost:6379/0"

CELERY_BACKEND_URL = "redis://localhost:6379/1"

最后是 tasks.py

from __future__ import absolute_import

from proj.celery import app

@app.task

def add(x, y):

return x + y

使用方法也很簡單，在proj的同一級目錄執行celery:

celery -A proj worker -l info

現在使用任務也很簡單，直接在客戶端代碼調用proj.tasks里的函數即可。

4.5 定時任務

Scheduler(定時任務，周期性任務)

一種常見的需求是每隔一段時間執行一個任務

在celery中執行定時任務非常簡單，只需要設置celery對象的CELERYBEAT_SCHEDULE屬性即可。

配置如下 config.py

from __future__ import absolute_import

BROKER_URL = "redis://localhost:6379/0"

CELERY_BACKEND_URL = "redis://localhost:6379/1"

CELERY_TIMEZONE = "Asia/Shanghai"

from datetime import timedelta

CELERYBEAT_SCHEDULE = {

'add-every-30-seconds':{

'task':'proj.tasks.add',

'schedule':timedelta(seconds=30),

'args':(16, 16)

},

}

注意配置文件需要指定時區，這段代碼表示每隔30秒執行add函數，一旦使用了scheduler，啟動celery需要加上-B參數。

celery -A proj worker -B -l info

對于celery的介紹就到這里了，歡迎交流技術難點。

總結

以上是生活随笔為你收集整理的celery java_Celery详解的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。