celery 設(shè)置多少時(shí)間后運(yùn)行

許多Django應(yīng)用需要執(zhí)行異步任務(wù), 以便不耽誤http request的執(zhí)行. 我們也可以選擇許多方法來(lái)完成異步任務(wù), 使用Celery是一個(gè)比較好的選擇, 因?yàn)镃elery

公司主營(yíng)業(yè)務(wù)：成都網(wǎng)站制作、網(wǎng)站建設(shè)、移動(dòng)網(wǎng)站開發(fā)等業(yè)務(wù)。幫助企業(yè)客戶真正實(shí)現(xiàn)互聯(lián)網(wǎng)宣傳，提高企業(yè)的競(jìng)爭(zhēng)能力。創(chuàng)新互聯(lián)公司是一支青春激揚(yáng)、勤奮敬業(yè)、活力青春激揚(yáng)、勤奮敬業(yè)、活力澎湃、和諧高效的團(tuán)隊(duì)。公司秉承以“開放、自由、嚴(yán)謹(jǐn)、自律”為核心的企業(yè)文化，感謝他們對(duì)我們的高要求，感謝他們從不同領(lǐng)域給我們帶來(lái)的挑戰(zhàn)，讓我們激情的團(tuán)隊(duì)有機(jī)會(huì)用頭腦與智慧不斷的給客戶帶來(lái)驚喜。創(chuàng)新互聯(lián)公司推出齊齊哈爾免費(fèi)做網(wǎng)站回饋大家。

有著大量的社區(qū)支持, 能夠完美的擴(kuò)展, 和Django結(jié)合的也很好. Celery不僅能在Django中使用, 還能在其他地方被大量的使用. 因此一旦學(xué)會(huì)使用Celery, 我

們可以很方便的在其他項(xiàng)目中使用它.

1. Celery版本

本篇博文主要針對(duì)Celery 3.0.x. 早期版本的Celery可能有細(xì)微的差別.

2. Celery介紹

Celery的主要用處是執(zhí)行異步任務(wù), 可以選擇延期或定時(shí)執(zhí)行功能. 為什么需要執(zhí)行異步任務(wù)呢?

第一, 假設(shè)用戶正發(fā)起一個(gè)request, 并等待request完成后返回. 在這一request后面的view功能中, 我們可能需要執(zhí)行一段花費(fèi)很長(zhǎng)時(shí)間的程序任務(wù), 這一時(shí)間

可能遠(yuǎn)遠(yuǎn)大于用戶能忍受的范圍. 當(dāng)這一任務(wù)并不需要立刻執(zhí)行時(shí), 我們便可以使用Celery在后臺(tái)執(zhí)行, 而不影響用戶瀏覽網(wǎng)頁(yè). 當(dāng)有任務(wù)需要訪問遠(yuǎn)程服務(wù)器完

成時(shí), 我們往往都無(wú)法確定需要花費(fèi)的時(shí)間.

第二則是定期執(zhí)行某些任務(wù). 比如每小時(shí)需要檢查一下天氣預(yù)報(bào), 然后將數(shù)據(jù)儲(chǔ)存到數(shù)據(jù)庫(kù)中. 我們可以編寫這一任務(wù), 然后讓Celery每小時(shí)執(zhí)行一次. 這樣我們

的web應(yīng)用便能獲取最新的天氣預(yù)報(bào)信息.

我們這里所講的任務(wù)task, 就是一個(gè)Python功能(function). 定期執(zhí)行一個(gè)任務(wù)可以被認(rèn)為是延時(shí)執(zhí)行該功能. 我們可以使用Celery延遲5分鐘調(diào)用function

task1, 并傳入?yún)?shù)(1, 2, 3). 或者我們也可以每天午夜運(yùn)行該function.

我們偏向于將Celery放入項(xiàng)目中, 便于task訪問統(tǒng)一數(shù)據(jù)庫(kù)和Django設(shè)置.

當(dāng)task準(zhǔn)備運(yùn)行時(shí), Celery會(huì)將其放入列隊(duì)queue中. queue中儲(chǔ)存著可以運(yùn)行的task的list. 我們可以使用多個(gè)queue, 但為了簡(jiǎn)單, 這里我們只使用一個(gè).

將任務(wù)task放入queue就像加入todo list一樣. 為了使task運(yùn)行, 我們還需要在其他線程中運(yùn)行的苦工worker. worker實(shí)時(shí)觀察著代運(yùn)行的task, 并逐一運(yùn)行這

些task. 你可以使用多個(gè)worker, 通常他們位于不同服務(wù)器上. 同樣為了簡(jiǎn)單起見, 我們這只是用一個(gè)worker.

我們稍后會(huì)討論queue, worker和另外一個(gè)十分重要的進(jìn)程, 接下來(lái)我們來(lái)動(dòng)動(dòng)手:

3. 安裝Celery

我們可以使用pip在vietualenv中安裝:

pip install django-celery

4. Django設(shè)置

我們暫時(shí)使用django runserver來(lái)啟動(dòng)celery. 而Celery代理人(broker), 我們使用Django database broker implementation. 現(xiàn)在我們只需要知道Celery

需要broker, 使用django自身便可以充當(dāng)broker. (但在部署時(shí), 我們最好使用更穩(wěn)定和高效的broker, 例如Redis.)

在settings.py中:

import djcelery

djcelery.setup_loader()

BROKER_URL = 'django://'

...

INSTALLED_APPS = (

...

'djcelery',

'kombu.transport.django',

...

)

第一二項(xiàng)是必須的, 第三項(xiàng)則告訴Celery使用Django項(xiàng)目作為broker.

在INSTALLED_APPS中添加的djcelery是必須的. kombu.transport.django則是基于Django的broker

最后創(chuàng)建Celery所需的數(shù)據(jù)表, 如果使用South作為數(shù)據(jù)遷移工具, 則運(yùn)行:

python manage.py migrate

否則運(yùn)行: (Django 1.6或Django 1.7都可以)

python manage.py syncdb

5. 創(chuàng)建一個(gè)task

正如前面所說的, 一個(gè)task就是一個(gè)Pyhton function. 但Celery需要知道這一function是task, 因此我們可以使用celery自帶的裝飾器decorator: @task. 在

