Offloading Websockets and Server-Sent Events AKA “Combine them with Django safely”

Offloading Websockets and Server-Sent Events AKA “Combine them with Django safely”

作者：Roberto De Ioris

日期：20140315

免责声明

这篇文章显示了将websockets（或者sse）与Django以一种“安全的方式”结合起来的一种相当高级的方法。它不会向你展示websockets和sse有多酷，或者如何用它们写出更好的应用，而是试图让你避免使用它们的最差实践。

在我看来，Python面向web的世界正面临着一种通信/市场问题：大量的人在非阻塞技术（例如gevent）上运行大量阻塞应用（例如django）只是因为有人告诉他们这很酷并且会解决他们所有的扩展问题。

这是完全错误，危险和邪恶的，你不能将阻塞应用和非阻塞引擎混在一起，即使是一个单一、非常小的阻塞部分都能潜在摧毁你整个栈。正如我已经说了几十次一样，如果你的应用是99.9999999%非阻塞的，那么它仍然是阻塞的。

并且不是，你的Django应用上的猴子补丁并非魔法。除非你正使用高度自定义的数据库适配器，对工作在非阻塞模式进行了调整，否则你就是错的。

以看起来就是个超级大混蛋的代价，我强烈建议你完全忽略人们让你将你的Django应用移到gevent, eventlet, tornado或者其他什么的，而不警告你你会遇到数百个问题的建议。

话虽如此，我爱gevent，它可能是uWSGI项目中支持得最好的（带perl的Coro::AnyEvent）循环引擎了。因此，在这篇文章中，我将使用gevent来为Django部分管理websocket/sse流量和纯多进程。

如果这最后一句对你来说像是废话，那么你可能不知道uWSGI的卸载是什么……

uWSGI卸载

这个概念并非是个新东西，或者是uWSGI特有的。诸如nodejs或者twisted这样的项目已经用它多年了。

注解

一个提供静态文件服务的web应用的例子并不非常有趣，也不是用来展示的最好的东西，但是在稍后展示一个使用X-Sendfile的真实世界的例子的时候，会游泳。

想象这个简单的WSGI应用：

def application(env, start_response):
    start_response('200 OK',[('Content-Type','text/plain')])
    f = open('/etc/services')
    # do not do it, if the file is 4GB it will allocate 4GB of memory !!!
    yield f.read()

这将会简单返回/etc/services的内容。它是一个相当小的文件，因此在几毫秒时间内，你的进程将会准备好处理另一个请求。

那要是/etc/services是4千兆字节呢？你的进程（或者线程）将会阻塞几秒（甚至几分钟），并且不能够管理其他请求，直到完全传输这个文件。

如果你可以告诉其他线程为你发送这个文件，这样你就能够管理其他请求，岂不是很酷？

卸载就是这样的：它会为你提供一个或多个线程来为你完成一些简单并且慢点任务。哪种任务呢？所有那些可以以一种非阻塞方式管理的任务，因此单个线程就可以为你管理上千个传输。

你可以将其视为你的电脑中的DMA引擎，你的CPU将会编程DMA来将内存从控制器传输到RAM，然后将会被释放，以完成其他任务，同时DMA在后台工作。

要在uWSGI中启用卸载，你只需要添加 —offload-threads <n> 选项，其中<n>是每个进程要生成的线程数。 (一般而言，单个线程就够了，但是如果你想要使用/滥用你的多CPU核，那么随意增加)

一旦启用了卸载，那么每当uWSGI检测到一个操作能够安全被卸载的时候，它将自动使用它。

在python/WSGI中，任何对wsgi.file_wrapper的使用将会被自动卸载，以及当你使用uWSGI代理特性来传递请求到其他使用uwsgi或者HTTP协议到服务器时。

一个很酷的例子 (甚至在uWSGI文档的Snippets页面也有展示) 是实现一个卸载助力的X-Sendfile特性：

[uwsgi]
; load router_static plugin (compiled in by default in monolithic profiles)
plugins = router_static
 
; spawn 2 offload threads
offload-threads = 2
 
; files under /etc can be safely served (DANGEROUS !!!)
static-safe = /etc
 
; collect the X-Sendfile response header as X_SENDFILE var
collect-header = X-Sendfile X_SENDFILE
 
; if X_SENDFILE is not empty, pass its value to the "static" routing action (it will automatically use offloading if available)
response-route-if-not = empty:${X_SENDFILE} static:${X_SENDFILE}
 
; now the classic options
plugins = python
; bind to HTTP port 8080
http-socket = :8080
; load a simple wsgi-app
wsgi-file = myapp.py

现在，在我们的应用中，我们可以X-Sendfile来在无阻塞情况下发送静态文件：

def application(env, start_response):
    start_response('200 OK',[('X-Sendfile','/etc/services')])
    return []

