python的websocket方法教程

WebSocket是一种网络通信协议,它在单个TCP连接上提供全双工的通信信道。在本篇文章中,我们将探讨如何在Python中使用WebSocket实现实时通信。

websockets是Python中最常用的网络库之一,也是websocket协议的Python实现。它不仅作为基础组件在众多项目中发挥着重要作用,其源码也值得广大“Python玩家”研究。
官网:https://github.com/python-websockets/websockets

1. 什么是WebSocket?

WebSocket协议是在2008年由Web应用程序设计师和开发人员创建的,目的是为了在Web浏览器和服务器之间提供更高效、更低延迟的双向通信。它允许客户端和服务器在任何时候发送消息,无需重新建立TCP连接。WebSocket可以在Web浏览器和服务器之间传输文本和二进制数据,使得构建实时Web应用程序变得更加简单。

2. 在Python中使用WebSocket

Python中有多个库可以帮助我们使用WebSocket,如:websockets、aiohttp等。在本文中,我们将使用websockets库来演示WebSocket编程。

要安装websockets库,你可以使用pip:

pip install websockets

3. 创建WebSocket服务器

使用websockets库,我们可以轻松地创建一个WebSocket服务器。以下是一个简单的示例:

import asyncio
import websocketsasync def echo(websocket, path):async for message in websocket:print(f"Received message: {message}")await websocket.send(f"Echo: {message}")start_server = websockets.serve(echo, "localhost", 8765)asyncio.get_event_loop().run_until_complete(start_server)
asyncio.get_event_loop().run_forever()

在这个示例中,我们定义了一个名为echo的协程函数,它接收两个参数:websocket和path。该函数使用async for循环读取客户端发送的消息,并将消息发送回客户端。

然后,我们使用websockets.serve()函数创建一个WebSocket服务器,监听本地主机的8765端口。最后,我们使用asyncio的事件循环启动服务器。

4. 创建WebSocket客户端

要创建一个WebSocket客户端,我们同样可以使用websockets库。以下是一个简单的客户端示例:

import asyncio
import websocketsasync def main():async with websockets.connect("ws://localhost:8765") as websocket:message = "Hello, server!"await websocket.send(message)print(f"Sent: {message}")response = await websocket.recv()print(f"Received: {response}")asyncio.run(main())

在这个示例中,我们使用websockets.connect()函数建立与WebSocket服务器的连接。然后,我们使用send()方法向服务器发送消息,并使用recv()方法接收服务器的响应。

5. 总结

WebSocket协议为Web浏览器和服务器之间提供了实时双向通信的能力,使得构建实时Web应用程序变得更加容易。在Python中,我们可以使用websockets库轻松地实现WebSocket编程。

6. 通过websockets这个项目,从大型开源项目中学习asyncio库。

一、asyncio.Transport
在官方文档中,Transport被描述成对socket的抽象,它控制着如何传输数据。除了websockets,uvicorn、daphne等ASGI实现都会用到Transport。

Transport继承于ReadTransport和WriteTransport,两者都继承于BaseTransport。顾名思义,Transport兼备读和写的功能,可以类比为读写socket对象。
在这里插入图片描述

Transport对象提供以下常用函数——

is_reading:判断该Transport是否在读。

set_write_buffer_limits:设置写入Transport的高和低水位。考虑到网络状况,有时不希望写入过多的数据。

write、write_eof、write_line:为当前Transport写入数据,分别表示写入二进制数据、eof和二进制行数据。其中eof写入后不会关闭Transport,但会flush数据。

abort:立刻关闭Transport,不接受新的数据。留在缓冲的数据也会丢失,后续调用Protocol的connection_lost函数。

在websockets中,Transport使用场景不多,一般都是通过Protocol对象的回调参数使用的。在websocket的初始化过程中,会设置Transport的最高水位。同样,在这种场景下,该对象也是作为回调参数使用的。
在这里插入图片描述

二、asyncio.Protocol
如果Transport是对socket的抽象,那么Protocol就是对协议的抽象。它提供了如何使用Transport的方式。
在这里插入图片描述