django app目錄中創(chuàng)建taske.py:

from celery import task

@task()

def add(x, y):

return x + y

當(dāng)settings.py中的djcelery.setup_loader()運(yùn)行時(shí), Celery便會(huì)查看所有INSTALLED_APPS中app目錄中的tasks.py文件, 找到標(biāo)記為task的function, 并

將它們注冊(cè)為celery task.

將function標(biāo)注為task并不會(huì)妨礙他們的正常執(zhí)行. 你還是可以像平時(shí)那樣調(diào)用它: z = add(1, 2).

6. 執(zhí)行task

讓我們以一個(gè)簡(jiǎn)單的例子作為開始. 例如我們希望在用戶發(fā)出request后異步執(zhí)行該task, 馬上返回response, 從而不阻塞該request, 使用戶有一個(gè)流暢的訪問

過程. 那么, 我們可以使用.delay, 例如在在views.py的一個(gè)view中:

from myapp.tasks import add

...

add.delay(2, 2)

...

Celery會(huì)將task加入到queue中, 并馬上返回. 而在一旁待命的worker看到該task后, 便會(huì)按照設(shè)定執(zhí)行它, 并將他從queue中移除. 而worker則會(huì)執(zhí)行以下代

碼:

import myapp.tasks.add

myapp.tasks.add(2, 2)

7. 關(guān)于import

這里需要注意的是, 在impprt task時(shí), 需要保持一致. 因?yàn)樵趫?zhí)行djcelery.setup_loader()時(shí), task是以INSTALLED_APPS中的app名,

加.tasks.function_name注冊(cè)的, 如果我們由于python path不同而使用不同的引用方式時(shí)(例如在tasks.py中使用from myproject.myapp.tasks import

add形式), Celery將無(wú)法得知這是同一task, 因此可能會(huì)引起奇怪的bug.

8. 測(cè)試

a. 啟動(dòng)worker

正如之前說到的, 我們需要worker來(lái)執(zhí)行task. 以下是在開發(fā)環(huán)境中的如何啟動(dòng)worker:

首先啟動(dòng)terminal, 如同開發(fā)django項(xiàng)目一樣, 激活virtualenv, 切換到django項(xiàng)目目錄. 然后啟動(dòng)django自帶web服務(wù)器: python manage.py runserver.

然后啟動(dòng)worker:

python manage.py celery worker --loglevel=info

此時(shí), worker將會(huì)在該terminal中運(yùn)行, 并顯示輸出結(jié)果.

b. 啟動(dòng)task

打開新的terminal, 激活virtualenv, 并切換到django項(xiàng)目目錄:

$ python manage.py shell

from myapp.tasks import add

add.delay(2, 2)

此時(shí), 你可以在worker窗口中看到worker執(zhí)行該task:

[2014-10-07 08:47:08,076: INFO/MainProcess] Got task from broker: myapp.tasks.add[e080e047-b2a2-43a7-af74-d7d9d98b02fc]

[2014-10-07 08:47:08,299: INFO/MainProcess] Task myapp.tasks.add[e080e047-b2a2-43a7-af74-d7d9d98b02fc] succeeded in 0.183349132538s: 4

9. 另一個(gè)例子

下面我們來(lái)看一個(gè)更為真實(shí)的例子, 在views.py和tasks.py中:

# views.py

from myapp.tasks import do_something_with_form_data

def view(request):

form = SomeForm(request.POST)

if form.is_valid():

data = form.cleaned_data

# Schedule a task to process the data later

do_something_with_form_data.delay(data)

return render_to_response(...)

# tasks.py

@task

def do_something_with_form_data(data):

call_slow_web_service(data['user'], data['text'], ...)

10. 調(diào)試

由于Celery的運(yùn)行需要啟動(dòng)多個(gè)部件, 我們可能會(huì)漏掉一兩個(gè). 所以我們建議:

使用最簡(jiǎn)單的設(shè)置

使用python debug和logging功能顯示當(dāng)前的進(jìn)程

11. Eager模式

如果在settings.py設(shè)置:

CELERY_ALWAYS_EAGER = True

那么Celery便以eager模式運(yùn)行, 則task便不需要加delay運(yùn)行:

# 若啟用eager模式, 則以下兩行代碼相同

add.delay(2, 2)

add(2, 2)

12. 查看queue

因?yàn)槲覀兪褂昧薲jango作為broker, queue儲(chǔ)存在django的數(shù)據(jù)庫(kù)中. 這就意味著我們可以通過django admin查看該queue:

# admin.py

from django.contrib import admin

from kombu.transport.django import models as kombu_models

admin.site.register(kombu_models.Message)

13. 檢查結(jié)果

每次運(yùn)行異步task后, Celery都會(huì)返回AsyncResult對(duì)象作為結(jié)果. 你可以將其保存, 然后在將來(lái)查看該task是否運(yùn)行成功和返回結(jié)果:

# views.py

result = add.delay(2, 2)

...

if result.ready():

print "Task has run"

if result.successful():

print "Result was: %s" % result.result

else:

if isinstance(result.result, Exception):

print "Task failed due to raising an exception"

raise result.result

else:

print "Task failed without raising exception"

else:

print "Task has not yet run"

14. 定期任務(wù)

還有一種Celery的常用模式便是執(zhí)行定期任務(wù). 執(zhí)行定期任務(wù)時(shí), Celery會(huì)通過celerybeat進(jìn)程來(lái)完成. Celerybeat會(huì)保持運(yùn)行, 一旦到了某一定期任務(wù)需要執(zhí)

行時(shí), Celerybeat便將其加入到queue中. 不像worker進(jìn)程, Celerybeat只有需要一個(gè)即可.

啟動(dòng)Celerybeat:

python manage.py celery beat

使Celery運(yùn)行定期任務(wù)的方式有很多種, 我們先看第一種, 將定期任務(wù)儲(chǔ)存在django數(shù)據(jù)庫(kù)中. 即使是在django和celery都運(yùn)行的狀態(tài), 這一方式也可以讓我們

方便的修改定期任務(wù). 我們只需要設(shè)置settings.py中的一項(xiàng)便能開啟這一方式:

