久久久久丰满少妇,九热最新视频

在學習Celery之前，我先簡單的去了解了一下什么是生產(chǎn)者消費者模式。

生產(chǎn)者消費者模式

在實際的軟件開發(fā)過程中，經(jīng)常會碰到如下場景：某個模塊負責產(chǎn)生數(shù)據(jù)，這些數(shù)據(jù)由另一個模塊來負責處理（此處的模塊是廣義的，可以是類、函數(shù)、線程、進程等）。產(chǎn)生數(shù)據(jù)的模塊，就形象地稱為生產(chǎn)者；而處理數(shù)據(jù)的模塊，就稱為消費者。

單單抽象出生產(chǎn)者和消費者，還夠不上是生產(chǎn)者消費者模式。該模式還需要有一個緩沖區(qū)處于生產(chǎn)者和消費者之間，作為一個中介。生產(chǎn)者把數(shù)據(jù)放入緩沖區(qū)，而消費者從緩沖區(qū)取出數(shù)據(jù)，如下圖所示：

生產(chǎn)者消費者模式是通過一個容器來解決生產(chǎn)者和消費者的強耦合問題。生產(chǎn)者和消費者彼此之間不直接通訊，而通過消息隊列（緩沖區(qū)）來進行通訊，所以生產(chǎn)者生產(chǎn)完數(shù)據(jù)之后不用等待消費者處理，直接扔給消息隊列，消費者不找生產(chǎn)者要數(shù)據(jù)，而是直接從消息隊列里取，消息隊列就相當于一個緩沖區(qū)，平衡了生產(chǎn)者和消費者的處理能力。這個消息隊列就是用來給生產(chǎn)者和消費者解耦的。------------->這里又有一個問題，什么叫做解耦？

解耦：假設生產(chǎn)者和消費者分別是兩個類。如果讓生產(chǎn)者直接調用消費者的某個方法，那么生產(chǎn)者對于消費者就會產(chǎn)生依賴（也就是耦合）。將來如果消費者的代碼發(fā)生變化，可能會影響到生產(chǎn)者。而如果兩者都依賴于某個緩沖區(qū)，兩者之間不直接依賴，耦合也就相應降低了。生產(chǎn)者直接調用消費者的某個方法，還有另一個弊端。由于函數(shù)調用是同步的（或者叫阻塞的），在消費者的方法沒有返回之前，生產(chǎn)者只好一直等在那邊。萬一消費者處理數(shù)據(jù)很慢，生產(chǎn)者就會白白糟蹋大好時光。緩沖區(qū)還有另一個好處。如果制造數(shù)據(jù)的速度時快時慢，緩沖區(qū)的好處就體現(xiàn)出來了。當數(shù)據(jù)制造快的時候，消費者來不及處理，未處理的數(shù)據(jù)可以暫時存在緩沖區(qū)中。等生產(chǎn)者的制造速度慢下來，消費者再慢慢處理掉。

因為太抽象，看過網(wǎng)上的說明之后，通過我的理解，我舉了個例子：吃包子。

假如你非常喜歡吃包子（吃起來根本停不下來），今天，你媽媽（生產(chǎn)者）在蒸包子，廚房有張桌子（緩沖區(qū)），你媽媽將蒸熟的包子盛在盤子（消息）里，然后放到桌子上，你正在看巴西奧運會，看到蒸熟的包子放在廚房桌子上的盤子里，你就把盤子取走，一邊吃包子一邊看奧運。在這個過程中，你和你媽媽使用同一個桌子放置盤子和取走盤子，這里桌子就是一個共享對象。生產(chǎn)者添加食物，消費者取走食物。桌子的好處是，你媽媽不用直接把盤子給你，只是負責把包子裝在盤子里放到桌子上，如果桌子滿了，就不再放了，等待。而且生產(chǎn)者還有其他事情要做，消費者吃包子比較慢，生產(chǎn)者不能一直等消費者吃完包子把盤子放回去再去生產(chǎn)，因為吃包子的人有很多，如果這期間你好朋友來了，和你一起吃包子，生產(chǎn)者不用關注是哪個消費者去桌子上拿盤子，而消費者只去關注桌子上有沒有放盤子，如果有，就端過來吃盤子中的包子，沒有的話就等待。對應關系如下圖：

