Python: Python网络协议和编程

所有设备都连接到公共总线上，结点间使用广播通信方式。一个结点发出的信息，总线上所有其他结点都可以接收到。一段时间只允许一个结点独占总线。

常见使用同轴电缆连接，总线两端需要终结器。

优点

• 结构简单、易于实现
• 易于扩充，增加或者移除结点比较灵活
• 可靠性较高，个别结点发生故障时，不影响网络中其他结点的正常工作

缺点

• 网络传输能力低，安全性低，总线发生故障时，会导致全网瘫痪
• 所有数据都需要经过总线传输，总线是整个网络的瓶颈。结点数量的增多会影响网络性能

环形结构是将联网的计算机由通信线路连接成一个闭合的环，在环形结构网络中信息按照固定方向流动，或顺时针方向，或逆时针方向。

优点

• 令牌控制，没有线路竞争，实时性强，传输控制容易

缺点

• 维护困难，可靠性不高。一个结点发生故障时，可能导致全网瘫痪。可以使用双环拓扑结构，但是复杂性提升

每个结点都由一条单独的通信线路与中心结点连结。其他各结点都与该中心结点 有着物理链路的直接互连，其他结点直接不能直接通信，其他结点直接的通信需 要该中心结点进行转发。因此中心结点必须有着较强的功能和较高的可靠性。需 要中心设备，例如hub、switch、router

优点

• 可靠性高，方便管理，易于扩展，传输效率高

缺点

• 线路利用率低，中心节点需要很高的可靠性和冗余度

注意, hub工作在一层，这种星型实际上就是芯片化的总线网络。只是物理拓扑结构上感觉像是星型。

IP v4 是一个32位二进制数，不便记忆，为了使用方便，使用“点分十进制”表示法，将这个二进制数每8位断开，每8位是一个字节，一个字节表示的十进制正整数范围是0~255 IP v4地址早期比较充足，随着全球连入互联网，在2011年IP v4地址分配完毕 IP地址的分类 公有地址： 需向因特网信息中心申请，在互联网上可以直接使用的IP地址

私有地址： 不需要注册，可以组织内部网络使用的IP地址

IP地址这个数被分成2部分，即网络位 + 主机位

网络位表示设备同属一个网络；主机位表示网络中不同的设备的唯一ID

子网掩码 子网掩码将IP地址划分为网络ID和主机ID IP地址 位与 子网掩码就是网络ID IP v4地址被分为A、B、C、D、E五类

A类 最高位是 0 第一字节（最高字节）为网络位。第一个字节变化为 0000 0001 到 0111 1111，共127，减去回环地址，剩余126个网络 A类IP地址范围1.0.0.0到127.255.255.255。二进制表示为： 00000001 00000000 00000000 00000000 至 01111111 11111111 11111111 11111111。最后一个是广播地址。 子网掩码 255.0.0.0 每一个网络中主机个数等于 256³ - 2 = 1677716 -2 = 1677714 B类 最高位是10 前2个字节为网络位，其变化为128.0~191.255，相当于 1000 0000 0000 0000 到 1011 1111 1111 1111，实际上就是后14位变化，共2^14 = 16384个网络 B类IP地址范围128.0.0.0-191.255.255.255 。二进制表示为： 10000000 00000000 00000000 00000000 至 10111111 11111111 11111111 11111111 最后一个是广播地址 子网掩码255.255.0.0 每一个网络中主机个数等于 256² - 2 = 65535 -2 = 65534 C类 最高位是110 前3个字节为网络位，其变化为192.0.0~223.255.255，相当于 1100 0000 0000 0000 0000 0000 到 1101 1111 1111 1111 1111 1111 ，实际上就是后21位变化，共2^21 = 2097152个网络。

C类IP地址地址范围192.0.0.0-223.255.255.255 。二进制表示为： 11000000 00000000 00000000 00000000 至 11011111 11111111 11111111 11111111。 最后一个是广播地址 子网掩码255.255.255.0 每一个网络中主机个数等于 256 -2 = 254 D类 多播地址，或组播地址 多播地址最高4位必须是1110，那么地址范围就是224.0.0.0到239.255.255.255 224.0.0.1特指所有主机

