nginx+flask+gevent+uwsgi实现websocket

Websocket简介

WebSocket是HTML5开始提供的一种在单个 TCP 链接上进行全双工通信的协议。在WebSocket API中，浏览器和服务器只须要作一个握手的动做，而后，浏览器和服务器之间就造成了一条快速通道。二者之间就直接能够数据互相传送。浏览器经过 JavaScript向服务器发出创建 WebSocket 链接的请求，链接创建之后，客户端和服务器端就能够经过 TCP 链接直接交换数据。

Websocket自己是一个持久化的新协议，和http没有很大的关系，但websocket是基于http协议的，借用http协议完成了部分握手。

咱们在某个linux服务器上部署了一个websocket，而后咱们发起链接请求，看下请求发送的内容，对比下http请求和websocket请求的不一样：

      

能够看到，主要的不一样点，Connection: Upgrade；Upgrade: websocket，还有3个Sec开头的参数。前两个参数是用来告诉nginx服务，我发起的是websocket协议，而不是http，后三个参数是用来校验websocket服务是不是websocket的，以及校验websocket协议的版本。

在消息发出后，websocket服务端返回一个response：

Connection:upgrade

Date:Thu, 26 Oct 2017 06:01:51 GMT

Sec-WebSocket-Accept:3P+uxAGozJzj3vcr89gxFzydOu8=

Sec-WebSocket-Origin:http://10.43.35.31:2327

Server:nginx/1.12.0

Upgrade:WebSocket

当收到服务端返回，证实协议已经转换成功，后面合http再无关系了，而Sec开头的是证实我是websocket。

为何使用websocket

没有websocket的时候，经常使用的实时同步数据的方式通常有两种：一、轮询，即每隔必定的时间客户端就向服务端发起一次请求，客户端根据返回数据更新界面，这须要服务器有很快的处理速度和资源。二、长连接，即客户端向服务端发送查询请求，若是服务端无数据，则一直阻塞，直到有数据返回，返回后客户端再重复此过程，这种须要很高的并发。

这两种方式都须要不断的创建http链接，服务端不能主动推送数据给客户端，比较被动，比较耗资源，好比屡次请求的HTTP头部，TCP链接复用会致使的线头阻塞(即因为前面数据丢失致使的后续数据堆积)。

Websocket很好的解决了上面的问题：只须要在创建链接的时候发送一次http请求，后面的交互都不须要再发了；WebSocket的链接是双向通讯的链接，在同一个TCP链接上，既能够发送，也能够接收；几个不一样的URI能够复用同一个WebSocket链接。

Websocket基础架构及实现

 本次websocket的搭建考虑性能问题，采起的是flask+python+uwsgi+gevent+nginx的基础架构。采用uwsgi+gevent的模式启动websocket进程，为gevent指定并发为100，web框架使用小而轻的Flask，使用virtualenv建立虚拟环境，保证websocket不影响主工程，且能够单独发布部署，本进程的依赖为Flask，gevent和uwsgi。

Nginx须要分别为http和https增长配置

上面是nginx的配置，nginx是从1.3开始支持websocket的，upgrade将http链接升级到websocket链接，Upgrade机制使用了Upgrade协议头和Connection协议头，所以当代理服务器拦截到来自客户端的Upgrade请求时，代理服务器须要将本身的Upgrade请求发送给后端服务器，包括适合的请求头。Nginx经过在客户端和后端服务器之间创建隧道来支持WebSockets通讯。为了让Nginx能够未来自客户端的Upgrade请求发送到后端服务器，Upgrade和Connection的头信息必须被显式的设置。

Webserver采用的是Flask，下面简单介绍下本工程代码实现，包含几个部分：

一、客户端部分，客户端采用

ws = new WebSocket('ws://' + document.domain + ':' + location.port + '/websocket');或

ws = new WebSocket('wss://' + document.domain + ':' + location.port + '/websocket');来创建链接，并监听onopen，onmessage，onclose事件。

二、服务端部分

首先建立一个flask的app，而后传入GeventWebSocket，获得一个websocket的对象。

app = Flask(__name__)
ws = GeventWebSocket(app)

GeventWebSocket继承了websocket，重写了init_app，由于要使用gevent，重写主要是为了打上猴子补丁，还替换了websocket中间件，也是为了适配gevent。

class GeventWebSocket(WebSocket):
    middleware = GeventWebSocketMiddleware

    def init_app(self, app):
        logger.info("init_app")
        aggressive = app.config.get('UWSGI_WEBSOCKET_AGGRESSIVE_PATCH', True)
        patch_all(aggressive=aggressive)
        super(GeventWebSocket, self).init_app(app)

Websocket模块主要为了按要求启动websocket服务端，主要的3个函数：

route：主要是为view添加url规则；

def route(self, rule, **options):
    def decorator(f):
        endpoint = options.pop('endpoint', None)
        self.add_url_rule(rule, endpoint, f, **options)
        return f
    return decorator