考察了一下，原來當初設計這個模式，主要就是用來處理并發(fā)問題的，而Celery就是一個用python寫的并行分布式框架。

然后我接著去學習Celery

Celery的定義

Celery（芹菜）是一個簡單、靈活且可靠的，處理大量消息的分布式系統(tǒng)，并且提供維護這樣一個系統(tǒng)的必需工具。

我比較喜歡的一點是：Celery支持使用任務隊列的方式在分布的機器、進程、線程上執(zhí)行任務調度。然后我接著去理解什么是任務隊列。

任務隊列

任務隊列是一種在線程或機器間分發(fā)任務的機制。

消息隊列

消息隊列的輸入是工作的一個單元，稱為任務，獨立的職程（Worker）進程持續(xù)監(jiān)視隊列中是否有需要處理的新任務。

Celery 用消息通信，通常使用中間人（Broker）在客戶端和職程間斡旋。這個過程從客戶端向隊列添加消息開始，之后中間人把消息派送給職程，職程對消息進行處理。如下圖所示：

Celery 系統(tǒng)可包含多個職程和中間人，以此獲得高可用性和橫向擴展能力。

Celery的架構

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

消息中間件

Celery本身不提供消息服務，但是可以方便的和第三方提供的消息中間件集成，包括，RabbitMQ,Redis,MongoDB等，這里我先去了解RabbitMQ,Redis。

任務執(zhí)行單元

Worker是Celery提供的任務執(zhí)行的單元，worker并發(fā)的運行在分布式的系統(tǒng)節(jié)點中

任務結果存儲

Task result store用來存儲Worker執(zhí)行的任務的結果，Celery支持以不同方式存儲任務的結果，包括Redis，MongoDB，Django ORM，AMQP等，這里我先不去看它是如何存儲的，就先選用Redis來存儲任務執(zhí)行結果。

然后我接著去安裝Celery，在安裝Celery之前，我已經(jīng)在自己虛擬機上安裝好了Python，版本是2.7，是為了更好的支持Celery的3.0以上的版本。

因為涉及到消息中間件，所以我先去選擇一個在我工作中要用到的消息中間件（在Celery幫助文檔中稱呼為中間人），為了更好的去理解文檔中的例子，我安裝了兩個中間件，一個是RabbitMQ,一個redis。

在這里我就先根據(jù)Celery3.1的幫助文檔安裝和設置RabbitMQ, 要使用 Celery，我們需要創(chuàng)建一個 RabbitMQ 用戶、一個虛擬主機，并且允許這個用戶訪問這個虛擬主機。下面是我在個人虛擬機Ubuntu14.04上的設置：

$ sudo rabbitmqctl add_user forward password

#創(chuàng)建了一個RabbitMQ用戶,用戶名為forward，密碼是password

$ sudo rabbitmqctl add_vhost ubuntu

#創(chuàng)建了一個虛擬主機，主機名為ubuntu

$ sudo rabbitmqctl set_permissions -p ubuntu forward ".*" ".*" ".*"

#允許用戶forward訪問虛擬主機ubuntu，因為RabbitMQ通過主機名來與節(jié)點通信

$ sudo rabbitmq-server

之后我啟用RabbitMQ服務器，結果如下，成功運行：

之后我安裝Redis,它的安裝比較簡單,如下：

$ sudo pip install redis

然后進行簡單的配置，只需要設置 Redis 數(shù)據(jù)庫的位置:

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

URL的格式為：

redis://:password@hostname:port/db_number

URL Scheme 后的所有字段都是可選的，并且默認為 localhost 的 6379 端口，使用數(shù)據(jù)庫 0。我的配置是：

redis://:password@ubuntu:6379/5

之后安裝Celery，我是用標準的Python工具pip安裝的，如下：

$ sudo pip install celery

為了測試Celery能否工作，我運行了一個最簡單的任務，編寫tasks.py，如下圖所示：

編輯保存退出后，我在當前目錄下運行如下命令：

$ celery -A tasks worker --loglevel=info

