2023年无需第三方工具实现Bottle+WebSocket极简示例可行吗？

答案是不行。Bottle是基于WSGI规范的框架，而WSGI本身是为HTTP设计的同步协议，不支持WebSocket所需的双向持久连接和协议切换逻辑。默认的Bottle服务器（基于wsgiref）没有实现WebSocket的握手流程和帧处理能力，所以你代码里尝试获取的wsgi.websocket对象始终为None，直接导致连接失败。

你的代码连接失败的原因 #

你代码中的wsock = bottle.request.environ.get('wsgi.websocket')只有在支持WebSocket扩展的WSGI服务器（比如gevent-websocket、eventlet）中才会被填充有效对象。默认服务器会把浏览器发起的WebSocket连接当作普通HTTP请求处理，自然无法完成协议切换，最终连接中断。

可行的Bottle+WebSocket实现（需轻量第三方依赖） #

如果要基于Bottle快速实现WebSocket功能，最成熟的方案是使用gevent-websocket，这也是生态中的标准做法。修正后的可运行代码如下：

Python代码

from bottle import Bottle, request, static_file
from gevent.pywsgi import WSGIServer
from geventwebsocket import WebSocketError
from geventwebsocket.handler import WebSocketHandler

app = Bottle()

@app.route("/")
def index():
    return static_file("index.html", root=".")

@app.route("/ws")
def handle_websocket():
    wsock = request.environ.get('wsgi.websocket')
    if not wsock:
        return "Expected WebSocket request."
    
    while True:
        try:
            message = wsock.receive()
            if message is None:
                break
            wsock.send(f"Your message was: {message!r}")
        except WebSocketError:
            break

if __name__ == "__main__":
    server = WSGIServer(("127.0.0.1", 5000), app, handler_class=WebSocketHandler)
    server.serve_forever()

index.html代码

<html>
<body>
<div>WebSocket Test</div>
<script>
const ws = new WebSocket("ws://127.0.0.1:5000/ws");
ws.onopen = () => {
    console.log("Connected to WebSocket");
    ws.send("Hello, world!");
};
ws.onmessage = (evt) => {
    console.log("Received:", evt.data);
};
ws.onclose = () => {
    console.log("Disconnected from WebSocket");
};
</script>
</body>
</html>

完全不依赖第三方库的方案（脱离Bottle路由） #

如果坚持完全不用第三方库，你需要手动实现WebSocket协议的握手和帧解析逻辑，直接使用Python原生socket模块处理连接。这种方式不再依赖Bottle的WSGI架构，适合学习WebSocket协议原理，示例代码如下：

import socket
import hashlib
import base64

def handle_websocket(conn):
    # 读取HTTP握手请求
    request_data = conn.recv(1024).decode()
    if "Upgrade: websocket" not in request_data:
        conn.close()
        return
    
    # 计算WebSocket握手响应密钥
    key = [line.split(": ")[1] for line in request_data.split("\r\n") if "Sec-WebSocket-Key" in line][0]
    accept_key = base64.b64encode(hashlib.sha1((key + "258EAFA5-E914-47DA-95CA-C5AB0DC85B11").encode()).digest()).decode()
    
    # 发送握手响应
    response = (
        "HTTP/1.1 101 Switching Protocols\r\n"
        "Upgrade: websocket\r\n"
        "Connection: Upgrade\r\n"
        f"Sec-WebSocket-Accept: {accept_key}\r\n\r\n"
    )
    conn.send(response.encode())
    
    # 简化处理文本消息帧
    while True:
        try:
            frame = conn.recv(1024)
            if not frame:
                break
            # 解析客户端发送的带掩码文本帧
            opcode = frame[0] & 0x0F
            if opcode == 1:
                payload_length = frame[1] & 0x7F
                mask = frame[2:6]
                payload = frame[6:6+payload_length]
                # 解掩码
                unmasked_payload = bytearray()
                for i in range(payload_length):
                    unmasked_payload.append(payload[i] ^ mask[i % 4])
                message = unmasked_payload.decode()
                print(f"Received: {message}")
                
                # 构造响应帧（不带掩码的文本帧）
                response_payload = f"Your message was: {message!r}".encode()
                response_frame = bytearray()
                response_frame.append(0x81)  # FIN标记 + 文本操作码
                if len(response_payload) < 126:
                    response_frame.append(len(response_payload))
                else:
                    response_frame.append(126)
                    response_frame.extend(len(response_payload).to_bytes(2, byteorder='big'))
                response_frame.extend(response_payload)
                conn.send(response_frame)
            elif opcode == 8:  # 处理关闭帧
                conn.close()
                break
        except:
            conn.close()
            break

def main():
    server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    server_socket.bind(("127.0.0.1", 5000))
    server_socket.listen(5)
    print("Server listening on ws://127.0.0.1:5000/ws")
    
    while True:
        conn, addr = server_socket.accept()
        print(f"New connection from {addr}")
        handle_websocket(conn)

if __name__ == "__main__":
    main()

内容的提问来源于stack exchange，提问作者Basj