E类 实验用地址 特殊IP地址 0.0.0.0表示当前主机 255.255.255.255 限制广播地址。路由器不会转发这个受限广播地址的数据报文，此地址只能用于本网广播 IP地址中以127开头的地址称为Loopback回环地址 169.254.x.x，windows主机使用了DHCP获取IP，但没有获得地址，windows会临时获得这样的地址 网关GATEWAY 网关(Gateway)又称网间连接器、协议转换器。网关在网络层以上实现网络互连，用于网络间互联 举例 IP地址192.168.3.200，要配合子网掩码使用，假设子网掩码为255.255.255.0，说明它是C类地址 网络ID为192.168.3.0，广播地址为192.168.3.255 剩余192.168.3.1~192.168.3.254能够分配给网络中其他设备 网关地址配置一般习惯使用1、100、254等。本例使用192.168.3.1 其作用是连接不同的网络，也称为处在不同的网段

又有一个IP地址为192.168.100.10/24，它也是C类地址，/16指B类地址，/8指A类地址。网络ID是192.168.100.0 和上面的IP处在不同的网络，这两个地址的主机通信，就需要使用网关，由网关将数据包转发到另一个网络

互联网上的公有地址在2011年分配完，而随着互联网的发展，接入设备越来越多，尤其是物联网的到来，此问题必须解决。由此，提出了IP v6 IP v6采用128位二进制数表示，基本解决IP地址短缺现象，同时，该协议还解决原有协议的诸多问题

创建socket对象。type=socket.Sock_DGRAM 绑定IP和Port，bing()方法 传输数据 接受数据，socket.recvform(bufsize[flags]),获得一个二元组(string,address) 发送数据，socket.sendto(string, address)发给某地址某信息 释放资源 import socket

server = socket.socket(type=socket.SOCK_DGRAM)

server.bing(('0.0.0.0', 9999)) # 立即绑定一个udp端口 data = server.recv(1024) # 阻塞等待数据 data = server.recvfrom(1024) # 阻塞等待数据(value, (ip, port)) server.sendto(b'hello', ('192.168.142.1', 10000))

server.close()

UDP客户端编程流程 创建socket对象。type=socket.Sock_DGRAM 发送数据，socket.sendto(string, address)发给某地址某信息 接受数据，socket.recvform(bufsize[flags]),获得一个二元组(string,address) 释放资源 import socket client = socket.socket(type=socket.SOCK_DGRAM) raddr = ('192.168.142.1', 9999)

client.connect(raddr) client.sendto(b'hello', raddr) client.send(b'hello') data = client.recvfrom(1024) # 阻塞等待数据(value, (ip, port)) data = client.recv(1024) # 阻塞等待数据

client.close()

注意：UDP是无连接协议，所以可以只有任何一端 例如客户端数据发往服务端，服务端存在与否无所谓 UDP编程中bind、connect、send、sendto、recv、recvfrom方法使用 UDP的socket对象创建后，是没有占用本地地址和端口的 方法 说明 bind方法 可以指定本地地址和端口laddr，会立即占用 connect方法 可以立即占用本地地址和端口laddr，填充远程地址和端口raddr sendto方法 可以立即占用本地地址和端口laddr，并把数据发往指定远端。 只有有了本地绑定端口，sendto就可以向任何远端发送数据 send方法 需要和connect方法配合， 可以使用已经从本地端口把数据发往raddr指定的远端 recv方法 要求一定要在占用了本地端口后，返回接受的数据 recvfrom方法 要求一定要占用了本地端口后，返回接受的数据和对端地址的二元祖

UDP是无连接协议，它基于以下假设：

网络足够好 消息不会丢包 包不会乱序 但是，即使再局域网，也不能保证不丢包，而且包的达到不一定有序