用户使用的Protocol直接继承自BaseProtocol,并提供了六个Unimplemented函数需要用户去实现——

connection_made:当连接建立时会执行该函数,该函数包含一个Transport类型的参数。

connection_lost:当连接丢失或者关闭时会执行该函数,该函数包含一个Exception类型的参数。

pause_writing:当Transport对象写入的数据高于之前设置的高水位时被调用,一般会暂停数据的写入。

resume_writing:当Transport对象写入的数据低于之前设置的低水位时被调用,一般用于恢复数据写入。

data_received:当有数据被接受时回调,该函数包含一个二进制对象data,用来表示接受的数据。

eof_received:当被Transport对象被调用write_eof时被调用。

在websockets中,server端的connection_made实现截图如图所示。在该函数中,websockets将用户实现的handler封装成task对象,并和websocket的server绑定。
在这里插入图片描述

而在client端中实现如第一节截图所示,只是在reader中注册该Transport对象。

websockets的connection_lost函数实现方式如下。主要操作即更新状态、关闭pings、更新对应的waiter状态,以及维护reader对象。
在这里插入图片描述

在其他函数的实现中,websockets也主要用到了reader对象完成数据流的暂停和恢复,以及数据的写入。

从上面代码实现可以看出,websockets通过reader代理完成数据流的操作。这个reader是一个asyncio.StreamReader对象。这个对象具体如何使用将在下一篇介绍。

附录:进阶版本:

python使用websockets库
serve:在server端使用,等待客户端的连接。如果连接成功,返回一个websocket。

connect: 在client端使用,用于建立连接。

send:发送数据

recv:接收数据

close:关闭连接

服务端

#!/usr/bin/python3
# 主要功能:创建1个基本的websocket server, 符合asyncio 开发要求
import asyncio
import websockets
from datetime import datetimeasync def handler(websocket):data = await websocket.recv()reply = f"Data received as \"{data}\".  time: {datetime.now()}"print(reply)await websocket.send(reply)print("Send reply")async def main():async with websockets.serve(handler, "localhost", 9999):await asyncio.Future()  # run foreverif __name__ == "__main__":asyncio.run(main())

客户端

import asyncio
import websockets
import timeasync def ws_client(url):for i in range(1, 40):async with websockets.connect(url) as websocket:await websocket.send("Hello, I am PyPy.")response = await websocket.recv()print(response)time.sleep(1)asyncio.run(ws_client('ws://localhost:9999'))

服务端

import asyncio
import websocketsIP_ADDR = "127.0.0.1"
IP_PORT = "9090"# 握手,通过接收Hi,发送"success"来进行双方的握手。
async def serverHands(websocket):while True:recv_text = await websocket.recv()print("recv_text=" + recv_text)if recv_text == "Hi":print("connected success")await websocket.send("success")return Trueelse:await websocket.send("connected fail")# 接收从客户端发来的消息并处理,再返给客户端success
async def serverRecv(websocket):while True:recv_text = await websocket.recv()print("recv:", recv_text)await websocket.send("success,get mess:"+ recv_text)# 握手并且接收数据
async def serverRun(websocket, path):print(path)await serverHands(websocket)await serverRecv(websocket)# main function
if __name__ == '__main__':print("======server======")server = websockets.serve(serverRun, IP_ADDR, IP_PORT)asyncio.get_event_loop().run_until_complete(server)asyncio.get_event_loop().run_forever()

客户端

import asyncio
import websocketsIP_ADDR = "127.0.0.1"
IP_PORT = "9090"async def clientHands(websocket):while True:# 通过发送hello握手await websocket.send("Hi")response_str = await websocket.recv()# 接收"success"来进行双方的握手if "success" in response_str:print("握手成功")return True# 向服务器端发送消息
async def clientSend(websocket):while True:input_text = input("input text: ")if input_text == "exit":print(f'"exit", bye!')await websocket.close(reason="exit")return Falseawait websocket.send(input_text)recv_text = await websocket.recv()print(f"{recv_text}")# 进行websocket连接
async def clientRun():ipaddress = IP_ADDR + ":" + IP_PORTasync with websockets.connect("ws://" + ipaddress) as websocket:await clientHands(websocket)await clientSend(websocket)# main function
if __name__ == '__main__':print("======client======")asyncio.get_event_loop().run_until_complete(clientRun())