# settings.py

CELERYBEAT_SCHEDULER = 'djcelery.schedulers.DatabaseScheduler'

python延時(shí)循環(huán)執(zhí)行導(dǎo)致cpu消耗多過能解決么

可以不用一直保持此程序運(yùn)行啊，

Linux的話可以用低消耗的crontab 來(lái)自定義時(shí)間執(zhí)行此python命令即可。

提供個(gè)思路，可以使用python的apscheduler庫(kù)。

python多線程延遲并發(fā)

python多線程延遲并發(fā)的解決方法如下：

1.python之中多線程的特點(diǎn)，實(shí)際上是將執(zhí)行耗時(shí)長(zhǎng)的任務(wù)放在前臺(tái)，耗時(shí)短的任務(wù)放在后臺(tái)，當(dāng)處理器有空閑時(shí)或者是后臺(tái)任務(wù)主動(dòng)調(diào)用時(shí)就會(huì)將其拿到前臺(tái)來(lái)執(zhí)行，而在這個(gè)過程之中實(shí)際上每次還是執(zhí)行的一個(gè)線程。

2.python多線程延遲并發(fā)指的則是當(dāng)前python程序內(nèi)有多個(gè)程序，也就是任務(wù)同時(shí)處于已啟動(dòng)運(yùn)行到運(yùn)行完畢之間，且這幾個(gè)程序都是在同一個(gè)處理機(jī)上運(yùn)行，但任一個(gè)時(shí)刻點(diǎn)上只有一個(gè)程序在處理機(jī)上運(yùn)行。

3.python多線程延遲并發(fā)的好處就在于可以更加合理的去調(diào)配資源，因?yàn)槎嗑€程是使用CPU的多核處理器去完成任務(wù)的。而并發(fā)則是在同一處理器上完成任務(wù)，多線程實(shí)現(xiàn)并發(fā)的話就可以提高運(yùn)行速度并且減少內(nèi)存占用。

python如何微秒級(jí)延時(shí)？

Python中的sleep函數(shù)可以傳小數(shù)進(jìn)去，就可以進(jìn)行毫秒級(jí)的延時(shí)了，代碼如下：

#?例1：循環(huán)輸出休眠1秒

import?time

i?=?1

while?i?lt;=?3:

print?i?#?輸出i

i?+=?1

time.sleep(1)?#?休眠1秒

#?例2：循環(huán)輸出休眠100毫秒

import?time

i?=?1

while?i?lt;=?3:

print?i?#?輸出i

i?+=?1

time.sleep(0.1)?#?休眠0.1秒

如何用 Python 實(shí)現(xiàn)一個(gè)圖數(shù)據(jù)庫(kù)（Graph Database）？

本文章是 重寫 500 Lines or Less 系列的其中一篇，目標(biāo)是重寫 500 Lines or Less 系列的原有項(xiàng)目：Dagoba: an in-memory graph database。

Dagoba 是作者設(shè)計(jì)用來(lái)展示如何從零開始自己實(shí)現(xiàn)一個(gè)圖數(shù)據(jù)庫(kù)（ Graph Database ）。該名字似乎來(lái)源于作者喜歡的一個(gè)樂隊(duì)，另一個(gè)原因是它的前綴 DAG 也正好是有向無(wú)環(huán)圖 （ Directed Acyclic Graph ） 的縮寫。本文也沿用了該名稱。

圖是一種常見的數(shù)據(jù)結(jié)構(gòu)，它將信息描述為若干獨(dú)立的節(jié)點(diǎn)（ vertex ，為了和下文的邊更加對(duì)稱，本文中稱為 node ），以及把節(jié)點(diǎn)關(guān)聯(lián)起來(lái)的邊（ edge ）。我們熟悉的鏈表以及多種樹結(jié)構(gòu)可以看作是符合特定規(guī)則的圖。圖在路徑選擇、推薦算法以及神經(jīng)網(wǎng)絡(luò)等方面都是重要的核心數(shù)據(jù)結(jié)構(gòu)。

既然圖的用途如此廣泛，一個(gè)重要的問題就是如何存儲(chǔ)它。如果在傳統(tǒng)的關(guān)系數(shù)據(jù)庫(kù)中存儲(chǔ)圖，很自然的做法就是為節(jié)點(diǎn)和邊各自創(chuàng)建一張表，并用外鍵把它們關(guān)聯(lián)起來(lái)。這樣的話，要查找某人所有的子女，就可以寫下類似下面的查詢：

還好，不算太復(fù)雜。但是如果要查找孫輩呢？那恐怕就要使用子查詢或者 CTE(Common Table Expression) 等特殊構(gòu)造了。再往下想，曾孫輩又該怎么查詢？孫媳婦呢？

這樣我們會(huì)意識(shí)到，SQL 作為查詢語(yǔ)言，它只是對(duì)二維數(shù)據(jù)表這種結(jié)構(gòu)而設(shè)計(jì)的，用它去查詢圖的話非常笨拙，很快會(huì)變得極其復(fù)雜，也難以擴(kuò)展。針對(duì)圖而言，我們希望有一種更為自然和直觀的查詢語(yǔ)法，類似這樣：

為了高效地存儲(chǔ)和查詢圖這種數(shù)據(jù)結(jié)構(gòu)，圖數(shù)據(jù)庫(kù)（ Graph Database ）應(yīng)運(yùn)而生。因?yàn)楹蛡鹘y(tǒng)的關(guān)系型數(shù)據(jù)庫(kù)存在極大的差異，所以它屬于新型數(shù)據(jù)庫(kù)也就是 NoSql 的一個(gè)分支（其他分支包括文檔數(shù)據(jù)庫(kù)、列數(shù)據(jù)庫(kù)等）。圖數(shù)據(jù)庫(kù)的主要代表包括 Neo4J 等。本文介紹的 Dagoba 則是具備圖數(shù)據(jù)庫(kù)核心功能、主要用于教學(xué)和演示的一個(gè)簡(jiǎn)單的圖數(shù)據(jù)庫(kù)。

