Django 通道簡要介紹

編譯：王堅

www.codeceo.com/article/django-channels.html

通道是 Django 即將支持的令人興奮的特性，它將使得 Django 不止支持普通請求外部工具和庫（即便不是 Python 的），還可能是整個框架。

根據官方文檔，Django 通道是：

…一個讓 Django 不僅可以處理純 HTTP 請求，包括 HTTP2 和WebSockets，也有能力在請求發送到縮略圖或後台運算時也能運行運行代碼。

如果你以前用過 Django，你就知道 project 是多重要。目前 Django 的諸多特性依賴於庫，比如 Celery（在請求之外處理複雜任務），或者 Node.js，dijango-websocket-redis，或者 gevent-stocketio 來支持 WebSocket。

由於 Celery 的原因（它是一個事實標準），其他所有的實現方法以非標準的形式在 Django 的局限內有各自的問題。我們在以往的博文里提到了成功實現的不同方式。

一個合乎標準的方式更容易維護，更安全，多數開發者熟悉其內容也更容易交接。

在這篇博客中我會快速的介紹開發應用 Django 通道網站所涉及的概念，同時介紹一個用 WebSocket 給客戶端推送通知的例子。

應用

我們舉的例子是對用 gevent-socketio 實現博客實時通知應用的修改。目的是讓你看到用 Django 通道在同樣的條件下實施會有多簡單，代碼可以在GitHub上找到。

面向事件的Diango

默認的 Django 網站請求-響應模型：一個請求進來，被傳遞給視圖，視圖產生一個回應，然後回應被發送到客戶端，所有一切都是單線程完成的。

在大多數應用中都完全勝任，但它有自己的局限。如果是多個請求就會讓工作進程持續好長時間，後續的請求要排隊等待。這就是用 Celery 來做縮略圖之類事情的原因：在圖片上傳時，我們建立縮略圖任務並及時響應到客戶端，在此同時 Celery 在自己的進程中處理圖片。

同樣的情況會發生在和客戶端的實時雙向對話中。設想一個請求-響應，我們需要一個進程對一個客戶端來收發消息直到連接終止。

Django 通道提供了另一個模型：面向事件的模型。在這個模型中，事件取代了請求和響應。一個請求事件被接收到會被傳遞給合適的處理者來產生一個新的響應事件被傳回到客戶端。

事件模型可以應用到其他情況而不只是對請求-響應模型的模仿。比如由外界條件觸發的感測器，它產生一個事件給事件處理者，然後會依次產生另一個事件通知所有對原始事件感興趣的人。

但是這個進程怎麼工作的？我們需要在開始實例前認識下channel。

什麼是通道

根據 Django 通道的官方文檔，通道是：

…一個有序的，先進先出的消息隊列，通常一次只有一個消息接收者。

多個生產者將消息寫入 channel（用一個名字來識別），然後一個用戶訂閱了那麼 channel 就可用，它會取出隊列中的第一條消息。就是這麼簡單。

通道改變了Django的工作方式，讓它像worker一樣工作。每個worker聽從通道上所有用戶的吩咐當有消息是用戶會被通知。要想這事發生，我們需要三個層：

介面伺服器：連接網站與客戶端，通過一個 WSGIn接頭和一個獨立的WebSocket伺服器。

通道後端：它在介面和worker間傳遞消息。（為單一伺服器提供存儲，一個資料庫或者Redis），Python 代碼都在這裡。

worker：它們收聽所有的通道，當消息來時喚醒用戶（函數）。

介面伺服器把連接（HTTP，WebSocket等）轉換成通道中的消息，worker負責處理這些消息。這裡的門道在於消息不需要從介面伺服器產生。消息可以在任何地方產生，view，form，signal，隨你心意。

是時候幹活了。

我們的第一個用戶

我們從安裝Django1.8（1.9也行）開始。首先，需要安裝安裝 channel 包，它是依賴 PyPi 的。如果你想安裝最新版本的通道，看看官方介紹文檔。

接下來把 channel 加入到 INSTALLED_APPS 設置中：

INSTALLED_APPS=(

...

channels_test ,# Our test app

channels ,

)

就是這樣。通道默認配置使用在內存中的後端，它能很好的在單個伺服器的網站進行工作。

我們將要寫一個簡單的用戶來接收「http.message」通道上的消息，然後回應通道一個新消息。讓我們在測試 Django 應用這中建立一個模組叫「consumer.py」：