服务端

# -*- coding:utf8 -*-import json
import socket
import asyncio
import logging
import websockets
import multiprocessingIP = '127.0.0.1'
PORT_CHAT = 9090USERS ={}#提供聊天的后台
async def ServerWs(websocket,path):logging.basicConfig(format='%(asctime)s - %(pathname)s[line:%(lineno)d] - %(levelname)s: %(message)s',filename="chat.log",level=logging.INFO)# 握手await websocket.send(json.dumps({"type": "handshake"}))async for message in websocket:data = json.loads(message)message = ''# 用户发信息if data["type"] == 'send':name = '404'for k, v in USERS.items():if v == websocket:name = kdata["from"] = nameif len(USERS) != 0:  # asyncio.wait doesn't accept an empty listmessage = json.dumps({"type": "user", "content": data["content"], "from": name})# 用户注册elif data["type"] == 'register':try:USERS[data["uuid"]] = websocketif len(USERS) != 0:  # asyncio.wait doesn't accept an empty listmessage = json.dumps({"type": "login", "content": data["content"], "user_list": list(USERS.keys())})except Exception as exp:print(exp)# 用户注销elif data["type"] == 'unregister':del USERS[data["uuid"]]if len(USERS) != 0:  # asyncio.wait doesn't accept an empty listmessage = json.dumps({"type": "logout", "content": data["content"], "user_list": list(USERS.keys())})#打印日志logging.info(data)# 群发await asyncio.wait([user.send(message) for user in USERS.values()])def server_run():print("server")start_server = websockets.serve(ServerWs, '0.0.0.0', PORT_CHAT)asyncio.get_event_loop().run_until_complete(start_server)asyncio.get_event_loop().run_forever()if __name__ == "__main__":from multiprocessing import Processmultiprocessing.freeze_support()server = Process(target=server_run, daemon=False)server.start()

服务端

import asyncio
import websockets
import time
import json
import threading
# 功能模块
class OutputHandler():async def run(self,message,send_ms,websocket):# 用户发信息await send_ms(message, websocket)# 单发消息# await send_ms(message, websocket)# 群发消息#await s('hi起来')# 存储所有的客户端
Clients = {}# 服务端
class WS_Server():def __init__(self):self.ip = "127.0.0.1"self.port = 9090# 回调函数(发消息给客户端)async def callback_send(self, msg, websocket=None):await self.sendMsg(msg, websocket)# 发送消息async def sendMsg(self, msg, websocket):print('sendMsg:', msg)# websocket不为空,单发,为空,群发消息if websocket != None:await websocket.send(msg)else:# 群发消息await self.broadcastMsg(msg)# 避免被卡线程await asyncio.sleep(0.2)# 群发消息async def broadcastMsg(self, msg):for user in Clients:await user.send(msg)# 针对不同的信息进行请求,可以考虑json文本async def runCaseX(self,jsonMsg,websocket):print('runCase')op = OutputHandler()# 参数:消息、方法、socketawait op.run(jsonMsg,self.callback_send,websocket)# 连接一个客户端,起一个循环监听async def echo(self,websocket, path):# 添加到客户端列表# Clients.append(websocket)# 握手await websocket.send(json.dumps({"type": "handshake"}))# 循环监听while True:# 接受信息try:# 接受文本recv_text = await websocket.recv()message = "Get message: {}".format(recv_text)# 返回客户端信息await websocket.send(message)# 转jsondata = json.loads(recv_text)# 用户发信息if data["type"] == 'send':name = '404'for k, v in Clients.items():if v == websocket:name = kdata["from"] = nameif len(Clients) != 0:  # asyncio.wait doesn't accept an empty listmessage = json.dumps({"type": "send", "content": data["content"], "from": name})await self.runCaseX(jsonMsg=message, websocket=websocket)# 用户注册elif data["type"] == 'register':try:Clients[data["uuid"]] = websocketif len(Clients) != 0:  # asyncio.wait doesn't accept an empty listmessage = json.dumps({"type": "register", "content": data["content"], "user_list": list(Clients.keys())})await self.runCaseX(jsonMsg=message, websocket=websocket)except Exception as exp:print(exp)# 用户注销elif data["type"] == 'unregister':del Clients[data["uuid"]]# 对message进行解析,跳进不同功能区# await self.runCaseX(jsonMsg=data,websocket=websocket)# 链接断开except websockets.ConnectionClosed:print("ConnectionClosed...", path)# del Clientsbreak# 无效状态except websockets.InvalidState:print("InvalidState...")# del Clientsbreak# 报错except Exception as e:print("ws连接报错",e)# del Clientsbreak# 启动服务器async def runServer(self):async with websockets.serve(self.echo, self.ip, self.port):await asyncio.Future()  # run forever# 多协程模式,防止阻塞主线程无法做其他事情def WebSocketServer(self):asyncio.run(self.runServer())# 多线程启动def startServer(self):# 多线程启动,否则会堵塞thread = threading.Thread(target=self.WebSocketServer)thread.start()# thread.join()if __name__=='__main__':print("server")s = WS_Server()s.startServer()

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.mzph.cn/news/208338.shtml

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