应用场景 视频、音频传输，一般来说，丢些包，问题不大，最多丢些图像、听不清话语，可以重新发话语来解决 海量采集数据，例如传感器发来的数据，丢几十、几百条数据也没有关系 DNS协议，数据内容小，一个包就能查询到结果，不存在乱序，丢包，重新请求解析 一般来说，UDP性能优于TCP，但是可靠性要求高的场合的还是要选择TCP协议

------------

------------

v ----------- ------------------

----------- ------------------

v ----------- --------------------

----------- --------------------

SocketServer简化了网络服务器的编写 它有4个同步类：

TCPServer UDPServer UnixStreamServer UnixDatagramServer 2个Mixin类：ForkingMixIn 和 ThreadingMixIn 类，用来支持异步。由此得到

class ForkingUDPServer(ForkingMixIn, UDPServer): pass class ForkingTCPServer(ForkingMixIn, TCPServer): pass class ThreadingUDPServer(ThreadingMixIn, UDPServer): pass class ThreadingTCPServer(ThreadingMixIn, TCPServer): pass fork是创建多进程，thread是创建多线程 fork需要操作系统支持，Windows不支持

socketserver.BaseServer(server_address, RequestHandlerClass) 需要提供服务器绑定的地址信息，和用于处理请求的RequestHandlerClass类 RequestHandlerClass类必须是BaseRequestHandler类的子类，在BaseServer中代码如下：

class BaseServer: def __init__(self, server_address, RequestHandlerClass): """Constructor. May be extended, do not override.""" self.server_address = server_address self.RequestHandlerClass = RequestHandlerClass self.__is_shut_down = threading.Event() self.__shutdown_request = False def finish_request(self, request, client_address): # 处理请求的方法 """Finish one request by instantiating RequestHandlerClass.""" self.RequestHandlerClass(request, client_address, self) # RequestHandlerClass构造

BaseRequestHandler类 它是和用户连接的用户请求处理类的基类，定义为 BaseRequestHandler(request, client_address, server) 服务端Server实例接收用户请求后，最后会实例化这个类 它被初始化时，送入3个构造参数：request, client_address, server自身 以后就可以在BaseRequestHandler类的实例上使用以下属性：

self.request是和客户端的连接的socket对象 self.server是TCPServer实例本身 self.client_address是客户端地址 这个类在初始化的时候，它会依次调用3个方法。子类可以覆盖这些方法

class BaseRequestHandler: def __init__(self, request, client_address, server): self.request = request self.client_address = client_address self.server = server self.setup() try: self.handle() finally: self.finish()

def setup(self): # 每一个连接初始化 pass

def handle(self): # 每一次请求处理 pass

def finish(self): # 每一个连接清理 pass

测试代码 import threading import socketserver

class Myhandler(socketserver.BaseRequestHandler): def handle(self):

print('-' * 30) print(self.server) # 服务 print(self.request) # 服务端负责客户端连接请求的socket对象 print(self.client_address) # 客户端地址 print(self.__dict__) print(self.server.__dict__) # 能看到负责accept的socket

print(threading.enumerate()) print(threading.current_thread()) print('-' * 30)

addr = ('192.168.142.1', 9999) server = socketserver.ThreadingTCPServer(addr, Myhandler)

server.serve_forever() # 永久循环执行

<socketserver.ThreadingTCPServer object at 0x000001A8055D80F0>

<socket.socket fd=492, family=AddressFamily.AF_INET, type=SocketKind.SOCK_STREAM, proto=0, \ laddr=('127.0.0.1', 9999), raddr=('127.0.0.1', 50280)>

('127.0.0.1', 50280)

{'request': <socket.socket fd=492, family=AddressFamily.AF_INET, type=SocketKind.SOCK_STREAM, \ proto=0, laddr=('127.0.0.1', 9999), raddr=('127.0.0.1', 50280)>, 'client_address': ('127.0.0.1', 50280),\ 'server': <socketserver.ThreadingTCPServer object at 0x000001A8055D80F0>}

{'server_address': ('127.0.0.1', 9999), 'RequestHandlerClass': <class 'main.Myhandler'>, \ '_BaseServer__is_shut_down': <threading.Event object at 0x000001A8055D85F8>, '_BaseServer__shutdown_request': False,\ 'socket': <socket.socket fd=540, family=AddressFamily.AF_INET, \ type=SocketKind.SOCK_STREAM, proto=0, laddr=('127.0.0.1', 9999)>}