原文代碼是使用 JavaScript 編寫的，在定義調(diào)用接口時(shí)大量使用了原型（ prototype ）這種特有的語(yǔ)言構(gòu)造。對(duì)于其他主流語(yǔ)言的用戶來(lái)說，原型的用法多少顯得有些別扭和不自然。

考慮到本系列其他數(shù)據(jù)庫(kù)示例大多是用 Python 實(shí)現(xiàn)的，本文也按照傳統(tǒng)，用 Python 重寫了原文的代碼。同樣延續(xù)之前的慣例，為了讓讀者更好地理解程序是如何逐步完善的，我們用迭代式的方法完成程序的各個(gè)組成部分。

原文在 500lines 系列的 Github 倉(cāng)庫(kù)中只包含了實(shí)現(xiàn)代碼，并未包含測(cè)試。按照代碼注釋說明，測(cè)試程序位于作者的另一個(gè)代碼庫(kù)中，不過和 500lines 版本的實(shí)現(xiàn)似乎略有不同。

本文實(shí)現(xiàn)的代碼參考了原作者的測(cè)試內(nèi)容，但跳過了北歐神話這個(gè)例子——我承認(rèn)確實(shí)不熟悉這些神祇之間的親緣關(guān)系，相信中文背景的讀者們多數(shù)也未必了解，雖然作者很喜歡這個(gè)例子，想了想還是不要徒增困惑吧。因此本文在編寫測(cè)試用例時(shí)只參考了原文關(guān)于家族親屬的例子，放棄了神話相關(guān)的部分，盡管會(huì)減少一些趣味性，相信對(duì)于入門級(jí)的代碼來(lái)說這樣也夠用了。

本文實(shí)現(xiàn)程序位于代碼庫(kù)的 dagoba 目錄下。按照本系列程序的同意規(guī)則，要想直接執(zhí)行各個(gè)已完成的步驟，讀者可以在根目錄下的 main.py 找到相應(yīng)的代碼位置，取消注釋并運(yùn)行即可。

本程序的所有步驟只需要 Python3 ，測(cè)試則使用內(nèi)置的 unittest , 不需要額外的第三方庫(kù)。原則上 Python3.6 以上版本應(yīng)該都可運(yùn)行，但我只在 Python3.8.3 環(huán)境下完整測(cè)試過。

本文實(shí)現(xiàn)的程序從最簡(jiǎn)單的案例開始，通過每個(gè)步驟逐步擴(kuò)展，最終形成一個(gè)完整的程序。這些步驟包括：

接下來(lái)依次介紹各個(gè)步驟。

回想一下，圖數(shù)據(jù)庫(kù)就是一些點(diǎn)（ node ）和邊（ edge ）的集合?，F(xiàn)在我們要做出的一個(gè)重大決策是如何對(duì)節(jié)點(diǎn)/邊進(jìn)行建模。對(duì)于邊來(lái)說，必須指定它的關(guān)聯(lián)關(guān)系，也就是從哪個(gè)節(jié)點(diǎn)指向哪個(gè)節(jié)點(diǎn)。大多數(shù)情況下邊是有方向的——父子關(guān)系不指明方向可是要亂套的！

考慮到擴(kuò)展性及通用性問題，我們可以把數(shù)據(jù)保存為字典（ dict ），這樣可以方便地添加用戶需要的任何數(shù)據(jù)。某些數(shù)據(jù)是為數(shù)據(jù)庫(kù)內(nèi)部管理而保留的，為了明確區(qū)分，可以這樣約定：以下劃線開頭的特殊字段由數(shù)據(jù)庫(kù)內(nèi)部維護(hù)，類似于私有成員，用戶不應(yīng)該自己去修改它們。這也是 Python 社區(qū)普遍遵循的約定。

此外，節(jié)點(diǎn)和邊存在互相引用的關(guān)系。目前我們知道邊會(huì)引用到兩端的節(jié)點(diǎn)，后面還會(huì)看到，為了提高效率，節(jié)點(diǎn)也會(huì)引用到邊。如果僅僅在內(nèi)存中維護(hù)它們的關(guān)系，那么使用指針訪問是很直觀的，但數(shù)據(jù)庫(kù)必須考慮到序列化到磁盤的問題，這時(shí)指針就不再好用了。

為此，最好按照數(shù)據(jù)庫(kù)的一般要求，為每個(gè)節(jié)點(diǎn)維護(hù)一個(gè)主鍵（ _id ），用主鍵來(lái)描述它們之間的關(guān)聯(lián)關(guān)系。

我們第一步要把數(shù)據(jù)庫(kù)的模型建立起來(lái)。為了測(cè)試目的，我們使用一個(gè)最簡(jiǎn)單的數(shù)據(jù)庫(kù)模型，它只包含兩個(gè)節(jié)點(diǎn)和一條邊，如下所示：

按照 TDD 的原則，首先編寫測(cè)試：

與原文一樣，我們把數(shù)據(jù)庫(kù)管理接口命名為 Dagoba 。目前，能夠想到的最簡(jiǎn)單的測(cè)試是確認(rèn)節(jié)點(diǎn)和邊是否已經(jīng)添加到數(shù)據(jù)庫(kù)中：

assert_item 是一個(gè)輔助方法，用于檢查字典是否包含預(yù)期的字段。相信大家都能想到該如何實(shí)現(xiàn)，這里就不再列出了，讀者可參考 Github 上的完整源碼。

現(xiàn)在，測(cè)試是失敗的。用最簡(jiǎn)單的辦法實(shí)現(xiàn)數(shù)據(jù)庫(kù)：

需要注意的是，不管添加節(jié)點(diǎn)還是查詢，程序都使用了拷貝后的數(shù)據(jù)副本，而不是直接使用原始數(shù)據(jù)。為什么要這樣做？因?yàn)樽值涫强勺兊模脩艨梢栽谌魏螘r(shí)候修改其中的內(nèi)容，如果數(shù)據(jù)庫(kù)不知道數(shù)據(jù)已經(jīng)變化，就很容易發(fā)生難以追蹤的一致性問題，最糟糕的情況下會(huì)使得數(shù)據(jù)內(nèi)容徹底混亂。