init_app：修改flask的app的属性，使在启动app时能以uwsgi启动，并重写wsgi_app，供uwsgi进行调用。

def init_app(self, app):
    self.app = app
    app.wsgi_app = self.middleware(app.wsgi_app, self)
    app.run = lambda **kwargs: self.run(**kwargs)

run：以uwsgi方式启动app，并指定相关启动函数。

def run(self, app=None, debug=False, host='127.0.0.1', port=5000, uwsgi_binary=None, **kwargs):
    if not app:
        app = self.app.name + ':app'

    if self.app.debug:
        debug = True
    run_uwsgi(app, debug, host, port, uwsgi_binary, **kwargs)

中间件替换为了子类的中间件，主要是提供了wsgi_app供uwsgi进行调用。wsgi server工做流程以下：服务器建立socket，监听端口，等待客户端链接；当有请求来时，服务器解析客户端信息放到环境变量environ中，并调用绑定的handler来处理请求；handler解析这个http请求，将请求信息例如method，path等放到environ中；wsgi handler再将一些服务器端信息也放到environ中，最后服务器信息，客户端信息，本次请求信息所有都保存到了环境变量environ中；wsgi handler 调用注册的wsgi app，并将environ和回调函数传给wsgi app；wsgi app 将reponse header/status/body 回传给wsgi handler；handler仍是经过socket将response信息塞回给客户端。

Wsgi app先经过environ获取到请求的url，而后定位到具体调用的函数，

urls = self.websocket.url_map.bind_to_environ(environ)
try:
    endpoint, args = urls.match()
    handler = self.websocket.view_functions[endpoint]
except HTTPException:
    handler = None

而后，根据environ获取websocket_key，以后与客户端完成握手，

uwsgi.websocket_handshake(environ['HTTP_SEC_WEBSOCKET_KEY'],
                          environ.get('HTTP_ORIGIN', ''))

而后建立基于gevent的发送事件、队列和监听事件、队列；而后建立client来操控这两个队列完成发送和接收。

send_event = Event()
send_queue = Queue()
recv_event = Event()
recv_queue = Queue()

client = self.client(environ, uwsgi.connection_fd(), send_event,
                     send_queue, recv_event, recv_queue,
                     self.websocket.timeout)

而后启动协程，建立发送消息的tcp链路，

handler = spawn(handler, client, **args)
def listener(client):
    select([client.fd], [], [], client.timeout)
    recv_event.set()
listening = spawn(listener, client)

进入等待，一旦监听到发送或接收事件，则调用uwsgi方法对消息进行处理，若是handler完成或者被杀死，则终止监听，

wait([handler, send_event, recv_event], None, 1)
# handle send events
if send_event.is_set():
    try:
        while True:
            uwsgi.websocket_send(send_queue.get_nowait())
    except Empty:
        send_event.clear()
    except IOError:
        client.connected = False

# handle receive events
elif recv_event.is_set():
    recv_event.clear()
    try:
        message = True
        while message:
            message = uwsgi.websocket_recv_nb()
            recv_queue.put(message)
        listening = spawn(listener, client)
    except IOError:
        client.connected = False

# handler done, we're outta here
elif handler.ready():
    listening.kill()
    return ''

这里面使用的client实例主要就实现了对于队列的操做和事件状态的变动，

def send(self, msg, binary=True):
    if binary:
        return self.send_binary(msg)
    self.send_queue.put(msg)
    self.send_event.set()

 

def receive(self):
    return self.recv_queue.get()

 

def close(self):
    self.connected = False

而后就是app里使用websocket的方法了，我在全局生命了一个client的list，用来存储已链接的wensocket对象，每次创建链接，都会将新的链接对象存进去，

client = []

@ws.route('/websocket')
def start(channel):
    global client
    client.append(channel)
    channel.send('websocket is running')
    while True:
        channel.receive()

又起了一个view，定义为post，接受咱们本身工程的消息上报，每次消息上报只上报正在链接的对象，已经不链接的从列表删除，

@app.route('/send', methods=['POST'])
def send_msg():
    message = request.data
    send_all_client(message)
    delete_close_client()
    return 'SUCCESS'

全部须要上报的消息，统一走这个rest接口通知就能够，app会将消息上报到每一个链接的客户端，最后为app指定gevent线程数和监听端口，

if __name__ == '__main__':
    app.run(gevent=100, port=5000)

每次经过python app.py启动工程便可，代码会自动拼装启动命令，以下：

/home/ngomm/websocket/ENV/bin/uwsgi --http 127.0.0.1:5000 --http-websockets --virtualenv /home/ngomm/websocket/ENV --gevent 100 --master --wsgi app:app

由于但愿能够独立部署，因此为工程建立了虚拟环境，该工程依赖的包为Flask (0.12.2)，gevent (1.1.2)，uWSGI (2.0.11.2)。