[<_MainThread(MainThread, started 53996)>, <Thread(Thread-1, started 53580)>]

<Thread(Thread-1, started 53580)>

测试结果说明，handle方法相当于socket的recv方法 每个不同的连接上的请求过来后，生成这个连接的socket对象即self.request，客户端地址是self.client_address

测试过程中，上面代码，连接后立即断开了,怎样才能客户端和服务器端长时间连接 import threading import socketserver import logging

FORMAT = "%(asctime)s %(threadName)s %(thread)d %(message)s" logging.basicConfig(format=FORMAT, level=logging.INFO)

class Myhandler(socketserver.BaseRequestHandler): def handle(self):

print('-' * 30) print(self.server) # 服务 print(self.request) # 服务端负责客户端连接请求的socket对象 print(self.client_address) # 客户端地址 print(self.__dict__) print(self.server.__dict__) # 能看到负责accept的socket

print(threading.enumerate()) print(threading.current_thread()) print('-' * 30) for i in range(3): # 三次后客户端断开连接 data = self.request.recv(1024) logging.info(data) logging.info('==end==')

addr = ('127.0.0.1', 9999) server = socketserver.ThreadingTCPServer(addr, Myhandler)

server.serve_forever() # 永久循环执行

将ThreadingTCPServer换成TCPServer，同时连接2个客户端观察效果

ThreadingTCPServer是异步的，可以同时处理多个连接 TCPServer是同步的，一个连接处理完了，即一个连接的handle方法执行完了，才能处理另一个连接，且只有主线程

总结 创建服务器需要几个步骤：

从BaseRequestHandler类派生出子类，并覆盖其handle()方法来创建请求处理程序类，此方法将处理传入请求 实例化一个服务器类，传参服务器的地址和请求处理类 调用服务器实例的handle_request()执行一次或serve_forever()永久执行方法 调用server_close()关闭套接字

顾名思义，Echo，来什么消息回显什么消息 客户端发来什么信息，返回什么信息

import threading import socketserver

class Handler(socketserver.BaseRequestHandler): def setup(self): super().setup() self.event = threading.Event()

def finish(self): super().finish() self.event.set()

def handle(self): super().handle() while not self.event.is_set(): data = self.request.recv(1024).decode() msg = '{} {}'.format(self.client_address, data).encode() self.request.send(msg)

server = socketserver.ThreadingTCPServer(('127.0.0.1', 9999), Handler) threading.Thread(target=server.serve_forever, name='EchoServer', daemon=True).start()

while True: cmd = input('>>').strip() if cmd == 'quit': server.server_close() break print(threading.enumerate())

使用ThreadingTCPServer改写ChatServer

import datetime import threading from socketserver import ThreadingTCPServer, BaseRequestHandler, StreamRequestHandler import logging

FORMAT = "%(asctime)s %(threadName)s %(thread)d %(message)s" logging.basicConfig(format=FORMAT, level=logging.INFO)

class ChatHandler(StreamRequestHandler): clients = {}

def setup(self): super().setup() self.event = threading.Event() self.clients[self.client_address] = self.wfile

def handle(self): super().handle()

while not self.event.is_set(): data = self.rfile.read1(1024) # 可以读取到数据 data = data.decode().rstrip()

if data ='quit' or data = '': # 主动退出和断开 break

msg = '{} {}:{} {}'.format(datetime.datetime.now(), *self.client_address, data)

for f in self.clients.values(): f.write(msg.encode()) f.flush()

def finish(self): super().finish() self.clients.pop(self.client_address) self.event.set()

server = ThreadingTCPServer(('127.0.0.1', 9999), ChatHandler) server.daemon_threads = True # 让所有启动线程都为daemon

threading.Thread(target=server.serve_forever, name='chatserver', daemon=True).start()

while True: cmd = input('>>') if cmd.strip() == 'quit': server.server_close() break print(threading.enumerate())

