翻译自https://websockets.readthedocs.io/en/stable/intro.html

要求

websockets 库要求Python版本 ≥ 3.6.1。

如果可能的话，您应该使用最新的python版本.。如果您使用的是旧版本，请注意，对于每个次要版本（3.x），只有最新的bugfix版本（3.x.y）才得到官方支持。

安装

用以下命令安装websockets

pip install websockets

基本例子

以下是一个websocket的服务端

它从客户端读取一个名称，发送一个问候语，然后关闭连接。

#!/usr/bin/env python# WS server exampleimport asyncio
import websocketsasync def hello(websocket, path):name = await websocket.recv()print(f"< {name}")greeting = f"Hello {name}!"await websocket.send(greeting)print(f"> {greeting}")start_server = websockets.serve(hello, "localhost", 8765)asyncio.get_event_loop().run_until_complete(start_server)
asyncio.get_event_loop().run_forever()

在服务器端，websockets为每个WebSocket连接执行一次处理程序coroutine hello。当处理程序协程返回时，它将关闭连接。

下面是一个对应的WebSocket客户端示例。

#!/usr/bin/env python# WS client exampleimport asyncio
import websocketsasync def hello():uri = "ws://localhost:8765"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())

使用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 certificateimport asyncio
import pathlib
import ssl
import websocketsasync def hello(websocket, path):name = await websocket.recv()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
)asyncio.get_event_loop().run_until_complete(start_server)
asyncio.get_event_loop().run_forever()

调整后的客户端

#!/usr/bin/env python# WSS (WS over TLS) client example, with a self-signed certificateimport asyncio
import pathlib
import ssl
import websocketsssl_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())

此客户端需要上下文，因为服务器使用自签名证书。

使用有效证书（即由Python安装信任的CA签名）连接到安全WebSocket服务器的客户机只需将ssl=True传递给connect（）而不是构建上下文。

基于浏览器的示例

下面是一个如何运行WebSocket服务器并从浏览器连接的示例。

在控制台中运行此脚本：

#!/usr/bin/env python# WS server that sends messages at random intervalsimport asyncio
import datetime
import random
import websocketsasync 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)asyncio.get_event_loop().run_until_complete(start_server)
asyncio.get_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 = document.createElement('ul');ws.onmessage = function (event) {var messages = document.getElementsByTagName('ul')[0],message = document.createElement('li'),content = document.createTextNode(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 clientsimport asyncio
import json
import logging
import websocketslogging.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 listmessage = 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 listmessage = 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):USERS.remove(websocket)await notify_users()async def counter(websocket, path):# register(websocket) sends user_event() to websocketawait register(websocket)try:await websocket.send(state_event())async for message in websocket:data = json.loads(message)if data["action"] == "minus":STATE["value"] -= 1await notify_state()elif data["action"] == "plus":STATE["value"] += 1await notify_state()else:logging.error("unsupported event: {}", data)finally:await unregister(websocket)start_server = websockets.serve(counter, "localhost", 6789)asyncio.get_event_loop().run_until_complete(start_server)
asyncio.get_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;}.plus {color: green;}.value {min-width: 2em;}.state {font-size: 2em;}</style></head><body><div class="buttons"><div class="minus button">-</div><div class="value">?</div><div class="plus button">+</div></div><div class="state"><span class="users">?</span> online</div><script>var minus = document.querySelector('.minus'),plus = document.querySelector('.plus'),value = document.querySelector('.value'),users = document.querySelector('.users'),websocket = new WebSocket("ws://127.0.0.1:6789/");minus.onclick = function (event) {websocket.send(JSON.stringify({action: 'minus'}));}plus.onclick = function (event) {websocket.send(JSON.stringify({action: 'plus'}));}websocket.onmessage = function (event) {data = JSON.parse(event.data);switch (data.type) {case 'state':value.textContent = data.value;break;case 'users':users.textContent = (data.count.toString() + " user" +(data.count == 1 ? "" : "s"));break;default:console.error("unsupported event", data);}};</script></body>
</html>

常见模式

您通常希望在连接的生命周期中处理多条消息。所以你必须写一个循环。下面是构建WebSocket服务器的基本模式。

客户consumer

用于接收消息并将其传递给客户协同程序：

async def consumer_handler(websocket, path):async for message in websocket:await consumer(message)

在本例中，consumer表示用于处理在WebSocket连接上接收的消息的业务逻辑。

当客户端断开连接时，迭代终止。

服务端 Producer

从服务端联程获取消息并发送消息：