pyside/qt03——人机协同的编程教学—直接面向chatGPT实战开发(做中学,事上练)

先大概有个草图框架,一点点丰富 我纠结好久,直接用Python写UI代码 还是用designer做UI 再转Python呢, 因为不管怎么样都要转成Python代码, 想了想还是学一下designer吧,有个中介,有直观理解。 直接这样也可…

智能优化算法应用:基于食肉植物算法无线传感器网络(WSN)覆盖优化 - 附代码

智能优化算法应用:基于食肉植物算法无线传感器网络(WSN)覆盖优化 - 附代码 文章目录 智能优化算法应用:基于食肉植物算法无线传感器网络(WSN)覆盖优化 - 附代码1.无线传感网络节点模型2.覆盖数学模型及分析3.食肉植物算法4.实验参数设定5.算法结果6.参考…

设计并实现一个多线程图书馆管理系统,涉及数据库操作

没有实现全部功能,希望路过的大佬,可以实现全部功能,在评论区聊聊 创建数据库library-demo CREATE DATABASE library-demo创建图书表book CREATE TABLE book (bookId int(11) NOT NULL AUTO_INCREMENT COMMENT 图书ID,bookName varchar(15)…

HarmonyOS创建JavaScript(类 Web开发模式)项目

上文 HarmonyOS带大家创建自己的第一个Page页面并实现路由跳转(ArkTS)带大家创建了我们项目中第一个自己创建的page 并完成了一个跳转逻辑的编写 上文的开发模式是 ArkTS 的 也被称为 声明式开发范式 还有一种 javaScript的 类Web开发模式 这种方式就类似于我们传统的前端开发模…

基于微群机器人的二次开发

请求URL: http://域名地址/modifyGroupName 请求方式: POST 请求头Headers: Content-Type:application/jsonAuthorization:login接口返回 参数: 参数名必选类型说明wId是String登录实例标识chatRoom…

读书笔记-《数据结构与算法》-摘要2[冒泡排序]

冒泡排序 核心:冒泡,持续比较相邻元素,大的挪到后面,因此大的会逐步往后挪,故称之为冒泡。 public class BubbleSort {public static void main(String[] args) {int unsortedArray[] new int[]{6, 5, 3, 1, 8, 7, 2…

Leetcode每日一题学习训练——Python3版(到达首都的最少油耗)

版本说明 当前版本号[20231205]。 版本修改说明20231205初版 目录 文章目录 版本说明目录到达首都的最少油耗理解题目代码思路参考代码 原题可以点击此 2477. 到达首都的最少油耗 前去练习。 到达首都的最少油耗 ​ 给你一棵 n 个节点的树(一个无向、连通、无环…

倒计时模块复习

经典回顾倒计时 倒计时的基本布局介绍。 一个内容区域和一个输入区域,内容区域进行划分 直接使用flex布局会更快一点。 js代码 我们利用一下模块化思想,直接把获得时间这个功能写成一个函数。方便后续的调用 function getTime() {const date new Date…

MES管理系统通过哪些方面提升产品质量管理水平

在当今高度竞争的市场环境中,质量成为了企业生存和发展的关键因素。工厂作为生产产品的核心场所,其质量管理水平直接影响到产品的质量和企业的声誉。为了应对这一挑战,许多工厂引入了MES管理系统解决方案。本文将探讨MES管理系统如何帮助工厂…

【UE5】监控摄像头效果(上)

目录 效果 步骤 一、视角切换 二、摄像头画面后期处理 三、在场景中显示摄像头画面 效果 步骤 一、视角切换 1. 新建一个Basic关卡,添加第三人称游戏资源到项目浏览器 2. 新建一个Actor蓝图,这里命名为“BP_SecurityCamera” 打开“BP_Securit…

模电笔记。。。。

模电 2.8 蜂鸣器 按照蜂鸣器驱动方式分为有源蜂鸣器和无源蜂鸣器 有源的有自己的震荡电路,无源的要写代码控制。 里面有个线圈,相当于电感,储能,通直隔交。 蜂鸣器的参数:额定电压,工作电压&#xff0…

【CCF-B】1/2区,录用见刊极快!2个月录用!

计算机类 • 好刊解读 今天小编带来Taylor and Francis旗下计算机领域快刊,CCF-B类推荐的期刊解读,期刊审稿周期短,投稿友好,如您有投稿需求,可作为重点关注!后文有相关领域真实发表案例,供您投…

防水,也不怕水。Mate X5是如何做到让你湿手湿屏也不影响操作的?

相信不少人都碰到过当手机屏幕存在小水珠时,触控变得不灵敏,或者出现“幽灵触屏”,指东打西的情况。 尤其是在洗澡、做饭,或者在户外遇到下雨天气时,如果打湿的手机收到重要聊天消息或者电话,却因为湿屏导…

生成fip.bin在Milkv-duo上跑rtthread的相关尝试,及其问题分析

前言 (1)PLCT实验室实习生长期招聘:招聘信息链接 (2)本来是想在Milkv-duo上跑rtthread的,做了很多努力,一直没有结果。虽然不知道最终能不能成功做出来,还是把自己的相关努力分享出来…

MDK官网如何下载stm32支持包

网站:https://www.keil.com/demo/eval/arm.htm 1 2 3点这个下载

基于Mint Mate 21.2 Victoria 的Anjuta安装与测试

序言 Linux mint mate 21.2 命名为 victoria 版,在vmware虚拟机中安装按提示默认安装即可,不做更多记录。mint mate的优点是稳定,窗口质感好。安装完成后,需要关注一些常用功能配置。主要有:显示器调整、桌面调整、工…

当然热门的原创改写改写大全【2023最新】

在信息时代,随着科技的不断发展,改写软件逐渐成为提高文案质量和写作效率的重要工具。本文将专心分享一些好用的改写软件,其中包括百度文心一言智能写作以及147SEO改写软件。这些工具不仅支持批量改写,而且在发布到各大平台后能够…

python爬取 HTTP_2 网站超时问题的解决方案

问题背景 在进行网络数据爬取时,使用 Python 程序访问支持 HTTP/2 协议的网站时,有时会遇到超时问题。这可能会导致数据获取不完整,影响爬虫程序的正常运行。 问题描述 在实际操作中,当使用 Python 编写的爬虫程序访问支持 HTT…

使用高防IP防护有哪些优势

高防IP是针对互联网服务器在遭受大流量的DDoS攻击后导致服务不可用的情况下,推出的付费增值服务,用户可以通过配置高防IP,将攻击流量引流到高防IP,确保源站的稳定可靠。高防IP相当于搭建完转发的服务器。 高防IP有两种接入方式&a…

Notepad安装

中文免安装版,下载解压即可。 NotepadV7.5.6 (访问密码: 1666)https://url48.ctfile.com/f/33868548-986668939-7a3316?p1666