拷貝數(shù)據(jù)可以避免上述問題，代價(jià)則是需要占用更多內(nèi)存和處理時(shí)間。對(duì)于數(shù)據(jù)庫(kù)來(lái)說，通常查詢次數(shù)要遠(yuǎn)遠(yuǎn)多于修改，所以這個(gè)代價(jià)是可以接受的。

現(xiàn)在測(cè)試應(yīng)該正常通過了。為了讓它更加完善，我們可以再測(cè)試一些邊緣情況，看看數(shù)據(jù)庫(kù)能否正確處理異常數(shù)據(jù)，比如：

例如，如果用戶嘗試添加重復(fù)主鍵，我們預(yù)期應(yīng)拋出 ValueError 異常。因此編寫測(cè)試如下：

為了滿足以上測(cè)試，代碼需要稍作修改。特別是按照 id 查找主鍵是個(gè)常用操作，通過遍歷的方法效率太低了，最好是能夠通過主鍵直接訪問。因此在數(shù)據(jù)庫(kù)中再增加一個(gè)字典：

完整代碼請(qǐng)參考 Github 倉(cāng)庫(kù)。

在上個(gè)步驟，我們?cè)诔跏蓟瘮?shù)據(jù)庫(kù)時(shí)為節(jié)點(diǎn)明確指定了主鍵。按照數(shù)據(jù)庫(kù)設(shè)計(jì)的一般原則，主鍵最好是不具有業(yè)務(wù)含義的代理主鍵（ Surrogate key ），用戶不應(yīng)該關(guān)心它具體的值是什么，因此讓數(shù)據(jù)庫(kù)去管理主鍵通常是更為合理的。當(dāng)然，在部分場(chǎng)景下——比如導(dǎo)入外部數(shù)據(jù)——明確指定主鍵仍然是有用的。

為了同時(shí)支持這些要求，我們這樣約定：字段 _id 表示節(jié)點(diǎn)的主鍵，如果用戶指定了該字段，則使用用戶設(shè)置的值（當(dāng)然，用戶有責(zé)任保證它們不會(huì)重復(fù)）；否則，由數(shù)據(jù)庫(kù)自動(dòng)為它分配一個(gè)主鍵。

如果主鍵是數(shù)據(jù)庫(kù)生成的，事先無(wú)法預(yù)知它的值是什么，而邊（ edge ）必須指定它所指向的節(jié)點(diǎn)，因此必須在主鍵生成后才能添加。由于這個(gè)原因，在動(dòng)態(tài)生成主鍵的情況下，數(shù)據(jù)庫(kù)的初始化會(huì)略微復(fù)雜一些。還是先寫一個(gè)測(cè)試：

為支持此功能，我們?cè)跀?shù)據(jù)庫(kù)中添加一個(gè)內(nèi)部字段 _next_id 用于生成主鍵，并讓 add_node 方法返回新生成的主鍵：

接下來(lái)，再確認(rèn)一下邊是否可以正常訪問：

運(yùn)行測(cè)試，一切正常。這個(gè)步驟很輕松地完成了，不過兩個(gè)測(cè)試（ DbModelTest 和 PrimaryKeyTest ）出現(xiàn)了一些重復(fù)代碼，比如 get_item 。我們可以把這些公用代碼提取出來(lái)。由于 get_item 內(nèi)部調(diào)用了 TestCase.assertXXX 等方法，看起來(lái)應(yīng)該使用繼承，但從 TestCase 派生基類容易引起一些潛在的問題，所以我轉(zhuǎn)而使用另一個(gè)技巧 Mixin ：

實(shí)現(xiàn)數(shù)據(jù)庫(kù)模型之后，接下來(lái)就要考慮如何查詢它了。