async def producer_handler(websocket, path):while True:message = await producer()await websocket.send(message)

在本例中，producer表示用于生成要在WebSocket连接上发送的消息的业务逻辑。

send（）在客户端断开连接时引发ConnectionClosed异常，该异常会中断while True循环。

我全要！

通过将上述两种模式组合起来并并行运行这两个任务，可以在同一连接上读写消息：

async def handler(websocket, path):consumer_task = asyncio.ensure_future(consumer_handler(websocket, path))producer_task = asyncio.ensure_future(producer_handler(websocket, path))done, pending = await asyncio.wait([consumer_task, producer_task],return_when=asyncio.FIRST_COMPLETED,)for task in pending:task.cancel()

注册Registration

如上面的同步示例所示，如果需要维护当前连接的客户端的列表，则必须在它们连接时注册它们，并在它们断开连接时注销它们。

connected = set()async def handler(websocket, path):# Register.connected.add(websocket)try:# Implement logic here.await asyncio.wait([ws.send("Hello!") for ws in connected])await asyncio.sleep(10)finally:# Unregister.connected.remove(websocket)

这个简单的例子跟踪内存中连接的客户机。这只在运行一个进程时有效。例如，在实际应用程序中，处理程序可以订阅消息代理上的某些通道。

这就是全部内容了

websockets API的设计是由简单性驱动的。

您不必担心执行打开或关闭握手、应答ping或规范要求的任何其他行为。

websockets在引擎盖下处理所有这些，所以你不必。

还有一件事…

websockets提供了一个交互式客户端：

$python-m websockets wss://echo.websocket.org/

快速上手python websockets相关推荐

简单python脚本实例-30个Python 小例子，帮你快速上手Python
可能很多人学编程有个误区,总想着从最基础的原理开始,看了一大堆书一堆视频还是不会,兴趣也早没了.而自己动手写代码是最好的开始方式,不管TM三七二十一,直接就是干,无论是看书还是看视频,确保身边有台电脑 ...
python入门要多久-初学者如何快速上手python入门要多久
今天跟大家讨论一下对于一个新手如何快速入门python. python是一种非常接近人类语言的解释性高级语言,如果你学过java,应该知道解释性语言区别于编译性语言.python作为一种高级语言,功能 ...
0基础入门，如何快速上手Python?
0基础入门,如何快速上手pythpn 新的改变因为清晰易读的风格,广泛的适用性,python已经成为最受欢迎的编程语言之一,在TIOBE排行榜位居第四,是名副其实的人工智能第一语言. python ...
机器学习算法快速上手-python语言与numpy库
1 Python快速上手 1.1.Python简介 Python 是一个高层次的结合了解释性.编译性.互动性和面向对象的脚本语言. Python 的设计具有很强的可读性,相比其他语言经常使用英文关键字 ...
python 编程快速上手,Python编程快速上手
**部分 Python编程基础 **章 Python基础 1.1 在交互式环境中输入表达式 1.2 整型.浮点型和字符串数据类型 1.3 字符串连接和复制 1.4 在变量中保存值 1.4.1 赋值语句 ...
「高效程序员的修炼」快速上手python主流测试框架pytest以及单元测试编写
如果对你有帮助,就点个赞吧~ 本文主要介绍如果编写Python的单元测试,包括如何使用断言,如何考虑测试哪些情况,如何避免外部依赖对测试的影响,如果用数据驱动的方式简化重复测试的编写等等等等文章目录 ...
心得丨如何快速上手疯狂上涨的Python？
纵观各大编程语言在 2017 年的发展情况,我们会发现涌现出诸如 Go.Swift 这类后起之秀,而其中最为耀眼的当属 Python. 快速上手 Python 有了一定的方向,接下来的问题及时如何快速 ...
Python编程快速上手，让繁琐工作自动化
大家好,我是辰哥(文末送书) 自学编程是不少人的选择,在著名的软件开发领域问答网站stackoverflow上,直接搜索learn programming可以得到47000多个回答,教我们如何学习某种 ...
Keras快速上手：基于Python的深度学习
Keras快速上手:基于Python的深度学习谢梁,鲁颖,劳虹岚著 ISBN:9787121318726 包装:平装开本:16开正文语种:中文出版社: 电子工业出版社出版时间:2017-0 ...
Python学习六大路线，教你快速上手
最近几年随着互联网的发展学习Python人越来越多,Python的初学者总希望能够得到一份Python学习路线图,小编经过多方面汇总,总结出比较全套Python学习路线,快速上手.对于一个零基础的想学 ...

快速上手python websockets

要求

安装