Python3: 支持 WebSocket 客户端/服务端、同步/异步的多个 Python 模块

Bertha 。 2021-09-21 23:00 1366阅读 0赞

本文链接: [https://xiets.blog.csdn.net/article/details/115558069][https_xiets.blog.csdn.net_article_details_115558069]

[Python3 学习笔记（目录）][Python3]

Python 有多个支持 WebSocket 客户端、服务端、同步、异步 的第三方相关模块。

用于 webscoket echo 回显测试的地址，连接上服务器后给服务器发送消息，服务器会把消息回传给客户端：

*  webscoket echo 测试地址: `ws://echo.websocket.org` 或 `wss://echo.websocket.org`
 *  webscoket HTML5 客户端: [https://websocket.org/echo.html][https_websocket.org_echo.html]

WebSocket 协议标准:

*  [https://datatracker.ietf.org/doc/html/rfc6455][https_datatracker.ietf.org_doc_html_rfc6455]
 *  [https://datatracker.ietf.org/doc/html/rfc7692][https_datatracker.ietf.org_doc_html_rfc7692]

# 1. websocket-client 模块 #

`websocket-client` 是 Python 的 WebSocket (同步)客户端，它提供了对 WebSocket 低级 API 的访问。websocket-client 目前版本实现了 WebSocket 协议的 hybi-13 版本，不支持来自 [RFC 7692][https_datatracker.ietf.org_doc_html_rfc7692] 的 permessage-deflate 扩展。

相关链接：

*  用户文档: [https://websocket-client.readthedocs.io/][https_websocket-client.readthedocs.io]
 *  PyPI 项目地址: [https://pypi.org/project/websocket-client/][https_pypi.org_project_websocket-client]
 *  GitHub 仓库: [https://github.com/websocket-client/websocket-client][https_github.com_websocket-client_websocket-client]

安装 `websocket-client` 模块：

$ pip3 install websocket-client

WebSocket 简单示例，连接、发送/接收消息、关闭连接：

>>> import websocket
    >>> 
    >>> # websocket.enableTrace(True) # 打开日志, 将详细输出通讯过程
    >>> 
    >>> ws = websocket.WebSocket()
    >>> ws.connect("ws://echo.websocket.org")
    >>> ws.send("Hello Server")
    18
    >>> ws.recv()
    'Hello Server'
    >>> ws.close()

作为长连接的 WebSocketApp 简单示例：

import websocket
    import time
    import threading
    
    
    def on_open(wsapp):
        print("on_open")
        
        def send_message():
            for i in range(10):
                wsapp.send(f"Hello { i}")
                time.sleep(1)
            wsapp.close()
            
        threading.Thread(target=send_message).start()
    
    
    def on_message(wsapp, message):
        print("on_message:", message)
    
    
    def on_close(wsapp):
        print("on_close")
    
    
    wsapp = websocket.WebSocketApp("ws://echo.websocket.org",
                                   on_open=on_open,
                                   on_message=on_message,
                                   on_close=on_close)
    wsapp.run_forever()

`WebSocket` 只适用于短期连接（通过自己封装也可以用于长期连接），长期连接需使用 `WebSocketApp`。

## 1.1 websocket.WebSocket ##

`websocket.WebSocket` 是适合用于短期的 WebSocket 客户端，需要手动接收数据。

### 1.1.1 `WebSocket` 类的 主要方法 ###

`WebSocket` 中主要的几个方法：

*  `class WebSocket()`
 *  `def connect()`
 *  `def send()`
 *  `def recv()`
 *  `def close()`

# 创建一个 WebSocket 客户端对象, 所有参数都是可选参数。
    #
    # get_mask_key: 自定义 mask key 的生成函数, 此参数主要是为了测试目的。
    # 参数类型为一个可调用对象(函数), 此函数有一个 int 类型的参数, 
    # 函数必须返回一个长度为传入参数长度的字符串 (字节数组)。
    #
    # sockopt: socket.setsockopt 的可选参数值, tuple 类型。
    #
    # sslopt: SSL Socket 的可选项参数, 可选项 dict 对象。
    # 例如禁用 SSL 证书校验: sslopt={"cert_reqs": ssl.CERT_NONE}
    # 禁用域名校验: sslopt={"check_hostname": False}
    #
    # fire_cont_frame: 是否为每个 连续帧 (cont frame) 触发 recv 事件。
    # enable_multithread: 如果设置为 True, 则锁定发送方法。
    # skip_utf8_validation: 是否跳过 utf8 校验。
    #
    class WebSocket(get_mask_key=None, sockopt=None, sslopt=None,
                    fire_cont_frame=False, enable_multithread=False,
                    skip_utf8_validation=False, **_)
    
    
    # 连接 WebSocket 服务器。
    #
    # url: WebSocket 服务器地址
    #
    # **options: 可选参数
    # header: list/dict, 自定义的 HTTP 请求头, 格式为: ["name1: value1", ...] 或 {"name1": "value1", ...}
    # cookie: str, 自定义 HTTP 请求头 Cookie 的值
    # host: str, 自定义 HTTP 请求头 Host 的值。
    # origin: str, 自定义 HTTP 请求头 Origin 的值(URL)。
    # connection: str, 自定义 HTTP 请求头 Connection 的值, 默认为 "Upgrade"。
    # suppress_origin: bool, 抑制输出 origin header。
    #
    # timeout: int/float, 超时时间, 单位为秒
    #
    # proxy_type: 代理的类型, 默认为 "http", 其他类型 'socks4', 'socks5', 'socks5h'
    # http_proxy_host: 代理的 host
    # http_proxy_port: 代理的 port, 默认为 80
    # http_no_proxy: 不使用代理的 host names, list 类型, 如: ["host1", "host2", ...]
    # http_proxy_auth: HTTP 代理认证, tuple 类型, 如: ("username", "password")
    #
    # redirect_limit: 最大重定向次数, int 类型
    # subprotocols: 可用的子协议数组
    # socket: 被预先初始化的 Socket 流
    #
    connect(url, **options)
    
    
    # 发送数据
    #
    # payload: 发送的数据(载荷), str/bytes 类型
    #
    # opcode: 操作码, 可选值:
    # ABNF.OPCODE_CONT
    # ABNF.OPCODE_TEXT
    # ABNF.OPCODE_BINARY
    # ABNF.OPCODE_CLOSE
    # ABNF.OPCODE_PING
    # ABNF.OPCODE_PONG
    #
    # 默认为发送 TEXT 数据帧
    #
    send(payload, opcode=ABNF.OPCODE_TEXT)
    
    
    # 接收数据
    # 
    # 接收 ABNF.OPCODE_TEXT 数据帧, 返回 str
    # 接收 ABNF.OPCODE_BINARY 数据帧, 返回 bytes
    # 其他类型的数据将 将被忽略(跳过/不返回) 或 直接返回 ""
    #
    recv()
    
    
    # 关闭 WebSocket, 先发送关闭帧再关闭底层 socket。
    #
    # status: 发送关闭数据帧时的数据(告诉服务端关闭的状态码), 默认为正常关闭。
    # reason: 发送关闭数据帧时的数据(告诉服务端关闭的原因)。
    # timeout: 直到超时还没有收到一个关闭帧, 则不再等待, 直接关闭底层 socket。
    # 正常为收到关闭帧后再关闭底层 socket。
    # int/float 秒数, None 表示一直等待直到收到一个关闭帧。
    #
    close(status=STATUS_NORMAL, reason=bytes('', encoding='utf-8'), timeout=3.0)

也可以调用 `websocket.create_connection()` 函数直接创建 WebSocket 客户端并连接服务器，函数原型：

# 创建 WebSocket 客户端, 并连接 url 服务器,
    # class_ 表示要创建的客户端的类型,
    # options 参数与 WebSocket.connect() 中的 options 参数一致
    websocket.create_connection(url, timeout=None, class_=WebSocket, **options)

### 1.1.2 `WebSocket` 类的 其他属性和方法 ###

handshake_response      # WebSocket 连接成功后的 握手响应
    getstatus()             # int, 握手响应的 HTTP 状态码
    getheaders()            # dict, 握手响应的 HTTP 响应头
    is_ssl()                # bool, 是否是 SSL 连接
    
    
    send_frame(frame)       # 发送数据帧, 数据帧是 操作码+数据 的封装, 所有操作码和数据发送最终都经过此方法
    send_binary(payload)    # 发送二进制数据, 即: send(payload, ABNF.OPCODE_BINARY)
    ping(payload="")        # 发送 ping 数据, 即: send(payload, ABNF.OPCODE_PING)
    pong(payload="")        # 发送 pong 数据, 即: send(payload, ABNF.OPCODE_PONG)
    
    
    recv_data(control_frame=False)          # 接收数据和操作码, 返回: (opcode, frame.data)
                                            # control_frame 表示如果是控制帧(PING/PONG) 是否返回数据
    recv_data_frame(control_frame=False)    # 接收数据和操作码, 返回: (opcode, frame)
    recv_frame()                            # 接收数据帧, 所有操作码和数据接收最终都经过此方法
    next()                                  # 接收下一条数据, 即: recv()
    
    
    send_close(status=STATUS_NORMAL,        # 手动发送关闭帧, 一般使用 close()
               reason=bytes('', encoding='utf-8'))
    abort()                 # 立即关闭底层 socket, 一般使用 close()
    shutdown()              # 关闭底层 socket, 一般使用 close()
    connected               # 连接是否已关闭
    
    
    set_mask_key(func)      # 构造方法中的 get_mask_key 参数
    getsubprotocol()        # 获取自协议
    gettimeout()            # connect() 中的 timeout
    settimeout(timeout)     # connect() 中的 timeout
    sock                    # 底层 socket
    fileno()                # sock.fileno()

### 1.1.3 `WebSocket` 类的 代码示例 ###

import websocket
    
    ws = websocket.WebSocket()              # 创建 WebSocket 客户端对象
    ws.connect("ws://echo.websocket.org")   # 连接 WebSocket 服务器
    
    ws.send("Hello")        # 发送 TEXT 数据帧, 数据为 "Hello"
    print(ws.recv())        # 接收 TEXT/BINARY 数据帧的数据, 输出: "Hello"
    
    ws.ping()               # 发送 PING 数据帧, 内部调用 send("", ABNF.OPCODE_PING)
    
    print(ws.recv_frame())  # 接收原始数据帧, 结果为 fin=1 opcode=10 data=b'',
                            # 其中 10 表示操作码 ABNF.OPCODE_PONG
    
    ws.close()              # 关闭客户端

`ping()`, `pong()` 和 `recv_xxx()` 方法的说明：

# ping() 和 pong() 是 WebSocket 协议的一部分, 用于 客户端 和 服务端 之间双向测试保活连接。
    #
    # 客户端: 客户端定时向服务端发送 ping 数据帧, 服务端收到后会立即返回一个 pong 数据帧。
    # 服务端: 服务端定时向客户端推送 ping 数据帧, 客户端收到后需要立即发回一个 pong 数据帧。
    #
    # 发出 ping 数据帧后, 如果超时没有收到 pong 数据帧的回复, 则连接可能已断开, 需要重新发起连接。
    # ping 是主动发送, pong 是被动发送, 主动发送 pong 会无反应。
    #
    # 接收数据帧的方法调用层级:
    #
    # recv() -> str/bytes
    # recv_data(control_frame=False) -> (opcode, data)
    # recv_data_frame(control_frame=False) -> (opcode, frame)
    # recv_frame() -> frame
    #
    # recv()
    # 只接收 TEXT 和 BINARY 操作码的数据(载荷),
    # 返回 str/bytes, 如果是其他操作码 则返回 ""
    #
    # recv_data(control_frame=False)
    # 接收 操作码 和 数据(载荷), 返回操作码和数据组成的元祖 (opcode, data)
    # control_frame 参数将直接传递给 recv_data_frame()
    #
    #
    # recv_data_frame(control_frame=False)
    # 接收 操作码 和 数据帧, 返回操作码和数据帧组成的元祖 (opcode, frame)
    #
    # control_frame 参数表示接收到控制帧(PING/PONG) 后是否返回。
    # control_frame=True 表示接收到控制帧(PING/PONG) 后返回, 不再继续接收数据帧。
    # control_frame=False 表示接收到控制帧(PING/PONG) 后不返回, 继续等待接收下一个数据帧。
    #
    # 注意:
    # 此次方法接收到 PING 数据帧后, 会自动给服务端回复 PONG 帧,
    # 无需手动回复, 回复后根据 control_frame 参数决定是否返回。
    #
    # 此方法接收到 CLOSE 数据帧后, 会自动调用 send_close() 回复 CLOSE 数据帧给服务端,
    # 并直接返回 (opcode, frame), 即不会判断 control_frame。
    # 接收到 CLOSE 数据帧一般为服务端通知客户端即将要关闭这个连接, 客户端则标记此连接已关闭, 
    # 并回复服务端一个 CLOSE 数据帧表示客户端成功收到关闭通知了。
    #
    # recv_frame()
    # 接收 数据帧 的原始方法, 返回数据帧对象, 不会自动处理 PING/PONG/CLOSE。
    #
    # 数据帧 = 操作码 + 数据(载荷)

## 1.2 websocket.WebSocketApp ##

`websocket.WebSocketApp` 是对 `websocket.WebSocket` 的封装，支持自动定时发送 PING 帧，支持事件驱动方式的数据帧接收，可用于长期的 WebSocket 连接。

### 1.2.1 `WebSocketApp` 类的主要方法 ###

`WebSocketApp` 的构造方法：

# WebSocketApp 构造方法
    class websocket.WebSocketApp(
                url,
                header=None,
                cookie=None,
                on_open=None,
                on_data=None, 
                on_message=None, 
                on_cont_message=None,
                on_ping=None, 
                on_pong=None,
                on_error=None,
                on_close=None, 
                get_mask_key=None, 
                subprotocols=None)
    
    # 参数说明:
    # on_ 开头的参数均为事件回调函数。
    #
    # url:
    # WebSocket 服务端 URL
    #
    # header:
    # 连接握手时的自定义 HTTP 请求头, 格式参考 WebSocket
    #
    # cookie:
    # 连接握手时自定义 HTTP 请求头 Cookie 的值, 参考 WebSocketApp
    #
    # on_open:
    # WebSocket 连接成功后调用的函数, 只调用一次。
    # 函数格式: 
    # on_open(wsapp)
    #
    # on_data:
    # 接收 TEXT, BINARY 或 CONT 数据帧时调用, 
    # 此方法在 on_message 或 on_cont_message 之前调用。
    # on_data 调用后, on_message 或 on_cont_message 也会调用。
    # 函数格式: 
    # on_data(wsapp, frame_data, frame_opcode, frame_fin)
    #
    # on_message:
    # 接收到 TEXT 或 BINARY 数据帧时调用的函数
    # 函数格式: 
    # on_message(wsapp, data)
    #
    # on_cont_message:
    # 接收到 CONT 数据帧时调用
    # 函数格式: 
    # on_cont_message(wsapp, frame_data, frame_fin)
    #
    # on_ping:
    # 接收到 PING 数据帧时调用, 
    # 不会自动回复 PONG, 需手动发送 PONG 回复服务端
    # 函数格式: 
    # on_ping(wsapp, frame_data)
    #
    # on_pong:
    # 接收到 PONG 数据帧时调用
    # 函数格式: 
    # on_pong(wsapp, frame_data)
    #
    # on_error:
    # 当事件循环中有任何异常抛出, 将调用此函数, 
    # 然后关闭 WebSocket 连接, 调用 on_close(),
    # 然后 run_forever() 将被返回。
    # 即调用 on_error 后必定会调用 on_close。
    # on_xxx 回调方法中抛出的异常将被捕获, 不会抛给 run_forever()
    # 函数格式: 
    # on_error(wsapp, exception)
    #
    # on_close:
    # WebSocket 被关闭时调用的函数。
    # 事件循环中抛出异常 或 收到服务端推送的 CLOSE 数据帧关闭客户端时将被调用。
    # 主动调用 wsapp.close() 不会调用 on_close()
    # 函数格式: 
    # on_close(wsapp, close_status_code, close_reason)
    #
    # get_mask_key:
    # 自定义 mask key 的生成函数, 参考 WebSocketApp
    #
    # subprotocols:
    # 可用的子协议数组, 参考 WebSocketApp

`WebSocketApp` 的主要方法：

*  `run_forever()`
 *  `send()`
 *  `close()`

# 运行 WebSocketApp 的事件循环, 先创建 WebSocket 对象, 
    # 然后 connect 连接服务器, 之后一直循环运行接收数据帧, 回调对应函数处理数据帧。
    # 当 WebSocket 客户端被关闭后, 将调用 on_close() 方法, 然后结束循环返回。
    # 当循环中引发异常被捕获, 将依次调用 on_error()、on_close() 方法, 然后结束循环返回。
    #
    #
    # 参数说明:
    # sockopt: socket.setsockopt 的可选参数值, tuple 类型。
    # sslopt: SSL Socket 的可选项参数, 可选项 dict 对象。
    # 例如禁用 SSL 证书校验: sslopt={"cert_reqs": ssl.CERT_NONE}
    # 禁用域名校验: sslopt={"check_hostname": False}
    #
    # ping_interval: 自动发送 PING 数据帧的间隔秒数, 默认为 0 表示不自动发送 PING
    # ping_timeout: PING 发送后接收 PONG 的超时秒数, 如果超时未收到 PONG, 
    # 则抛出异常并调用 on_error()、on_close(), 然后结束循环。
    # None 表示不检测超时。
    # ping_payload: 发送 PING 数据帧时携带的数据, 默认为空字符串
    #
    # proxy_type: 代理的类型, 默认为 "http", 其他类型 'socks4', 'socks5', 'socks5h'
    # http_proxy_host: 代理的 host
    # http_proxy_port: 代理的 port, 默认为 80
    # http_no_proxy: 不使用代理的 host names, list 类型, 如: ["host1", "host2", ...]
    # http_proxy_auth: HTTP 代理认证, tuple 类型, 如: ("username", "password")
    #
    # skip_utf8_validation: 是否跳过 utf8 校验。
    # host: str, 自定义 HTTP 请求头 Host 的值。
    # origin: str, 自定义 HTTP 请求头 Origin 的值(URL)。
    # dispatcher: 自定义从 socket 读取数据的函数
    # suppress_origin: bool, 抑制输出 origin header。
    #
    run_forever(sockopt=None, 
                sslopt=None,
                ping_interval=0, 
                ping_timeout=None,
                ping_payload="",
                proxy_type=None,
                http_proxy_host=None, 
                http_proxy_port=None,
                http_no_proxy=None,,
                http_proxy_auth=None
                skip_utf8_validation=False,
                host=None, 
                origin=None, 
                dispatcher=None,
                suppress_origin=False)
    
    
    # 发送数据
    #
    # payload: 发送的数据(载荷), str/bytes 类型
    #
    # opcode: 操作码, 可选值:
    # ABNF.OPCODE_CONT
    # ABNF.OPCODE_TEXT
    # ABNF.OPCODE_BINARY
    # ABNF.OPCODE_CLOSE
    # ABNF.OPCODE_PING
    # ABNF.OPCODE_PONG
    #
    # 默认为发送 TEXT 数据帧
    #
    send(data, opcode=ABNF.OPCODE_TEXT)
    
    
    # 关闭 WebSocket, 先发送关闭帧再关闭底层 socket。
    #
    # status: 发送关闭数据帧时的数据(告诉服务端关闭的状态码), 默认为正常关闭。
    # reason: 发送关闭数据帧时的数据(告诉服务端关闭的原因)。
    # timeout: 直到超时还没有收到一个关闭帧, 则不再等待, 直接关闭底层 socket。
    # 正常为收到关闭帧后再关闭底层 socket。
    # int/float 秒数, None 表示一直等待直到收到一个关闭帧。
    #
    close(status=STATUS_NORMAL, reason=bytes('', encoding='utf-8'), timeout=3.0)

### 1.2.2 `WebSocketApp` 类的 代码示例 ###

import websocket
    import time
    import threading
    
    
    def on_open(wsapp):
        print("on_open")
    
        def send_message():
            for i in range(5):
                wsapp.send(f"Hello { i}")
                time.sleep(1)
    
        threading.Thread(target=send_message).start()
    
    
    def on_data(wsapp, frame_data, frame_opcode, frame_fin):
        print("on_data", frame_data, frame_opcode, frame_fin, sep=", ")
    
    
    def on_message(wsapp, data):
        print("on_message", data, sep=", ")
    
    
    def on_cont_message(wsapp, frame_data, frame_fin):
        print("on_cont_message", frame_data, frame_fin, sep=", ")
    
    
    def on_ping(wsapp, frame_data):
        print("on_ping", frame_data, sep=", ")
        # 接收到 PING 数据帧后, 需要立即给服务端回复 PONG 数据帧
        wsapp.send("", websocket.ABNF.OPCODE_PONG)
    
    
    def on_pong(wsapp, frame_data):
        print("on_pong", frame_data, sep=", ")
    
    
    def on_error(wsapp, e):
        print("on_error", e, sep=", ")
    
    
    def on_close(wsapp, close_status_code, close_reason):
        print("on_close", close_status_code, close_reason, sep=", ")
    
    
    wsapp = websocket.WebSocketApp("ws://echo.websocket.org",
                                   on_open=on_open,
                                   on_data=on_data,
                                   on_message=on_message,
                                   on_cont_message=on_cont_message,
                                   on_ping=on_ping,
                                   on_pong=on_pong,
                                   on_error=on_error,
                                   on_close=on_close)
    
    wsapp.run_forever(ping_interval=5, ping_timeout=2)

# 2. 异步 websocket 客户端/服务端 aiohttp 模块 #

`aiohttp` 是一个基于 asyncio 异步的 HTTP 客户端 和 服务端，并且支持 WebSockets 客户端 和 服务端，这里只介绍 WebSockets。

相关链接：

*  用户文档: [https://docs.aiohttp.org/][https_docs.aiohttp.org]
 *  PyPI 项目地址: [https://pypi.org/project/aiohttp/][https_pypi.org_project_aiohttp]
 *  GitHub 仓库: [https://github.com/aio-libs/aiohttp][https_github.com_aio-libs_aiohttp]

安装 `aiohttp` 模块：

# (必选) 安装 aiohttp 模块
    $ pip3 install aiohttp
    
    # (可选) 安装
    $ pip3 install cchardet     # 使用更快的 cchardet 库作为 chardet 的替代品
    $ pip3 install aiodns       # 为了通过客户端 API 加速 DNS 解析

## 2.1 aiohttp 的 HTTP 请求 ##

import aiohttp
    import asyncio
    
    
    async def main():
        # 创建 HTTP 会话客户端
        session = aiohttp.ClientSession()
    
        # 异步请求, 经过 async with 异步处理后在 __aenter__() 方法中返回响应对象 response
        async with session.get("https://httpbin.org/get", verify_ssl=False) as response:
            print(type(response))           # <class 'aiohttp.client_reqrep.ClientResponse'>
            print(response.version)         # HttpVersion(major=1, minor=1)
            print(response.status)          # 200
            print(response.reason)          # OK
            print(response.headers)         # <CIMultiDictProxy('Content-Type': 'application/json'...
            print(response.content)         # <StreamReader 316 bytes eof>,
                                            # content 是 Body 的 IO 流对象, 可以通过 await read(n) 方法逐块读取响应
            body = await response.text()    # 异步读取响应并解码为文本
            print(body)
    
        # 异步关闭客户端 (session 也支持 async with)
        await session.close()
    
    
    asyncio.run(main())

## 2.2 aiohttp WebSocket Server ##

使用 `aiohttp` 实现一个用于 echo 回显的 WebSocket 服务端（顺便实现一个 HTTP Server）：

from aiohttp import web
    from aiohttp.web_request import Request
    from aiohttp.web_response import Response
    from aiohttp.web_ws import WebSocketResponse
    
    
    async def http_handle(request: Request) -> Response:
        """ HTTP 请求处理器 """
        # web.get("/{name}", ...) 中的 name
        name = request.match_info.get("name")
    
        # 请求 Body, 也可以通过 await request.content.read(n) 逐块读取
        body = await request.text()
    
        resp_text = f"Your Request Info:\n" \
                    f"name: { name}\n" \
                    f"IP: { request.remote}\n" \
                    f"Method: { request.method}\n" \
                    f"Version: { request.version}\n" \
                    f"Url: { request.url}\n" \
                    f"Path: { request.path}\n" \
                    f"Headers: { request.headers}\n" \
                    f"Body: { body}"
    
        # 返回一个 响应对象
        return web.Response(status=200,
                            text=resp_text,
                            content_type="text/plain")
    
    
    async def ws_handle(request: Request) -> WebSocketResponse:
        """ WebSocket 请求处理器 """
        # 创建一个 WebSocket 响应, 自动响应 CLOSE, 收到 PING 后自动回复 PONG
        ws = web.WebSocketResponse(autoclose=True, autoping=True)
        # 预处理请求
        await ws.prepare(request)
    
        # 循环处理消息, 直到 WebSocket 退出
        # ws.__anext__() 方法中调用了 await ws.receive() 接收消息并返回
        async for msg in ws:
            # msg: <class 'aiohttp.http_websocket.WSMessage'>
            if msg.type == web.WSMsgType.TEXT:
                await ws.send_str(msg.data)
            elif msg.type == web.WSMsgType.BINARY:
                await ws.send_bytes(msg.data)
            elif msg.type == web.WSMsgType.CLOSE:
                break
    
        print("websocket connection closed.")
        return ws
    
    
    # 创建一个 Web 应用
    app = web.Application()
    
    # 添加路由路径对应的处理器, 按列表顺序依次匹配
    app.add_routes([web.get("/", http_handle),          # curl http://localhost:8080/
                    web.post("/", http_handle),         # curl -d "aa=bb" http://localhost:8080/
                    web.get("/echo", ws_handle),        # ws://localhost:8080/echo
                    web.get("/{name}", http_handle)])   # curl http://localhost:8080/hello
    
    # 运行一个 Web 应用,
    # 这里不需要在异步方法中运行, 内部会在 asyncio 异步事件循环中处理
    web.run_app(app, host=None, port=8080)

## 2.3 aiohttp WebSocket Client ##

import asyncio
    import aiohttp
    from aiohttp import web
    from aiohttp.client_ws import ClientWebSocketResponse
    import random
    
    
    async def message_handler(ws: ClientWebSocketResponse):
        while True:
            # msg: <class 'aiohttp.http_websocket.WSMessage'>
            msg = await ws.receive()
            if msg.type == web.WSMsgType.TEXT:
                print("RECV:", msg.data)
            elif msg.type == web.WSMsgType.BINARY:
                print("RECV:", msg.data)
            elif msg.type == web.WSMsgType.PING:
                print("RECV: PING")
                await ws.pong()     # 收到 PING 后给服务端回复 PONG
            elif msg.type == web.WSMsgType.PONG:
                print("RECV: PONG")
            elif msg.type == web.WSMsgType.CLOSE:
                print("RECV: CLOSE")
            else:
                print("RECV:", msg)
    
    
    async def main():
        async with aiohttp.ClientSession() as session:
            # 链接 WebSocket 服务器, 不自动处理 PING
            async with session.ws_connect("ws://localhost:8080/echo", autoping=False) as ws:
                # ws: <class 'aiohttp.client_ws.ClientWebSocketResponse'>
    
                # 异步 循环接收消息
                recv_task = asyncio.create_task(message_handler(ws))
    
                while True:
                    if ws.closed:
                        break
    
                    # 发送文本
                    text = f"Hello: { random.randint(100, 999)}"
                    print("SEND:", text)
                    await ws.send_str(text)
    
                    await asyncio.sleep(2)
    
                    # 发送二进制
                    bs = f"World: { random.randint(100, 999)}".encode()
                    print("SEND:", bs)
                    await ws.send_bytes(bs)
    
                    await asyncio.sleep(2)
    
                    # 发送 PING 数据帧
                    print("SEND: PING")
                    await ws.ping()
    
                    await asyncio.sleep(2)
    
                await recv_task
    
    
    asyncio.run(main())

# 3. 异步 websockets 模块 #

`websockets` 是用于在 Python 中构建 WebSocket 客户端 和 服务端 的一个库。它建立在 Python 的标准异步 I/O 框架 `asyncio` 之上，提供了一个优雅的基于协程的 API。

`websockets` 需要 Python 3.6.1+。