在設(shè)計(jì)查詢時(shí)要考慮幾個(gè)問題。對(duì)于圖的訪問來(lái)說，幾乎總是由某個(gè)節(jié)點(diǎn)（或符合條件的某一類節(jié)點(diǎn)）開始，從與它相鄰的邊跳轉(zhuǎn)到其他節(jié)點(diǎn)，依次類推。所以鏈?zhǔn)秸{(diào)用對(duì)查詢來(lái)說是一種很自然的風(fēng)格。舉例來(lái)說，要知道 Tom 的孫子養(yǎng)了幾只貓，可以使用類似這樣的查詢：

可以想象，以上每個(gè)方法都應(yīng)該返回符合條件的節(jié)點(diǎn)集合。這種實(shí)現(xiàn)是很直觀的，不過存在一個(gè)潛在的問題：很多時(shí)候用戶只需要一小部分結(jié)果，如果它總是不計(jì)代價(jià)地給我們一個(gè)巨大的集合，會(huì)造成極大的浪費(fèi)。比如以下查詢：

為了避免不必要的浪費(fèi)，我們需要另外一種機(jī)制，也就是通常所稱的“懶式查詢”或“延遲查詢”。它的基本思想是，當(dāng)我們調(diào)用查詢方法時(shí)，它只是把查詢條件記錄下來(lái)，而并不立即返回結(jié)果，直到明確調(diào)用某些方法時(shí)才真正去查詢數(shù)據(jù)庫(kù)。

如果讀者比較熟悉流行的 Python ORM，比如 SqlAlchemy 或者 Django ORM 的話，會(huì)知道它們幾乎都是懶式查詢的，要調(diào)用 list(result) 或者 result[0:10] 這樣的方法才能得到具體的查詢結(jié)果。

在 Dagoba 中把觸發(fā)查詢的方法定義為 run 。也就是說，以下查詢執(zhí)行到 run 時(shí)才真正去查找數(shù)據(jù)：

和懶式查詢（ Lazy Query ）相對(duì)應(yīng)的，直接返回結(jié)果的方法一般稱作主動(dòng)查詢（ Eager Query ）。主動(dòng)查詢和懶式查詢的內(nèi)在查找邏輯基本上是相同的，區(qū)別只在于觸發(fā)機(jī)制不同。由于主動(dòng)查詢實(shí)現(xiàn)起來(lái)更加簡(jiǎn)單，出錯(cuò)也更容易排查，因此我們先從主動(dòng)查詢開始實(shí)現(xiàn)。

還是從測(cè)試開始。前面測(cè)試所用的簡(jiǎn)單數(shù)據(jù)庫(kù)數(shù)據(jù)太少，難以滿足查詢要求，所以這一步先來(lái)創(chuàng)建一個(gè)更復(fù)雜的數(shù)據(jù)模型：

此關(guān)系的復(fù)雜之處之一在于反向關(guān)聯(lián)：如果 A 是 B 的哥哥，那么 B 就是 A 的弟弟/妹妹，為了查詢到他們彼此之間的關(guān)系，正向關(guān)聯(lián)和反向關(guān)聯(lián)都需要存在，因此在初始化數(shù)據(jù)庫(kù)時(shí)需要定義的邊數(shù)量會(huì)很多。

當(dāng)然，父子之間也存在反向關(guān)聯(lián)的問題，為了讓問題稍微簡(jiǎn)化一些，我們目前只需要向下（子孫輩）查找，可以稍微減少一些關(guān)聯(lián)數(shù)量。

因此，我們定義數(shù)據(jù)模型如下。為了減少重復(fù)工作，我們通過 _backward 字段定義反向關(guān)聯(lián)，而數(shù)據(jù)庫(kù)內(nèi)部為了查詢方便，需要把它維護(hù)成兩條邊：

然后，測(cè)試一個(gè)最簡(jiǎn)單的查詢，比如查找某人的所有孫輩：

這里 outcome/income 分別表示從某個(gè)節(jié)點(diǎn)出發(fā)、或到達(dá)它的節(jié)點(diǎn)集合。在原作者的代碼中把上述方法稱為 out/in 。當(dāng)然這樣看起來(lái)更加簡(jiǎn)潔，可惜的是 in 在 Python 中是個(gè)關(guān)鍵字，無(wú)法作為函數(shù)名。我也考慮過加個(gè)下劃線比如 out_.in_ 這種形式，但看起來(lái)也有點(diǎn)怪異，權(quán)衡之后還是使用了稍微啰嗦一點(diǎn)的名稱。

現(xiàn)在我們可以開始定義查詢接口了。在前面已經(jīng)說過，我們計(jì)劃分別實(shí)現(xiàn)兩種查詢，包括主動(dòng)查詢（ Eager Query ）以及延遲查詢（ Lazy Query ）。

它們的內(nèi)在查詢邏輯是相通的，看起來(lái)似乎可以使用繼承。不過遵循 YAGNI 原則，目前先不這樣做，而是只定義兩個(gè)新類，在滿足測(cè)試的基礎(chǔ)上不斷擴(kuò)展。以后我們會(huì)看到，與繼承相比，把共同的邏輯放到數(shù)據(jù)庫(kù)本身其實(shí)是更為合理的。

接下來(lái)實(shí)現(xiàn)訪問節(jié)點(diǎn)的方法。由于 EagerQuery 調(diào)用查詢方法會(huì)立即返回結(jié)果，我們把結(jié)果記錄在 _result 內(nèi)部字段中。雖然 node 方法只返回單個(gè)結(jié)果，但考慮到其他查詢方法幾乎都是返回集合，為統(tǒng)一起見，讓它也返回集合，這樣可以避免同時(shí)支持集合與單結(jié)果的分支處理，讓代碼更加簡(jiǎn)潔、不容易出錯(cuò)。此外，如果查詢對(duì)象不存在的話，我們只返回空集合，并不視為一個(gè)錯(cuò)誤。

查詢輸入/輸出節(jié)點(diǎn)的方法實(shí)現(xiàn)類似這樣：

查找節(jié)點(diǎn)的核心邏輯在數(shù)據(jù)庫(kù)本身定義：

以上使用了內(nèi)部定義的一些輔助查詢方法。用類似的邏輯再定義 income ，它們的實(shí)現(xiàn)都很簡(jiǎn)單，讀者可以直接參考源碼，此處不再贅述。

在此步驟的最后，我們?cè)賹?shí)現(xiàn)一個(gè)優(yōu)化。當(dāng)多次調(diào)用查詢方法后，結(jié)果可能會(huì)返回重復(fù)的數(shù)據(jù)，很多時(shí)候這是不必要的。就像關(guān)系數(shù)據(jù)庫(kù)通常支持 unique/distinct 一樣，我們也希望 Dagoba 能夠過濾重復(fù)的數(shù)據(jù)。

假設(shè)我們要查詢某人所有孩子的祖父，顯然不管有多少孩子，他們的祖父應(yīng)該是同一個(gè)人。因此編寫測(cè)試如下：

現(xiàn)在來(lái)實(shí)現(xiàn) unique 。我們只要按照主鍵把重復(fù)數(shù)據(jù)去掉即可：

在上個(gè)步驟，初始化數(shù)據(jù)庫(kù)指定了雙向關(guān)聯(lián)，但并未測(cè)試它們。因?yàn)槲覀冞€沒有編寫代碼去支持它們，現(xiàn)在增加一個(gè)測(cè)試，它應(yīng)該是失敗的：

運(yùn)行測(cè)試，的確失敗了。我們看看要如何支持它?；叵胍幌?，當(dāng)從邊查找節(jié)點(diǎn)時(shí)，使用的是以下方法：

這里也有一個(gè)潛在的問題：調(diào)用 self.edges 意味著遍歷所有邊，當(dāng)數(shù)據(jù)庫(kù)內(nèi)容較多時(shí)，這是巨大的浪費(fèi)。為了提高性能，我們可以把與節(jié)點(diǎn)相關(guān)的邊記錄在節(jié)點(diǎn)本身，這樣要查找邊只要看節(jié)點(diǎn)本身即可。在初始化時(shí)定義出入邊的集合：

在添加邊時(shí)，我們要同時(shí)把它們對(duì)應(yīng)的關(guān)系同時(shí)更新到節(jié)點(diǎn)，此外還要維護(hù)反向關(guān)聯(lián)。這涉及對(duì)字典內(nèi)容的部分復(fù)制，先編寫一個(gè)輔助方法：

然后，將添加邊的實(shí)現(xiàn)修改如下：

這里的代碼同時(shí)添加正向關(guān)聯(lián)和反向關(guān)聯(lián)。有的朋友可能會(huì)注意到代碼略有重復(fù)，是的，但是重復(fù)僅出現(xiàn)在該函數(shù)內(nèi)部，本著“三則重構(gòu)”的原則，暫時(shí)不去提取代碼。