# consumers.py

from json import dumps

fromdjango.http import HttpResponse

def http_consumer(message):

response=HttpResponse(

"It is now {} and you ve requested {} with {} as request parameters.".format(

now(),

message.content[ path ],

dumps(message.content[ get ])

)

)

message.reply_channel.send(response.channel_encode())

消息從標準「reques.http」通道中出來，用戶寫一個新消息在回應通道中回應。值得注意的是，他們是兩種不同的通道：普通通道傳遞消息給用戶，和回應通道。只用介面伺服器偵聽回應通道，它知道那個通道連接那個客戶端，所以他知道回應該發給誰。

在開始進程前，我們需要一個方法來告訴 Django 將「request.http」通道消息發送給我們的新用戶。在設置中繼續創建一個模組叫「routing.py」：

channel_routing={

}

現在我們運行伺服器（是開發伺服器或者 WSGI 伺服器無關緊要），給我們的網站一個請求：

$curlhttp://localhost:8000/some/path?foo=bar

Itisnow2016-02-0111:49:25.166799+00:00andyou verequested/some/pathwith{"foo":["bar"]}asrequestparameters.

我們得到了想要的請求，通道解決了我們大部分問題。現在讓我們玩點更有趣的。

實時通知

我們已經多次提及了實時通知的要點，這給了我們一個很好的機會來應用一個通道，並比較了兩個解決方案。

我們將改進現有的項目，追蹤用戶在特定地理位置發出有趣的實時通知。在這個項目中我們用到了 gevent-socketio], SocketIO, 和 RabbitMQ(還有 Node.js)。我們將用通道、普通WebSockets和Redis來做同樣的事情。

如前所述，我們用 WebStocket 推送通知到客戶端。通道對 WebStocket 有完整的支持。所有我們要做的不過是在我們的「tracker」應用中加上幾個通道：

# routing.py

channel_routing={

}

我們不在意「websocket.message」通道也不打算接收用戶消息。我們的目標是向所有連接的客戶端發出推送消息。用一個群來做這件事非常容易。讓我們看看用戶：

# tracker/consumers.py

import logging

from channels import Group

fromchannels.sessions import channel_session

fromchannels.auth import channel_session_user_from_http

logger=logging.getLogger(__name__)

# Connected to websocket.connect and websocket.keepalive

@channel_session_user_from_http

def websocket_connect(message):

logger.info( websocket_connect. message = %s ,message)

# transfer_user(message.http_session, message.channel_session)

Group("notifications").add(message.reply_channel)

# Connected to websocket.keepalive

@channel_session

def websocket_keepalive(message):

logger.info( websocket_keepalive. message = %s ,message)

Group("notifications").add(message.reply_channel)

# Connected to websocket.disconnect

@channel_session

def websocket_disconnect(message):

logger.info( websocket_disconnect. message = %s ,message)

Group("notifications").discard(message.reply_channel)

無論何時一個客戶端連接，就會有一條消息通過「websocket.connect」發送，然後我們需要加上一個回應通道，還要把它放進「notifications」群。群允許我們同時發送相同的消息到所有的通道。我們要保持通道群的更新，即當客戶端連接，我們將回應通道就啊如群；斷開連接，將它移除。群也會在一定時間後清理通道，我們用「websocket.keepalive」通道把接收到keepalive消息的通道加入到「notifications」群。如果通道已存在不會被再次加入。

注意我們沒有向群里發送任何東西。我們要在 AreaOfInterest 中的 Incident 被報告或更新時通知用戶。我們簡單的加入 post_save 信號：

# signals.py

import logging

from json import dumps

fromdjango.dispatch import receiver

from channels import Group

from.models importIncident,AreaOfInterest

logger=logging.getLogger(__name__)

def send_notification(notification):

logger.info( send_notification. notification = %s ,notification)

Group("notifications").send({ text :dumps(notification)})

@receiver(post_save,sender=Incident)

def incident_post_save(sender,**kwargs):

send_notification({

type : post_save ,

created :kwargs[ created ],

feature :kwargs[ instance ].geojson_feature

})

ifnotkwargs[ instance ].closed:

areas_of_interest=[

polygon__contains=kwargs[ instance ].location,

severity__in=kwargs[ instance ].alert_severities,

)

]

