Socket.IO是一种传输协议,支持客户端与服务器之间基于事件的实时双向通信。客户端通常但不总是 web浏览器,一般情况下,客户端是采用 js 编写的,当然,也可以用 python、Java 等。
客户端和服务器组件的官方实现是用 JavaScript 编写的。这个包提供了这两种语言的 Python 实现,每个都有标准和异步变体。
1.客户端示例
下面的示例是一个简单的Python客户端:
import socketio
sio = socketio.Client()
@sio.event
def connect():
print('connection established')
@sio.event
def my_message(data):
print('message received with ', data)
sio.emit('my response', {'response': 'my response'})
@sio.event
def disconnect():
print('disconnected from server')
sio.connect('http://localhost:5000')
sio.wait()
下面是一个类似的客户端,编码为asyncio(仅Python 3.5+):
import asyncio
import socketio
sio = socketio.AsyncClient()
@sio.event
async def connect():
print('connection established')
@sio.event
async def my_message(data):
print('message received with ', data)
await sio.emit('my response', {'response': 'my response'})
@sio.event
async def disconnect():
print('disconnected from server')
async def main():
await sio.connect('http://localhost:5000')
await sio.wait()
if __name__ == '__main__':
asyncio.run(main())
2.服务端示例
以下应用程序是使用 Eventlet 异步服务器的基本服务器示例:
import eventlet
import socketio
sio = socketio.Server()
app = socketio.WSGIApp(sio, static_files={
'/': {'content_type': 'text/html', 'filename': 'index.html'}
})
@sio.event
def connect(sid, environ):
print('connect ', sid)
@sio.event
def my_message(sid, data):
print('message ', data)
@sio.event
def disconnect(sid):
print('disconnect ', sid)
if __name__ == '__main__':
eventlet.wsgi.server(eventlet.listen(('', 5000)), app)
下面是一个类似的应用程序,编码为asyncio(仅Python 3.5+)和Uvicorn web服务器:
from aiohttp import web
import socketio
sio = socketio.AsyncServer()
app = web.Application()
sio.attach(app)
async def index(request):
"""Serve the client-side application."""
with open('index.html') as f:
return web.Response(text=f.read(), content_type='text/html')
@sio.event
def connect(sid, environ):
print("connect ", sid)
@sio.event
async def chat_message(sid, data):
print("message ", data)
@sio.event
def disconnect(sid):
print('disconnect ', sid)
app.router.add_static('/static', 'static')
app.router.add_get('/', index)
if __name__ == '__main__':
web.run_app(app)