使用Python Websockets库建立WebSocket客户端链接

说明：

这里是一个比较好用的WebSocket测试网址，即便使用其他语言实现的WebSocket也可以使用该地址进行测试。
http://www.websocket-test.com/

本文的部分例子可在此找到，https://websockets.readthedocs.io/en/stable/

1. 安装Websockets

$ sudo pip3 install websockets

2. 简单示例

#!/usr/bin/env python3

# WS client example

import asyncio
import websockets

async def hello():
    uri = "ws://121.40.165.18:8800"
    async with websockets.connect(uri) as websocket:
        name = input("What's your name? ")

        await websocket.send(name)
        print(f"> {name}")

        greeting = await websocket.recv()
        print(f"< {greeting}")

asyncio.get_event_loop().run_until_complete(hello())

使用with语句进行connect连接后的上下文自动管理，当hello协程退出时，自动关闭该WebSocket连接。

3. 带安全认证的示例

#!/usr/bin/env python

# WSS (WS over TLS) client example, with a self-signed certificate

import asyncio
import pathlib
import ssl
import websockets

ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
localhost_pem = pathlib.Path(__file__).with_name("localhost.pem")
ssl_context.load_verify_locations(localhost_pem)

async def hello():
    uri = "wss://localhost:8765"
    async with websockets.connect(
        uri, ssl=ssl_context
    ) as websocket:
        name = input("What's your name? ")

        await websocket.send(name)
        print(f"> {name}")

        greeting = await websocket.recv()
        print(f"< {greeting}")

asyncio.get_event_loop().run_until_complete(hello())

不同于基本示例的是，connect函数中需要指定ssl参数的内容

4. 长连接示例以及不使用with语句进行上下文(context)自动管理

import asyncio
import websockets

cmd = "maybe login command"

async permanant_conn():
  wss_url = 'ws://localhost:8888'
  conn_handler = await websockets.connect(wss_url)
  await conn_handler.send(wss_url)
  while True:
    response = await conn_handler.recv()
    ## Your process code is here

asyncio.get_event_loop().run_until_complete(permanant_conn)

可以看到与基本示例不同的地方在于：1.不再使用with进行自动的上下文管理；2.使用while语句进行长连接的处理，如果需要能够自己控制长连接的关闭，那么可以设置一个变量比如conn_flag，初始化其值为True，当收到退出连接的命令时，将conn_flag置为False，即可实现连接的可控退出。
Enjoy it！