犀渠玉剑良家子,白马金羁侠少年。这篇文章主要讲述#yyds干货盘点# Socket 编程与测试详细介绍相关的知识,希望能为你提供帮助。
在python文档中,socket相关的内容记录在这些页面:
- socket --- 底层网络接口
- socketserver --- 用于简化网络服务端编写的类
- 套接字编程方法
- Socket application programming interface overview
文件IO:
- 打开文件
- 对文件内容读取
- 对文件内容写入
- 建立管道,conn1 和conn2 ,分别给不同的进程
- conn1 输入内容,conn2可以输出内容
- conn2输入内容,conn1 可以输出内容
计算机之间的通信:
- 建立socket连接,两个计算机分别充当 server和client
- serves输入内容,client输出内容
- client输入内容,server输出内容
【#yyds干货盘点# Socket 编程与测试详细介绍】计算机通信标准:OSI七层模型
文章图片
核心思想:各层依赖下一层,但是不必关系下一层的细节,各司其职。分工合作
实际的应用:TCP / IP 协议簇,没有完全安装OSI 将模型分为7层,而是4层
文章图片
OSI 是理论,是知道思想
TCP/IP 是现实,是具体的工具
所以在创建socket的时候,有一些选项来说明本次使用的协议具体是什么
常用的两个:
文章图片
SOCK_STREAM | SOCK_DGRAM | SOCK_RAW | |
---|---|---|---|
AF_UNIX | UNIX DOMAIN SOCKET | UNIX DOMAIN SOCKET | |
AF_INET | IPv4 + TCP | IPv4+ UDP | |
AF_INET6 | IPv6+ TCP | IPv6+ UDP | |
AF_PACKET | 自定义协议 |
2. 创建和使用Socketsocket.socket 类
- 接收数据的方法
- 发送数据的方法
- 监听端口的方法
- ……
socket函数
- 创建客户端
- 创建服务端
- 创建客户端 +服务端(已连接)
- ……
2.1.1 基本函数本质就是 socket.socket类的实例化
import sockets1 = socket.socket(
type=socket.SOCK_STREAM,
family=socket.AF_INET,
)# TCP+ ipv4实现跨计算机通信s2 = socket.socket(
type=socket.SOCK_STREAM,
family=socket.AF_UNIX,
)# unix socket 同计算机跨进程通信
2.1.2 便捷函数
- 创建服务端socket
- 创建socket
- 绑定地址(ip+port)
- 监听地址
import socketserver_address = ("127.0.0.1", 9001) server = socket.create_server( server_address, family=socket.AF_INET )# 可以指定IPV4 不该改变类型 TCPserver_address_2 = ("::1", 9002) server_2 = socket.create_server( server_address_2, family=socket.AF_INET6 )#或者 IPV6,不该改变类型 TCP
- 创建客户端socket
- 创建socket
- 绑定地址(ip+port)
- 想指定的服务端发起连接
client = socket.create_connection( server_address, source_address=("127.0.0.1", 9005) )# 为空的会自动分配
- 创建socket
文章图片
文章图片
- 创建已连接的socket
- 创建两个socket,其中作为server监听端口,另一个作为client连接端口
- 连接成功后,把两个socket返回
s1, s2 = socket.socketpair() # 如果是linux 使用AF_UNIX(同计算机跨进行通信) # 如果不是linux 使用AF_INTE(同计算机跨进行通信,地址127.0.0.1)
- 创建两个socket,其中作为server监听端口,另一个作为client连接端口
文章图片
?
文章图片
2.2 socket对象
socket提供了一些列的方法,完成网络连接 、 数据通信,常用的:
- 网络连接
- 服务端
- bind 绑定地址(IP+PORT)
- listen 监听端口,允许其他socket发起连接
- accpet 接收连接,生成新的socket,完成数据收发
- close 关闭
- 客户端
- bind 绑定地址(IP+PORT)
- connect 连接指定地址
- close 关闭
- 数据通信(IO)
- sendall 发送数据
- recv 接收数据
import logging.configimport yamlwith open("logging.conf.yaml") as f:
config = yaml.safe_load(f)
logging.config.dictConfig(config)logger_server = logging.getLogger("server")
logger_client = logging.getLogger("client")
2.2.2 创建server端
server = socket.socket(family=socket.AF_INET, type=socket.SOCK_STREAM)
logger_server.info("socket创建")server.bind(server_address)
logger_server.info(f"绑定了地址 {server_address}")server.listen()
logger_server.info(f"监听成功,服务器已就绪")# ---
# 2.使用便捷函数,快速配置
server = socket.create_server(server_address)# python3.8+
logger_server.info(f"监听{server_address}成功,服务器已就绪")
2.2.3 创建客户端
client = socket.socket(family=socket.AF_INET, type=socket.SOCK_STREAM)
logger_client.info(f"socket已创建,准备连接 {server_address=}")
client.connect(server_address)
logger_client.info("连接成功")# ---
client = socket.create_connection(server_address)
logger_client.info("连接成功")
2.2.4 通信传递的是二进制数据流,字节流
conn, address = server.accept()# 服务端接收请求
logger_server.info(f"新的客户端进入: {address=}")data = https://www.songbingjia.com/android/conn.recv(1024)
logger_server.info(f"收到数据 {data=https://www.songbingjia.com/android/}")conn.sendall(data)# 发送数据
conn.close()# ----data = https://www.songbingjia.com/android/client.recv(1024)
logger_client.info(f"收到数据 {data=https://www.songbingjia.com/android/}")client.close()
2.2.5 测试驱动开发测试驱动开发 TDD
- 编写测试用例
- 执行测试用例 (测试失败)(红)
- 编写业务代码
- 执行测试用例 (测试通过)(绿)
- 重构代码
- 执行测试用例 (测试通过)(绿)
- 分解被测模块
- server
- 连接
- 通信
- 业务功能
- client
- 连接
- 通信
- server
- 了解被测对象
- 获取socket属性、状态
- 改变socket属性、状态
- 甚至mock
- 编写测试用例
- 使用测试框架
- 必要的话,自定义断言
测试用例中的client ---> 业务中的server,
业务中的server-----> 业务中client
文件结构:
- server.py
- test_server.py# unittest 默认发现规则
- server_test.py# 需要修改发现规则:python -m unittest discover -p *_test.py
socketserver
预置常见的socketserver:- TCPServer
- ThreadingTCPServer
- ForkingTCPServer
- 能处理客户端中断的异常,不会导致server退出
- 停供了并发的支持
- 将网络连接、数据处理 分开
- 封装工作细节
- 将IO,封装成了两个(读、写)like file
- 提供了和文件读写相同方法,完成数据收发
- rfile只读
- wfile 只写
- ThreadingTCPServer 使用
- socket服务测试
多人同时在线,每个人都可以发言,发言内容可以被每个人收到
4.2 协议
- 传输协议
- 使用TCP进行传输
- 使用换行符作" \\n" 为结束标记
- 数据格式
- name 昵称
- msg 内容
- time 时间
- {" name" : " 张三" , " msg" :" 你好" , " time" : " 2020-11-1" }(使用json字符串)
4.3 编写socket客户端
- 建立socket,断开socket
- 发送消息能力
- 确定要发送的内容
- 组装固定格式的字典
- 讲字段转字符串
- 字符串 +”\\n"
- 字符串 转字节流
- 借助socket发送字节流
- 接收消息能力
- 从socket读取字节流
- 字节流转字符串
- 去掉结束标记
- 字符串转字典
- 返回字典
- 服务端是否就绪
- 多线程
- 多进程
- 子进程
- 服务器能否关闭
- 多进程
- 测试用例
- 服务器可用
- 服务器多人可用
- 服务器可以回复消息
- 服务器可以对多个用户广播消息
- 服务器在用户掉线后,依然正常
import socketserverclass MyUDPHandler(socketserver.StreamRequestHandler):
"""
This class works similar to the TCP handler class, except that
self.request consists of a pair of data and client socket, and since
there is no connection the client address must be given explicitly
when sending data back via sendto().
"""def handle(self):
while True:
data = https://www.songbingjia.com/android/self.rfile.readline()
print(f"收到客户端数据 {data=https://www.songbingjia.com/android/}")with socketserver.TCPServer(("127.0.0.1", 9002), MyUDPHandler) as server:
server.serve_forever()
推荐阅读
- 技术分享| 如何做一款容纳百人的视频会议()
- Kubernetes中spinnaker使用二
- 五子棋游戏(简单易懂,入门都能学)
- Python 中的反转字符串(reversed()切片等)
- 如何为WinXP系统设置电脑自动锁屏
- win8系统设置图片密码失败怎样办
- 如何处理WindowsXP系统电脑不能关机的图文详细教程
- 如何处理xp系统下WPS文档忘记保存的办法
- 设置WinXP系统电脑从不待机的办法