快速上⼿pythonwebsockets
要求
websockets 库要求Python版本 ≥ 3.6.1。
如果可能的话，您应该使⽤最新的python版本.。如果您使⽤的是旧版本，请注意，对于每个次要版本（3.x），只有最新的bugfix版本（3.x.y）才得到官⽅⽀持。
安装
⽤以下命令安装websockets
pip install websockets
基本例⼦
以下是⼀个websocket的服务端
它从客户端读取⼀个名称，发送⼀个问候语，然后关闭连接。
#!/usr/bin/env python
# WS server example
import asyncio
import websockets
async def hello(websocket, path):
name =v()
print(f"< {name}")
greeting = f"Hello {name}!"
await websocket.send(greeting)
print(f"> {greeting}")
start_server = websockets.serve(hello,"localhost",8765)
<_event_loop().run_until_complete(start_server)
<_event_loop().run_forever()
在服务器端，websockets为每个WebSocket连接执⾏⼀次处理程序coroutine hello。当处理程序协程返回时，它将关闭连接。
下⾯是⼀个对应的WebSocket客户端⽰例。
# WS client example
import asyncio
import websockets
async def hello():
uri ="ws://localhost:8765"
async t(uri)as websocket:
name =input("What's your name? ")
await websocket.send(name)
print(f"> {name}")
greeting =v()
print(f"< {greeting}")
<_event_loop().run_until_complete(hello())
使⽤connect()函数作为异步上下⽂管理器可以确保在退出hello协程之前关闭连接。
安全样例
安全的WebSocket连接提⾼了机密性和可靠性，因为它们减少了坏代理的⼲扰风险。
WSS协议对于WS就像HTTPS对于HTTP⼀样：连接是⽤传输层安全（Transport Layer Security，TLS）加密的，传输层安全（Transport Layer Security，TLS）通常被称为安全套接字层（Secure Sockets Layer，SSL）。WSS需要像HTTPS这样的TLS证书。
下⾯介绍如何调整服务器⽰例以提供安全连接。有关安全配置上下⽂的信息，请参阅ssl模块的⽂档。
#!/usr/bin/env python
# WSS (WS over TLS) server example, with a self-signed certificate
import asyncio
import pathlib
import ssl
import websockets
async def hello(websocket, path):
name =v()
print(f"< {name}")
greeting = f"Hello {name}!"
await websocket.send(greeting)
print(f"> {greeting}")
ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
localhost_pem = pathlib.Path(__file__).with_name("localhost.pem")
ssl_context.load_cert_chain(localhost_pem)
start_server = websockets.serve(
hello,"localhost",8765, ssl=ssl_context
)
<_event_loop().run_until_complete(start_server)
<_event_loop().run_forever()
调整后的客户端
# WSS (WS over TLS) client example, with a self-signed certificate
notify for mi bandimport 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 t(
uri, ssl=ssl_context
)as websocket:
name =input("What's your name? ")
await websocket.send(name)
print(f"> {name}")
greeting =v()
print(f"< {greeting}")
<_event_loop().run_until_complete(hello())
此客户端需要上下⽂，因为服务器使⽤⾃签名证书。
使⽤有效证书（即由Python安装信任的CA签名）连接到安全WebSocket服务器的客户机只需将ssl=True传递给connect（）⽽不是构建上下⽂。
基于浏览器的⽰例
下⾯是⼀个如何运⾏WebSocket服务器并从浏览器连接的⽰例。
在控制台中运⾏此脚本：
#!/usr/bin/env python
# WS server that sends messages at random intervals
import asyncio
import datetime
import random
import websockets
async def time(websocket, path):
while True:
now = datetime.datetime.utcnow().isoformat()+"Z"
await websocket.send(now)
await asyncio.sleep(random.random()*3)
start_server = websockets.serve(time,"127.0.0.1",5678)
<_event_loop().run_until_complete(start_server)
<_event_loop().run_forever()
在浏览器打开该HTML⽂件
<!DOCTYPE html>
<html>
<head>
<title>WebSocket demo</title>
</head>
<body>
<script>
var ws = new WebSocket("ws://127.0.0.1:5678/"),
messages = ateElement('ul');
var messages = ElementsByTagName('ul')[0],
message = ateElement('li'),
content = ateTextNode(event.data);
message.appendChild(content);
messages.appendChild(message);
};
document.body.appendChild(messages);
</script>
</body>
</html>
同步⽰例
WebSocket服务器可以从客户端接收事件，处理它们以更新应⽤程序状态，并跨客户端同步结果状态。下⾯是⼀个⽰例，任何客户端都可以增加或减少计数器。更新将传播到所有连接的客户端。
异步的并发模型保证更新是序列化的。
在控制台中运⾏此脚本：
#!/usr/bin/env python
# WS server example that synchronizes state across clients
import asyncio
import json
import logging
import websockets
logging.basicConfig()
STATE ={"value":0}
USERS =set()
def state_event():
return json.dumps({"type":"state",**STATE})
def users_event():
return json.dumps({"type":"users","count":len(USERS)})
async def notify_state():
if USERS:# asyncio.wait doesn't accept an empty list
message = state_event()
await asyncio.wait([user.send(message)for user in USERS])
async def notify_users():
if USERS:# asyncio.wait doesn't accept an empty list
message = users_event()
await asyncio.wait([user.send(message)for user in USERS])
async def register(websocket):
USERS.add(websocket)
await notify_users()
async def unregister(websocket):
await notify_users()
async def counter(websocket, path):
# register(websocket) sends user_event() to websocket await register(websocket)
try:
await websocket.send(state_event())
async for message in websocket:
data = json.loads(message)
if data["action"]=="minus":
STATE["value"]-=1
await notify_state()
elif data["action"]=="plus":
STATE["value"]+=1
await notify_state()
else:
<("unsupported event: {}", data) finally:
await unregister(websocket)
start_server = websockets.serve(counter,"localhost",6789)
<_event_loop().run_until_complete(start_server) _event_loop().run_forever()
然后在⼏个浏览器中打开这个HTML⽂件。
<!DOCTYPE html>
<html>
<head>
<title>WebSocket demo</title>
<style type="text/css">
body {
font-family:"Courier New", sans-serif;
text-align: center;
}
.buttons {
font-size: 4em;
display: flex;
justify-content: center;
}
.button,.value {
line-height:1;
padding: 2rem;
margin: 2rem;
border: medium solid;
min-height: 1em;
min-width: 1em;
}
.button {
cursor: pointer;
user-select: none;
}
.
minus {
color: red;

版权声明：本站内容均来自互联网，仅供演示用，请勿用于商业和其他非法用途。如果侵犯了您的权益请与我们联系QQ:729038198，我们将在24小时内删除。