實(shí)現(xiàn)之后，前面的測(cè)試就可以正常通過了。

在這個(gè)步驟中，我們來(lái)實(shí)現(xiàn)延遲查詢（ Lazy Query ）。

延遲查詢的要求是，當(dāng)調(diào)用查詢方法時(shí)并不立即執(zhí)行，而是推遲到調(diào)用特定方法，比如 run 時(shí)才執(zhí)行整個(gè)查詢，返回結(jié)果。

延遲查詢的實(shí)現(xiàn)要比主動(dòng)查詢復(fù)雜一些。為了實(shí)現(xiàn)延遲查詢，查詢方法的實(shí)現(xiàn)不能直接返回結(jié)果，而是記錄要執(zhí)行的動(dòng)作以及傳入的參數(shù)，到調(diào)用 run 時(shí)再依次執(zhí)行前面記錄下來(lái)的內(nèi)容。

如果你去看作者的實(shí)現(xiàn)，會(huì)發(fā)現(xiàn)他是用一個(gè)數(shù)據(jù)結(jié)構(gòu)記錄執(zhí)行操作和參數(shù)，此外還有一部分邏輯用來(lái)分派對(duì)每種結(jié)構(gòu)要執(zhí)行的動(dòng)作。這樣當(dāng)然是可行的，但數(shù)據(jù)處理和分派部分的實(shí)現(xiàn)會(huì)比較復(fù)雜，也容易出錯(cuò)。

本文的實(shí)現(xiàn)則選擇了另外一種不同的方法：使用 Python 的內(nèi)部函數(shù)機(jī)制，把一連串查詢變換成一組函數(shù)，每個(gè)函數(shù)取上個(gè)函數(shù)的執(zhí)行結(jié)果作為輸入，最后一個(gè)函數(shù)的輸出就是整個(gè)查詢的結(jié)果。由于內(nèi)部函數(shù)同時(shí)也是閉包，盡管每個(gè)查詢的參數(shù)形式各不相同，但是它們都可以被閉包“捕獲”而成為內(nèi)部變量，所以這些內(nèi)部函數(shù)可以采用統(tǒng)一的形式，無(wú)需再針對(duì)每種查詢?cè)O(shè)計(jì)額外的數(shù)據(jù)結(jié)構(gòu)，因而執(zhí)行過程得到了很大程度的簡(jiǎn)化。

首先還是來(lái)編寫測(cè)試。 LazyQueryTest 和 EagerQueryTest 測(cè)試用例幾乎是完全相同的（是的，兩種查詢只在于內(nèi)部實(shí)現(xiàn)機(jī)制不同，它們的調(diào)用接口幾乎是完全一致的）。

因此我們可以把 EagerQueryTest 的測(cè)試原樣不變拷貝到 LazyQueryTest 中。當(dāng)然拷貝粘貼不是個(gè)好注意，對(duì)于比較冗長(zhǎng)而固定的初始化部分，我們可以把它提取出來(lái)作為兩個(gè)測(cè)試共享的公共函數(shù)。讀者可參考代碼中的 step04_lazy_query/tests/test_lazy_query.py 部分。

程序把查詢函數(shù)的串行執(zhí)行稱為管道（ pipeline ），用一個(gè)變量來(lái)記錄它：

然后依次實(shí)現(xiàn)各個(gè)調(diào)用接口。每種接口的實(shí)現(xiàn)都是類似的：用內(nèi)部函數(shù)執(zhí)行真正的查詢邏輯，再把這個(gè)函數(shù)添加到 pipeline 調(diào)用鏈中。比如 node 的實(shí)現(xiàn)類似下面：

其他接口的實(shí)現(xiàn)也與此類似。最后， run 函數(shù)負(fù)責(zé)執(zhí)行所有查詢，返回最終結(jié)果；

完成上述實(shí)現(xiàn)后執(zhí)行測(cè)試，確保我們的實(shí)現(xiàn)是正確的。

在前面我們說過，延遲查詢與主動(dòng)查詢相比，最大的優(yōu)勢(shì)是對(duì)于許多查詢可以按需要訪問，不需要每個(gè)步驟都返回完整結(jié)果，從而提高性能，節(jié)約查詢時(shí)間。比如說，對(duì)于下面的查詢：

以上查詢的意思是從孫輩中找到一個(gè)符合條件的節(jié)點(diǎn)即可。對(duì)該查詢而言，主動(dòng)查詢會(huì)在調(diào)用 outcome('son') 時(shí)就遍歷所有節(jié)點(diǎn)，哪怕最后一步只需要第一個(gè)結(jié)果。而延遲查詢?yōu)榱颂岣咝?，?yīng)在找到符合條件的結(jié)果后立即停止。

目前我們尚未實(shí)現(xiàn) take 方法。老規(guī)矩，先添加測(cè)試：

主動(dòng)查詢的 take 實(shí)現(xiàn)比較簡(jiǎn)單，我們只要從結(jié)果中返回前 n 條記錄：

延遲查詢的實(shí)現(xiàn)要復(fù)雜一些。為了避免不必要的查找，返回結(jié)果不應(yīng)該是完整的列表（ list ），而應(yīng)該是個(gè)按需返回的可迭代對(duì)象，我們用內(nèi)置函數(shù) next 來(lái)依次返回前 n 個(gè)結(jié)果：

寫完后運(yùn)行測(cè)試，確保它們是正確的。

從外部接口看，主動(dòng)查詢和延遲查詢幾乎是完全相同的，所以用單純的數(shù)據(jù)測(cè)試很難確認(rèn)后者的效率一定比前者高，用訪問時(shí)間來(lái)測(cè)試也并不可靠。為了測(cè)試效率，我們引入一個(gè)節(jié)點(diǎn)訪問次數(shù)的概念，如果延遲查詢效率更高的話，那么它應(yīng)該比主動(dòng)查詢?cè)L問節(jié)點(diǎn)的次數(shù)更少。

為此，編寫如下測(cè)試：