在这篇文章中将会使用一个非常类似的概念：我们将会使用一个正常的Django来设置我们的会话，来认证用户，以及任意（快的）东东，然后我们会返回一个特别的头，它会指示uWSGI卸载连接到另一个uWSGI实例 (监听一个私有socket)，这个实例将使用gevent，以一种非阻塞方式管理websocket/sse事务。

我们的SSE应用

SSE部分将非常简单，一个基于gevent的WSGI应用将会每秒发送当前时间：

from sse import Sse
import time
 
def application(e, start_response):
    print e
    # create the SSE session
    session = Sse()
    # prepare HTTP headers
    headers = []
    headers.append(('Content-Type','text/event-stream'))
    headers.append(('Cache-Control','no-cache'))
    start_response('200 OK', headers)
    # enter the loop
    while True:
        # monkey patching will prevent sleep() blocking
        time.sleep(1)
        # add the message
        session.add_message('message', str(time.time()))
        # send to the client
        yield str(session)

让我们在/tmp/foo UNIX socket上运行它 (将应用保存为sseapp.py)

uwsgi --wsgi-file sseapp.py --socket /tmp/foo --gevent 1000 --gevent-monkey-patch

(time.sleep()需要猴子补丁，如果你想要/喜欢的话，随意使用gevent原语来休眠)

（无趣的）HTML/Javascript

<html>
    <head>
    </head>
    <body>
      <h1>Server sent events</h1>
      <div id="event"></div>
      <script type="text/javascript">
 
      var eventOutputContainer = document.getElementById("event");
      var evtSrc = new EventSource("/subscribe");
 
      evtSrc.onmessage = function(e) {
          console.log(e.data);
          eventOutputContainer.innerHTML = e.data;
      };
 
      </script>
    </body>
  </html>

它非常简单，它将连接到/subscribe，并且将开始等待事件。

Django视图

我们的django视图，将非常简单，它将简单生成一个特色的响应头 (我们讲称之为X-Offload-to-SSE)，并且把登录用户的用户名作为它的值：

def subscribe(request):
    response = HttpResponse()
    response['X-Offload-to-SSE'] = request.user
    return response

现在，我们已经为“高级”部分准备好了。

让我们卸载SSE事务

配置看起来会有点复杂，但是它与之前看到的X-Sendfile概念相同：

[uwsgi]
; the boring part
http-socket = :9090
offload-threads = 2
wsgi-file = sseproject/wsgi.py
 
; collect X-Offload-to-SSE header and store in var X_OFFLOAD
collect-header = X-Offload-to-SSE X_OFFLOAD
; if X_OFFLOAD is defined, do not send the headers generated by Django
response-route-if-not = empty:${X_OFFLOAD} disableheaders:
; if X_OFFLOAD is defined, offload the request to the app running on /tmp/foo
response-route-if-not = empty:${X_OFFLOAD} uwsgi:/tmp/foo,0,0

唯一“新的”部分是使用 disableheaders 路由动作。这是必须的，否则Django生成的头将会伴着由基于gevent的应用生成的头发送。

你可以避免它 (记住，只在2.0.3添加了 disableheaders )，在gevent应用中移除到start_response()到调用 (冒着被一些WSGI神诅咒的风险)，然后修改Django视图来设置正确的头部：

def subscribe(request):
    response = HttpResponse()
    response['Content-Type'] = 'text/event-stream'
    response['X-Offload-to-SSE'] = request.user
    return response

最终，你或许想要更加“精简”，并简单检测’text/event-stream’ content_type存在：

[uwsgi]
; the boring part
http-socket = :9090
offload-threads = 2
wsgi-file = sseproject/wsgi.py
 
; collect Content-Type header and store in var CONTENT_TYPE
collect-header = Content-Type CONTENT_TYPE
; if CONTENT_TYPE is 'text/event-stream', forward the request
response-route-if = equal:${CONTENT_TYPE};text/event-stream uwsgi:/tmp/foo,0,0

现在，如何在gevent应用中访问Django登录用户的用户名呢？

你应该注意到，gevent应用在每个请求中打印了WSGI环境变量的内容。那个环境变量与Django应用+已收集头部相同。因此，访问environ[‘X_OFFLOAD’]将会返回已登录用户名。 (显然，在第二个例子中，使用了内容类型，不再收集带用户名的变量，因此你应该修复它)

你可以使用相同的方法传递所有你需要的信息，你可以收集所有你需要的变量，等等等等。

你甚至可以在运行时添加变量：

[uwsgi]
; the boring part
http-socket = :9090
offload-threads = 2
wsgi-file = sseproject/wsgi.py
 
; collect Content-Type header and store in var CONTENT_TYPE
collect-header = Content-Type CONTENT_TYPE
 
response-route-if = equal:${CONTENT_TYPE};text/event-stream addvar:FOO=BAR
response-route-if = equal:${CONTENT_TYPE};text/event-stream addvar:TEST1=TEST2
 