#查詢文檔，了解到該命令中-A參數(shù)表示的是Celery APP的名稱，這個實例中指的就是tasks.py,后面的tasks就是APP的名稱，worker是一個執(zhí)行任務角色，后面的loglevel=info記錄日志類型默認是info,這個命令啟動了一個worker,用來執(zhí)行程序中add這個加法任務（task）。

然后看到界面顯示結果如下：

我們可以看到Celery正常工作在名稱ubuntu的虛擬主機上，版本為3.1.23，在下面的[config]中我們可以看到當前APP的名稱tasks，運輸工具transport就是我們在程序中設置的中間人redis://127.0.0.1:6379/5，result我們沒有設置，暫時顯示為disabled,然后我們也可以看到worker缺省使用perfork來執(zhí)行并發(fā)，當前并發(fā)數(shù)顯示為1，然后可以看到下面的[queues]就是我們說的隊列，當前默認的隊列是celery,然后我們看到下面的[tasks]中有一個任務tasks.add.

了解了這些之后，根據(jù)文檔我重新打開一個terminal,然后執(zhí)行Python,進入Python交互界面，用delay()方法調用任務，執(zhí)行如下操作：

這個任務已經(jīng)由之前啟動的Worker異步執(zhí)行了，然后我打開之前啟動的worker的控制臺,對輸出進行查看驗證，結果如下：

綠色部分第一行說明worker收到了一個任務：tasks.add,這里我們和之前發(fā)送任務返回的AsyncResult對比我們發(fā)現(xiàn)，每個task都有一個唯一的ID，第二行說明了這個任務執(zhí)行succeed,執(zhí)行結果為12。

查看資料說調用任務后會返回一個AsyncResult實例，可用于檢查任務的狀態(tài)，等待任務完成或獲取返回值（如果任務失敗，則為異常和回溯）。但這個功能默認是不開啟的，需要設置一個 Celery 的結果后端（backend），這塊我在下一個例子中進行了學習。

通過這個例子后我對Celery有了初步的了解，然后我在這個例子的基礎上去進一步的學習。

因為Celery是用Python編寫的，所以為了讓代碼結構化一些，就像一個應用，我使用python包，創(chuàng)建了一個celery服務，命名為pj。文件目錄如下：

celery.py

from __future __ import absolute_import

#定義未來文件的絕對進口，而且絕對進口必須在每個模塊的頂部啟用。

from celery import Celery

#從celery導入Celery的應用程序接口

App.config_from_object(‘pj.config’)

#從config.py中導入配置文件

if __name__ == ‘__main__’:

app.start()

#執(zhí)行當前文件，運行celery

app = Celery(‘pj’,

broker=‘redis://localhost’,

backend=‘redis://localhost’,

include=[‘pj.tasks’]

)

#首先創(chuàng)建了一個celery實例app,實例化的過程中，制定了任務名pj(與當前文件的名字相同)，Celery的第一個參數(shù)是當前模塊的名稱，在這個例子中就是pj,后面的參數(shù)可以在這里直接指定，也可以寫在配置文件中，我們可以調用config_from_object()來讓Celery實例加載配置模塊，我的例子中的配置文件起名為config.py,配置文件如下：

在配置文件中我們可以對任務的執(zhí)行等進行管理，比如說我們可能有很多的任務,但是我希望有些優(yōu)先級比較高的任務先被執(zhí)行,而不希望先進先出的等待。那么需要引入一個隊列的問題. 也就是說在我的broker的消息存儲里面有一些隊列，他們并行運行，但是worker只從對應的隊列里面取任務。在這里我們希望tasks.py中的add先被執(zhí)行。task中我設置了兩個任務：

所以我通過from celery import group引入group,用來創(chuàng)建并行執(zhí)行的一組任務。然后這塊現(xiàn)需要理解的就是這個@app.task,@符號在python中用作函數(shù)修飾符，到這塊我又回頭去看python的裝飾器(在代碼運行期間動態(tài)增加功能的方式)到底是如何實現(xiàn)的，在這里的作用就是通過task()裝飾器在可調用的對象（app）上創(chuàng)建一個任務。