我們?yōu)? Dagoba 類添加一個(gè)成員來(lái)記錄總的節(jié)點(diǎn)訪問次數(shù)，以及兩個(gè)輔助方法，分別用于獲取和重置訪問次數(shù)：

然后瀏覽代碼，查找修改點(diǎn)。增加計(jì)數(shù)主要在從邊查找節(jié)點(diǎn)的時(shí)候，因此修改部分如下：

此外還有 income/outcome 方法，修改都很簡(jiǎn)單，這里就不再列出。

實(shí)現(xiàn)后再次運(yùn)行測(cè)試。測(cè)試通過，表明延遲查詢確實(shí)在效率上優(yōu)于主動(dòng)查詢。

不像關(guān)系數(shù)據(jù)庫(kù)的結(jié)構(gòu)那樣固定，圖的形式可以千變?nèi)f化，查詢機(jī)制也必須足夠靈活。從原理上講，所有查詢無(wú)非是從某個(gè)節(jié)點(diǎn)出發(fā)按照特定方向搜索，因此用 node/income/outcome 這三個(gè)方法幾乎可以組合出任意所需的查詢。

但對(duì)于復(fù)雜查詢，寫出的代碼有時(shí)會(huì)顯得較為瑣碎和冗長(zhǎng)，對(duì)于特定領(lǐng)域來(lái)說，往往存在更為簡(jiǎn)潔的名稱，例如：母親的兄弟可簡(jiǎn)稱為舅舅。對(duì)于這些場(chǎng)景，如果能夠類似 DSL （領(lǐng)域特定語(yǔ)言）那樣允許用戶根據(jù)專業(yè)要求自行擴(kuò)展，從而簡(jiǎn)化查詢，方便閱讀，無(wú)疑會(huì)更為友好。

如果讀者去看原作者的實(shí)現(xiàn)，會(huì)發(fā)現(xiàn)他是用一種特殊語(yǔ)法 addAlias 來(lái)定義自己想要的查詢，調(diào)用方法時(shí)再進(jìn)行查詢以確定要執(zhí)行的內(nèi)容，其接口和內(nèi)部實(shí)現(xiàn)都是相當(dāng)復(fù)雜的。

而我希望有更簡(jiǎn)單的方法來(lái)實(shí)現(xiàn)這一點(diǎn)。所幸 Python 是一種高度動(dòng)態(tài)的語(yǔ)言，允許在運(yùn)行時(shí)向類中增加新的成員，因此做到這一點(diǎn)可能比預(yù)想的還要簡(jiǎn)單。

為了驗(yàn)證這一點(diǎn)，編寫測(cè)試如下：

無(wú)需 Dagoba 的實(shí)現(xiàn)做任何改動(dòng)，測(cè)試就可以通過了！其實(shí)我們要做的就是動(dòng)態(tài)添加一個(gè)自定義的成員函數(shù)，按照 Python 對(duì)象機(jī)制的要求，成員函數(shù)的第一個(gè)成員應(yīng)該是名為 self 的參數(shù)，但這里已經(jīng)是在 UnitTest 的內(nèi)部，為了和測(cè)試類本身的 self 相區(qū)分，新函數(shù)的參數(shù)增加了一個(gè)下劃線。

此外，函數(shù)應(yīng)返回其所屬的對(duì)象，這是為了鏈?zhǔn)秸{(diào)用所要求的。我們看到，動(dòng)態(tài)語(yǔ)言的靈活性使得添加新語(yǔ)法變得非常簡(jiǎn)單。

到此，一個(gè)初具規(guī)模的圖數(shù)據(jù)庫(kù)就形成了。

和原文相比，本文還缺少一些內(nèi)容，比如如何將數(shù)據(jù)庫(kù)序列化到磁盤。不過相信讀者都看到了，我們的數(shù)據(jù)庫(kù)內(nèi)部結(jié)構(gòu)基本上是簡(jiǎn)單的原生數(shù)據(jù)結(jié)構(gòu)（列表+字典），因此序列化無(wú)論用 pickle 或是 JSON 之類方法都應(yīng)該是相當(dāng)簡(jiǎn)單的。有興趣的讀者可以自行完成它們。

我們的圖數(shù)據(jù)庫(kù)實(shí)現(xiàn)為了提高查詢性能，在節(jié)點(diǎn)內(nèi)部存儲(chǔ)了邊的指針（或者說引用）。這樣做的好處是，無(wú)論數(shù)據(jù)庫(kù)有多大，從一個(gè)節(jié)點(diǎn)到相鄰節(jié)點(diǎn)的訪問是常數(shù)時(shí)間，因此數(shù)據(jù)訪問的效率非常高。

但一個(gè)潛在的問題是，如果數(shù)據(jù)庫(kù)規(guī)模非常大，已經(jīng)無(wú)法整個(gè)放在內(nèi)存中，或者出于安全性等原因要實(shí)現(xiàn)分布式訪問的話，那么指針就無(wú)法使用了，必須要考慮其他機(jī)制來(lái)解決這個(gè)問題。分布式數(shù)據(jù)庫(kù)無(wú)論采用何種數(shù)據(jù)模型都是一個(gè)棘手的問題，在本文中我們沒有涉及。有興趣的讀者也可以考慮 500lines 系列中關(guān)于分布式和集群算法的其他一些文章。

本文的實(shí)現(xiàn)和系列中其他數(shù)據(jù)庫(kù)類似，采用 Python 作為實(shí)現(xiàn)語(yǔ)言，而原作者使用的是 JavaScript ，這應(yīng)該和作者的背景有關(guān)。我相信對(duì)于大多數(shù)開發(fā)者來(lái)說， Python 的對(duì)象機(jī)制比 JavaScript 基于原型的語(yǔ)法應(yīng)該是更容易閱讀和理解的。

當(dāng)然，原作者的版本比本文版本在實(shí)現(xiàn)上其實(shí)是更為完善的，靈活性也更好。如果想要更為優(yōu)雅的實(shí)現(xiàn)，我們可以考慮使用 Python 元編程，那樣會(huì)更接近于作者的實(shí)現(xiàn)，但也會(huì)讓程序的復(fù)雜性大為增加。如果讀者有興趣，不妨對(duì)照著去讀讀原作者的版本。