; if CONTENT_TYPE is 'text/event-stream', forward the request
response-route-if = equal:${CONTENT_TYPE};text/event-stream uwsgi:/tmp/foo,0,0

或者 (使用goto以获得更好的可读性):

[uwsgi]
; the boring part
http-socket = :9090
offload-threads = 2
wsgi-file = sseproject/wsgi.py
 
; collect Content-Type header and store in var CONTENT_TYPE
collect-header = Content-Type CONTENT_TYPE
 
response-route-if = equal:${CONTENT_TYPE};text/event-stream goto:offload
response-route-run = last:
 
response-route-label = offload
response-route-run = addvar:FOO=BAR
response-route-run = addvar:TEST1=TEST2
response-route-run = uwsgi:/tmp/foo,0,0

使用uwsgi api (>= uWSGI 2.0.3) 进行简化

虽然处理头部是相当HTTP友好型的，但uWSGI 2.0.3增加了在代码中直接定义每个请求变量的可能性。

这允许一个更“优雅”的方式 (即使高度不可移植):

import uwsgi
 
def subscribe(request):
    uwsgi.add_var("LOGGED_IN_USER", request.user)
    uwsgi.add_var("USER_IS_UGLY", "probably")
    uwsgi.add_var("OFFLOAD_TO_SSE", "y")
    uwsgi.add_var("OFFLOAD_SERVER", "/tmp/foo")
    return HttpResponse()

现在，配置可以修改成更优雅：

; the boring part
http-socket = :9090
offload-threads = 2
wsgi-file = sseproject/wsgi.py
 
; if OFFLOAD_TO_SSE is 'y', do not send the headers generated by Django
response-route-if = equal:${OFFLOAD_TO_SSE};y disableheaders:
; if OFFLOAD_TO_SSE is 'y', offload the request to the app running on 'OFFLOAD_SERVER'
response-route-if = equal:${OFFLOAD_TO_SSE};y uwsgi:${OFFLOAD_SERVER},0,0

你注意到我们如何允许Django应用设置后端服务器来使用请求变量了吗？

现在，我们可以更进一步。我们不会使用路由框架 (除了禁用头部生成):

import uwsgi
 
def subscribe(request):
    uwsgi.add_var("LOGGED_IN_USER", request.user)
    uwsgi.add_var("USER_IS_UGLY", "probably")
    uwsgi.route("uwsgi", "/tmp/foo,0,0")
    return HttpResponse()

以及一个简单的：

; the boring part
http-socket = :9090
offload-threads = 2
wsgi-file = sseproject/wsgi.py
 
response-route = ^/subscribe disableheaders:

Websockets怎样？

我们已经看到了如何卸载SSE (那是单向的)。我们也可以卸载websockets (那是双向的)。

概念是相同的，你只需要保证 (和之前一样) Django没有发送任何头部，(否则，websocket握手将会失败)，然后，你可以修改你的gevent应用：

import time
import uwsgi
 
def application(e, start_response):
    print e
    uwsgi.websocket_handshake()
    # enter the loop
    while True:
        # monkey patching will prevent sleep() to block
        time.sleep(1)
        # send to the client
        uwsgi.websocket_send(str(time.time()))

使用redis或者uWSGI缓存框架

请求变量是方便的 (并且有趣)，但是它们有限 (见下)。如果你需要在Django和sse/websocket应用之间传递大量的数据，那么Redis是个不错的方式 (并且和gevent完美契合)。基本上，你存储来自Django的信息到redis中，然后只传递哈希键 (通过请求变量) 到sse/websocket应用。

可以用uWSGI缓存框架完成相同的工作，但是考虑到redis拥有大量的数据原语，而uWSGI只支持key->value项。

常见陷阱

你可以添加到每个请求上的变量总数是由uwsgi包缓存（默认是4k）限制的。你可以用–buffer-size选项将其增至64k。
这是这篇文章的重点：不要在你的gevent应用中使用Django ORM，除非你知道你在干什么！！！(读一读：你有一个Django数据库适配器，它支持gevent，并且与标准的相比，它并不糟糕……)
忘记找到一种方式来禁用Django中的头部生成吧。这是它的WSGI适配器的一个“限制/特性”，使用uWSGI设施 (如果可用的话)，或者不要在你的gevent应用中生成头部。最终，你可以以这种方式修改wsgi.py：

"""
WSGI config for sseproject project.
 
It exposes the WSGI callable as a module-level variable named ``application``.
 
For more information on this file, see
https://docs.djangoproject.com/en/1.6/howto/deployment/wsgi/
"""
 
import os
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "sseproject.settings")
 
from django.core.wsgi import get_wsgi_application
django_application = get_wsgi_application()
 
def fake_start_response(status, headers, exc_info=None):
    pass
 
def application(environ, start_response):
    if environ['PATH_INFO'] == '/subscribe':
        return django_application(environ, fake_start_response)
    return django_application(environ, start_response)