问题

上例 self.clients.pop(self.client_address) 能执行到吗？ 如果连接的线程中handle方法中抛出异常，例如客户端主动断开导致的异常，线程崩溃，self.clients的pop方法还能执行吗？ 当然能执行，基类源码保证了即使异常，也能执行finish方法。但不代表不应该不捕获客户端各种异常 注意：此程序线程不安全，需要加锁来解决

为每一个连接提供RequestHandlerClass类实例，依次调用setup、handle、finish方法，且使用了try…finally结构保证finish方法一定能被调用。这些方法依次执行完成，如果想维持这个连接和客户端通信，就需要在handle函数中使用循环。 socketserver模块提供的不同的类，但是编程接口是一样的，即使是多进程、多线程的类也是一样，大大减少了编程的难度 将socket编程简化，只需要程序员关注数据处理本身，实现Handler类就行了。 这种风格在Python十分常见

函数或方法被调用的时候，调用者是否得到最终结果的

直接得到最终结果的，就是同步调用 不直接得到最终结果的，就是异步调用

函数或方法调用的时候，是否立刻返回

立即返回就是非阻塞调用 不立即返回就是阻塞调用

同步、异步，与阻塞、非阻塞不相关 同步、异步强调的是，是否得到（最终的）结果 阻塞、非阻塞强调是时间，是否等待

同步与异步区别在于：调用者是否得到了想要的最终结果 同步就是一直要执行到返回最终结果 异步就是直接返回了，但是返回的不是最终结果。调用者不能通过这种调用得到结果，以后可以通过被调用者提供的某种方式（被调用着通知调用者、调用者反复查询、回调），来取回最终结果

阻塞与非阻塞的区别在于，调用者是否还能干其他事 阻塞，调用者就只能干等 非阻塞，调用者可以先去忙会别的，不用一直等

同步阻塞，我啥事不干，就等你打饭打给我。打到饭是结果，而且我啥事不干一直等，同步加阻塞 同步非阻塞，我等着你打饭给我，但我可以玩会手机、看看电视。打饭是结果，但是我不一直等

异步阻塞，我要打饭，你说等叫号，并没有返回饭给我，我啥事不干，就干等着饭好了你叫我。例如，取了号什么不干就等叫自己的号 异步非阻塞，我要打饭，你给我号，你说等叫号，并没有返回饭给我，我在旁边看电视、玩手机，饭打好了叫我

同步IO模型包括 阻塞IO、非阻塞IO、IO多路复用、信号驱动IO

进程发起异步IO请求，立即返回。内核完成IO的两个阶段，内核给进程发一个信号。 举例

来打饭，跟大师傅说饭好了叫你，饭菜准备好了，窗口服务员把饭盛好了打电话叫你。两阶段都是异步的。在整个过程中，进程都可以忙别的，等好了才过来。 今天不想出去到饭店吃饭了，点外卖，饭菜在饭店做好了（第一阶段），快递员从饭店送到你家门口（第二阶段）。 Linux的aio的系统调用，内核从版本2.6开始支持

前4个都是同步IO，因为核心操作recv函数调用时，进程阻塞直到拿到最终结果为止。 而异步IO进程全程不阻塞

3.4版本提供selectors库，高级IO复用库

类层次结构︰ BaseSelector +– SelectSelector 实现select +– PollSelector 实现poll +– EpollSelector 实现epoll +– DevpollSelector 实现devpoll +– KqueueSelector 实现kqueue

selectors.DefaultSelector返回当前平台最有效、性能最高的实现。 但是，由于没有实现Windows下的IOCP，所以，Windows下只能退化为select

在selects模块源码最下面有如下代码

if 'KqueueSelector' in globals(): DefaultSelector = KqueueSelector elif 'EpollSelector' in globals(): DefaultSelector = EpollSelector elif 'DevpollSelector' in globals(): DefaultSelector = DevpollSelector elif 'PollSelector' in globals(): DefaultSelector = PollSelector else: DefaultSelector = SelectSelector