ifareas_of_interest:

send_notification(dict(

type= alert ,

feature=kwargs[ instance ].geojson_feature,

areas_of_interest=[

{

id :area_of_interest[ id ],

name :area_of_interest[ properties ][ name ],

severity :area_of_interest[ properties ][ severity ],

url :area_of_interest[ properties ][ url ],

}

forarea_of_interestinareas_of_interest

]

))

@receiver(post_save,sender=AreaOfInterest)

def area_of_interest_post_save(sender,**kwargs):

send_notification({

type : post_save ,

created :kwargs[ created ],

feature :kwargs[ instance ].geojson_feature

})

@receiver(post_delete,sender=Incident)

@receiver(post_delete,sender=AreaOfInterest)

def post_delete(sender,**kwargs):

send_notification({

type : post_delete ,

feature :kwargs[ instance ].geojson_feature

})

所有消息都發生在 send_notification，如你所見只是一行代碼（回頭去看舊的實現方法）。其餘代碼都和以前的一樣。

至此，我們只用了內存通道後端，要使我們的統治系統在多伺服器環境中工作，我們需要使用資料庫後端或者 Redis 後端。讓我們用後者，必須在setting.py模組中加入以下代碼片段：

CHANNEL_LAYERS={

"default":{

"BACKEND":"asgi_redis.RedisChannelLayer",

"CONFIG":{

"hosts":[("localhost",6379)],

},

},

}

讓這個後端工作，我們還需要安裝 asgi_redis 包（內存和資料庫層包含在通道包）。最後一件要在 Django 這邊做的更該是建一個 asgi.py 模組，它的功能相當於WSGI伺服器的wsgi.py。

# asgi.py

import os

fromchannels.asgi import get_channel_layer

channel_layer=get_channel_layer()

這個模塊將會被用來運行介面伺服器。

我們還需要把客戶端代碼用 WebSocket 取代 Scoket.io，鑒於它們非常相似，我們不去涉及細節，你可以去這裡看看。

剩下的就是運行伺服器了，為了開發，我們可以用 runserver 管理命令。它已經修改用來運行 Daphne ASGI 伺服器和一個 worker：

(tracker_project_venv)$./manage.py runserver

Worker threadrunning,channels enabled

Performing systemchecks...

System check identified no issues(silenced).

February01,2016-13:19:43

Djangoversion1.8.8,usingsettings tracker_project.settings

Starting development server athttp://127.0.0.1:8000/

Quit the server withCONTROL-C.

在真實環境中，我們需要運行 Daphne 和足夠多的 worker。Daphne 這樣運行：

(tracker_project_venv)$ daphne tracker_project.asgi:channel_layer

我們在 asgi.py 模組告訴 Daphne 去尋找通道層。每個 worker 這樣運行：

(venv)$ python ./manage.py runworker

所有東西具備且運行，你就可以報告事件接收通知，如果視圖是打開的。

授權認證說明

我們沒有說到的是 WebSocket 是有授權認證的

WebSocket 的初始連接是普通的 HTTP 請求，所以我們可以利用 Django 現有的通過會話管理運用通道的@channel_session_user_from_http 裝飾器。如果用戶沒有得到授權（相當於沒有有效會話）便不能回復通道消息，也不會加入「notification」群。

我們要確保構建 WebSocket 是通過了會話密鑰：

var socket = new WebSocket( ws://localhost:8000?session_key= + sessionKey);

還不完善，但我們實現了。有更多的 WebSocket 授權方式（我偏愛給予令牌的系統，比如 JWT），但要接受這個需要另一篇文章了。

結論

Django 通道將會根本上改變我們工作的方式。它讓我們生活更容易，使我們能用 Django 網站解決更廣泛地問題，但這個項目最大的特點在於：他是完全易於操作的。在我們的例子中，我們只是引入了我們喜愛的通道而已。我們不需要改變給予請求-響應的原有代碼。我們得到了世上最好的。

通道還沒有為生產環境準備好，但他一旦被集成到下一版本的 Django 中，不需要多長時間就能成為一個穩定的框架。另一個很酷的事情是，它會作為一個 Django 1.8 的外部應用，你可以將它集成到你現有的網站。

看完本文有收穫？請轉發分享給更多人

關注「Python開發者」，提升Python技能

喜歡這篇文章嗎？立刻分享出去讓更多人知道吧！