Python Socket.IO全方位指南

一、Socket.IO简介

Socket.IO 是一个基于网络实时通讯的 JavaScript 库。它包含了服务器端（Node.js）和客户端（浏览器）两个部分，通过 WebSockets 的技术，它能够实现实时通讯的功能。Socket.IO 广泛运用于在线游戏、在线聊天等实时通讯应用中。

二、Python Socket.IO

Python 的 Socket.IO 实现是建立在 gevent 或 eventlet 基础上，并且支持 Python 2.x 和 Python 3.x。Python Socket.IO 不仅继承了 Socket.IO 的优点，如实时通讯、基于事件、封装的消息传输机制等特点，还具有 Python 语言本身的优点，如语言易用、有大量的优秀库和框架支持。

三、Python Socket.IO 安装

Python Socket.IO 的安装很简单，可以通过 pip 直接安装：

pip install python-socketio
pip install python-socketio[client]
pip install python-socketio[asyncio_client]
pip install python-socketio[django]
pip install python-socketio[gunicorn]
pip install python-socketio[gevent]
pip install python-socketio[eventlet]
pip install python-socketio[eventlet_old]
pip install python-socketio[gevent_uwsgi]

四、Python Socket.IO 服务器端编程

Python Socket.IO 服务器端编程非常简单直接，下面是一个简单的 Python Socket.IO 服务器端示例：

import socketio

app = socketio.WSGIApp()

@sio.on(‘my event‘)
def my_event(sid, data):
    print(‘my_event’, data)

if __name__ == '__main__':
    app.run(‘0.0.0.0‘, 5000)

五、Python Socket.IO 客户端编程

Python Socket.IO 客户端编程也非常简单，下面是一个简单的 Python Socket.IO 客户端示例：

import socketio

sio = socketio.Client()

@sio.on(‘my response‘)
def my_response(data):
    print(‘my_response‘, data)

sio.connect(‘http://localhost:5000‘)
sio.emit(‘my event‘, {‘data’: ‘my data‘})

六、Python Socket.IO 实时通讯应用示例

Python Socket.IO 可以用于各种在线实时通讯应用，下面是一个在线游戏示例：

# 服务器端
import socketio
import json

sio = socketio.Server()
app = socketio.WSGIApp(sio)

room = {}
sid2room = {}

@sio.on('join')
def join(sid, data):
    roomid = data.get('roomid')
    nickname = data.get('nickname')
    sio.enter_room(sid, roomid)
    members = room.get(roomid, {}).get('members', [])
    members.append({'sid': sid, 'nickname': nickname})
    room[roomid] = {'members': members}
    sid2room[sid] = roomid
    sio.emit('join_response', room[roomid], room=roomid)
    sio.emit('members_response', room[roomid], room=roomid)

@sio.on('direction')
def direction(sid, data):
    roomid = sid2room[sid]
    sio.emit('direction_response', data, room=roomid, skip_sid=sid)

@sio.event
def disconnect(sid):
    roomid = sid2room.get(sid)
    if roomid:
        members = room.get(roomid, {}).get('members', [])
        members = [m for m in members if m['sid'] != sid]
        room[roomid]['members'] = members
        sio.emit('members_response', room[roomid], room=roomid)

if __name__ == '__main__':
    app.run()

# 客户端
import socketio
import pygame

sio = socketio.Client()

def run():
    pygame.init()
    size = width, height = 600, 400
    screen = pygame.display.set_mode(size)
    pygame.display.set_caption('Python Socket.IO Game Demo')
    clock = pygame.time.Clock()
    while True:
        clock.tick(30)
        for event in pygame.event.get():
            if event.type == pygame.QUIT:
                pygame.quit()
                return
            if event.type == pygame.KEYDOWN:
                if event.key == pygame.K_LEFT:
                    data = {'nickname': nickname, 'direction': 'left'}
                    sio.emit('direction', data)
                elif event.key == pygame.K_RIGHT:
                    data = {'nickname': nickname, 'direction': 'right'}
                    sio.emit('direction', data)
                elif event.key == pygame.K_UP:
                    data = {'nickname': nickname, 'direction': 'up'}
                    sio.emit('direction', data)
                elif event.key == pygame.K_DOWN:
                    data = {'nickname': nickname, 'direction': 'down'}
                    sio.emit('direction', data)
        sio.sleep(0.01)

@sio.on('connect')
def connect():
    print('connected')
    join_data = {'roomid': roomid, 'nickname': nickname}
    sio.emit('join', join_data)

@sio.on('join_response')
def join_response(data):
    print('join_response', data)

@sio.on('members_response')
def members_response(data):
    print('members_response', data)

@sio.on('direction_response')
def direction_response(data):
    print('direction_response', data)

if __name__ == '__main__':
    roomid = 'room1'
    nickname = 'player1'
    sio.connect('http://localhost:5000')
    run()