事件注册 class SelectSelector(_BaseSelectorImpl): """Select-based selector.""" def register(fileobj, events, data=None) -> SelectorKey: pass

为selector注册一个文件对象，监视它的IO事件。返回SelectKey对象。

fileobj 被监视文件对象，例如socket对象 events 事件，该文件对象必须等待的事件 data 可选的与此文件对象相关联的不透明数据，例如，关联用来存储每个客户端的会话ID，关联方法。通过这个参数在关注的事件产生后让selector干什么事 Event常量 含义 EVENT_READ 可读 0b01，内核已经准备好输入输出设备，可以开始读了 EVENT_WRITE 可写 0b10，内核准备好了，可以往里写了 selectors.SelectorKey 有4个属性：

fileobj 注册的文件对象 fd 文件描述符 events 等待上面的文件描述符的文件对象的事件 data 注册时关联的数据

完成一个TCP Server，能够接受客户端请求并回应客户端消息

import selectors import socket

s = selectors.DefaultSelector() # 拿到selector

server = socket.socket() server.bind(('127.0.0.1', 9999)) server.listen()

server.setblocking(False)

def accept(sock:socket.socket, mask:int): """mask:事件的掩码""" conn, raddr = sock.accept() conn.setblocking(False) # 非阻塞 key = s.register(conn, selectors.EVENT_READ, recv) # 读和写可同时监控, selectors.EVENT_READ | selectors.EVENT_WRITE print(key)

def recv(conn:socket.socket, mast:int): data = conn.recv(1024) print(data) msg = 'Your msg = {} from {}'.format(data.decode(), conn.getpeername()).encode() conn.send(msg)

key = s.register(server, selectors.EVENT_READ, accept) #socket fileobject print(key) print(key.__class__.mro())

while True:

events = s.select() # epoll select, 默认是阻塞的

print(events) # [(key, mask)]

for key, mask in events:

print(type(key), type(mask)) print(key.data) key.data(key.fileobj, mask)

将ChatServer改写成IO多路复用的方式 不需要启动多线程来执行socket的accept、recv方法了

import socket import threading import selectors import logging

FORMAT = "%(asctime)s %(threadName)s %(thread)d %(message)s" logging.basicConfig(format=FORMAT, level=logging.INFO)

class ChatServer: def __init__(self, ip='127.0.0.1', port=9999): self.sock = socket.socket() self.addr = ip, port self.event = threading.Event()

self.selector = selectors.DefaultSelector()

def start(self): self.sock.bind(self.addr) self.sock.listen() self.sock.setblocking(False)

key = self.selector.register(self.sock, selectors.EVENT_READ, self.accept) logging.info(key) # 只有一个

threading.Thread(target=self.select, name='select', daemon=True).start()

def select(self): while not self.event.is_set():

events = self.selector.select() # 阻塞 print(events) # [(key, mask)] for key, mask in events:

key.data(key.fileobj, mask) # select线程

def accept(self, sock:socket.socket, mask): """mask:事件的掩码""" conn, raddr = self.sock.accept() conn.setblocking(False) # 非阻塞

key = self.selector.register(conn, selectors.EVENT_READ, self.recv) logging.info(key) #n个

def recv(self, conn:socket.socket, mask): data = conn.recv(1024)

if data.strip() = b'quit' or data.strip() = b'':

self.selector.unregister(conn) conn.close() return

for key in self.selector.get_map().values():

if key.data == self.recv: # key.data 注册时注入的绑定的对象 s = key.fileobj msg = 'Your msg = {} from {}'.format(data.decode(), conn.getpeername()).encode() s.send(msg)

def stop(self): self.event.set()

fs = [] for key in self.selector.get_map().values(): fs.append(key.fileobj) for f in fs: self.selector.unregister(f) f.close()

self.selector.close()

if name = '__main__': cc = ChatServer() cc.start() while True: cmd input('>>').strip() if cmd == 'quit': cc.stop() break logging.info(threading.enumerate())

本例只完成基本功能，其他功能如有需要，请自行完成。 注意使用IO多路复用，使用了几个线程？ 特别注意key.data == self.recv