python中的网络通信，socket、select、selectors、socketserver

楔子

网络通信用于获取一个算法在本地运行所需的数据，还可以共享信息实现分布式处理，另外可以用来管理云服务。 python的标准库提供了一些模块来创建网络服务以及访问现有服务ipaddress模块提供了一些类来验证、比较和处理IPV4/IPV6网络地址。底层socket库允许直接访问原生C套接字库，可以用于与任何网络服务通信。selectors提供了一个高层接口，可以同时监视多个套接字，这对于支持网络服务器同时与多个客户通信很有用。select提供了selectors使用的底层API。socketserver中的框架抽象了创建一个新的网络服务器所需要的大量重复性工作。可以结合这些类创建服务器来建立或使用线程以及支持TCP或UDP。应用只需要完成实际的消息处理

ipaddress：Internet地址

介绍

ipaddress模块提供了处理IPV4和IPV6网络地址的类。这些类支持验证，查找网络上的地址和主机，以及其他常见操作

地址

import binascii
import ipaddress
 
 
'''
最基本的对象表示网络地址本身。可以像ip_address函数传入一个字符串，整数或者字节序列来构造一个地址。
返回值是一个IPv4Address或IPv6Address实例，这取决于使用什么类型的地址
'''
address = "61.135.169.125"
 
addr = ipaddress.ip_address(address)
print(addr)
print(f"ip version: {addr.version}")
print(f"is private: {addr.is_private}")
print(f"packed form: {binascii.hexlify(addr.packed)}")
print(f"integer: {int(addr)}")
'''
61.135.169.125
ip version: 4
is private: False
packed form: b'3d87a97d'
integer: 1032300925
'''

socket：网络通信

介绍

sock模块提供了一个底层的C API，可以使用BSD套接字接口实现网络通信。它包括socket类，用于处理具体的数据通道，还包括用来完成网络相关任务的函数，如将一个服务器名转换为一个地址以及格式化数据以便在网络上发送。

寻址、协议簇和套接字类型

套接字(socket)是程序在本地或者通过互联网来回传递数据时所用的通信通道的一个端点。套接字有两个主要属性用于控制如何发送数据：地址簇(address family)控制所用的OSI网络协议；套接字类型(socket type)控制传输层协议。
python通常支持三个地址簇。最常用的是AF_INET，用于IPV4寻址。IPV4地址长度为4个字节，通常表示为4个数的序列，每个字节对应一个数，用点号分隔(如127.0.0.1)。这些值通常被称为IP地址，目前几乎所有的互联网网络通信都是用IPV4.
AF_INET6用于IPV6寻址。IPV6是下一代Internet协议，它支持128位地址和通信流调整，还支持IPV4不支持的一些路由特性。采用IPV6的应用在不断增多，特别是伴随着云计算的大量普及以及物联网项目而为网络增加很多额外的设备，都促使IPV6得到更广泛的应用
AF_UNIX是unix域套接字(unix domain socket,uds)的地址簇，这是一种POSIX兼容系统上的进程间通信协议。uds的实现通常允许操作系统直接从进程向进程传递数据，而不用通过网络栈。这比使用AF_INET更加高效，但是由于要使用文件系统作为寻址的命名空间，所以uds的优势仅限于同一个系统上的进程。相比其他的ipc机制(如命名管道或共享内存)，使用uds的优势在于它与ip网络应用的编程接口是一致的。这说明，应用在单个主机上运行时可以利用高效的通信，在网络上发送数据时仍然可以使用相同的代码。
套接字类型往往是SOCK_DGRAM或SOCK_STREAM，其中SOCK_DGRAM对应面向消息的数据报传输，而SOCK_STREAM对应面向流的传输。数据报套接字通常与UDP关联，即用户数据报协议(user datagram protocol)。这些套接字能提供不可靠的消息传送。面向流的套接字与TCP相关，即传输控制协议(transmission control protocol)。它们可以在客户和服务器之间提供字节流，通过超时管理、重传和其他特性确保提供消息传送或失败通知。
大多数传送数据的应用协议(如HTTP)都建立在TCP基础上，因为这样可以更容易地创建自动处理消息排序和传送的复杂应用。UDP通常用于顺序不太重要的协议(因为消息是自包含的，而且通常很小，如通过DNS的名字查找)，或者用于组播(向多个主机发送相同的数据)。UDP和TCP都可以用于IPV4或IPV6寻址
举个栗子：UDP就好比发短信，发完了就不管了，信息是可能丢失的，但是不管，只要发了就ok，对方是否接收到，我不管。TCP就好比打电话，我拨号，必须要确保对方接电话，才可以。所以说TCP相比UDP操作更复杂，但更安全

在网络上查找主机

import socket

'''
socket包含一些与网络上的域名服务交互的函数，这使得程序可以将服务器的主机名转换为数字网路地址。
应用使用地址链接服务器之前并不需要显式地转换地址，不过报告错误时除了报告所用的名字之外，如果还能包含这个数字地址，那么便会很有用
'''
# 要查找当前主机的正式名，可以使用gethostname函数
print(socket.gethostname())  # 憨八嘎

# 还可以使用gethostbyname函数根据主机名获取ip地址
print(socket.gethostbyname(socket.gethostname()))  # 192.88.88.110
# 不仅如此，还可以根据url，找出网站的ip。
# 总所周知，我们访问百度，可以通过www.baidu.com，但是我们是通过这个域名来访问百度的ip地址，只是ip比较难记罢了
# 我们也可以根据url找到对应网站的ip
print(socket.gethostbyname("www.baidu.com"))  # 61.135.169.125

# 如果想访问更多的信息，可以使用这个函数
name, aliases, addressed = socket.gethostbyname_ex("www.baidu.com")
print(name)  # www.a.shifen.com
print(aliases)  # ['www.baidu.com']
print(addressed)  # ['61.135.169.121', '61.135.169.125']

查找服务信息

import socket
 
 
'''
除了ip地址之外，每个套接字地址好包括一个整数端口号。
很多应用可以在同一个主机上运行并监听一个ip地址，不过只有一个套接字可以使用该地址的端口。
通过结合ip地址、协议和端口号，可以唯一地标识一个通信通道，并确保通过一个套接字发送的消息能到达正确的目标
 
有些端口号已经预先分配给特定的协议。例如，使用SMTP的email服务器使用TCP在端口25完成通信，web客户和服务器使用端口80完成HTTP通信。
'''
# 可以使用getservbyname方法查找网络服务的端口号和标准名
print(socket.getservbyname("http"))  # 80
print(socket.getservbyname("https"))  # 443
print(socket.getservbyname("ftp"))  # 21
print(socket.getservbyname("smtp"))  # 25
 
# 也可以使用getservbyport根据端口号查找协议
print(socket.getservbyport(80))  # http
print(socket.getservbyport(666))  # doom

查找服务器地址

import socket
 
 
'''
getaddrinfo函数将一个服务的基本地址转换为一个元组列表，其中包含建立一个连接所需的全部信息。每个元组可能包含不同的网络簇或协议
'''
res = socket.getaddrinfo("www.baidu.com", "https")
print(res)
# [(<AddressFamily.AF_INET: 2>, 0, 0, '', ('61.135.169.125', 443)), (<AddressFamily.AF_INET: 2>, 0, 0, '', ('61.135.169.121', 443))]

TCP/IP客户和服务器

服务端：

import socket
 
 
# 创建tcp/ip套接字，这里作为服务端
server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
# 绑定ip和端口，客户端连接这个ip和端口就可以向我们这里的服务端发送消息
# 这里要传入元组
server.bind(("localhost", 8888))
# 同时监听多少个到来的链接，也就是一次最多可以让多少个客户连接
# 虽然可以监听多个链接，但是一次性只能处理一个。
# 比如监听5，表示接收5个链接，当来了5个链接，并不是同时处理5个，而是一次性处理一个，其他的4个要排队。而再来第6个链接连队也不让排了，直接拒绝接收了
server.listen(5)
 
# 创建循环，表示一个链接关闭了，继续循环处理下一个链接
while True:
    # accept：等待客户端的链接，一旦到来，就会与之交互
    # 会有两个返回值。conn：客户端与服务端之间建立的一个链接。addr：客户的地址
    # 这个过程是会阻塞的
    conn, addr = server.accept()
    # 创建循环，因为消息不止发送一次，会多次来回发送，你一句我一句，直到没话了，结束循环
    # 从而执行外层循环，等待下一个链接，然后继续循环你一句我一句
    while True:
        # 发送消息和接收消息都是conn这个链接实例去执行的
        # recv表示接收数据，1024表示最多一次性最多接收1024个字节
        data = conn.recv(1024)
        # 如果对方断开链接了，那么发送的数据就为空了
        # 所以要进行判断，如果得到的数据为空，那么就break，也就是断开当前的链接
        if not data:
            print(f"客户端{addr}：断开链接")
            break
        # 有数据的话，那么打印出来
        print(str(data, encoding="utf-8"))
        # 同时也给客户端发送数据
        conn.sendall(data + bytes(" 我收到了", encoding="utf-8"))
     
    # 清理链接，进行资源、端口的释放等等
    conn.close()

客户端：

import socket
 
 
# 我们是客户端，所以要创建一个绑定到指定ip和端口的链接
client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
# 这里不再是bind了，而是connect，因为我们不是创建服务器等待链接，而是客户端，要绑定指定的链接去连接别人
client.connect(("localhost", 8888))
while True:
    # 发送数据，到时候服务端也会回数据
    client.sendall(bytes(input("你要发送的数据："), encoding="utf-8"))
    data = client.recv(1024)
    print(str(data, encoding="utf-8"))

可以看到，这里还有一个不完美的地方，那就是当客户端断开链接时，服务端报错了，这要咋办呢？可以在第二层循环来一个异常捕获就行了

select：高效等待I/O

select模块允许访问特定于平台的I/O监视函数。最可移植的接口是POSIX函数select()，Unix和Windows都提供了这个函数。这个模块还包括poll(这嗝api只适用于Unix)，另外还提供了很多只适用于一些Unix特定版本的选项

使用select()

python的select函数是底层操作系统实现的一个直接接口。它会监视套接字、打开的文件和管道(可以是任何有fileno()方法的对象，这个方法会返回一个合法的文件描述符)，直到这个对象可读或可写，或者出现一个通信错误。利用select函数可以很容易地同时创建多个连接，这比python中使用套接字超时编写一个轮询循环更为高效，因为监视发生在操作系统网络层而不是在解释器中完成。

服务端

import select
import socket
import sys
import queue
 
server = socket.socket()
server.bind(("localhost", 8888))
server.listen(5)
'''
select函数接收三个列表，包含要监视的信息通道
首先检查第一个对象列表中的对象以得到要读取的数据
第二个列表包含的对象将接收发出的数据(如果缓冲区有空间)
第三个列表包含那些可能有错误的对象(通常是输入和输出通道对象的组合)。
因此：首先要建立这三个列表
'''
# 这个server是必须的，原因后面解释
inputs = [server]
outputs = []
message_queue = {}
 
while inputs:
    # select返回三个列表，里面分别存储：可读套接字，可写套接字，异常套接字
    # readable：列表中套接字会缓存所有到来的数据，可供读取
    # writable：列表中套接字的缓冲区中有自由空间，可以写入数据
    # exceptional：列表中套接字都有一个错误(异常条件的具体定义取决于平台)
    readable, writable, exceptional = select.select(inputs, outputs, inputs)
    # 可读套接字表示三种可能的情况。
    # 如果是主服务器套接字，即用来监听链接的套接字，那么可读条件就意味着它已经准备就绪，可以接收另一个入站链接
    # 当链接到来，我们就将其添加到inputs里面，如果是链接活跃了，意味着对方发数据了，如果是主套接字活跃了，意味着新链接到来了
    # 所以inputs里面必须要有一个主套接字，不然连链接都没有
 
    # 循环readable，也就是inputs里面那些活跃的套接字
    for s in readable:
        # 如果是主套接字活跃了
        # 意味着新链接到来了
        if s is server:
            conn, addr = s.accept()
            print(f"来自{addr}的链接")
            # 将链接添加到inputs里面，进行监视
            inputs.append(conn)
            # 给当前链接一个队列，用于存储数据
            message_queue[conn] = queue.Queue()
 
        # 如果不是主套接字活跃了，那么肯定是我监听的链接活跃了，说明发送数据了
        # 但是第一次活跃的一定是主套接字，因为一开始inputs里面只有它，只有它活跃了，链接来了，我们才能添加到inputs里面
        # 所以第一次活跃的一定是主套接字，至于到底活跃几次，要看有几个新链接到来
        else:
            data = s.recv(1024)
            # 如果有数据
            if data:
                print(f"收到来自于{s.getpeername()}的信息{str(data, encoding='utf-8')}")
                message_queue[s].put(data)
                # 因为我们要从outputs里面读数据
                # 所以如果s不在outputs里面，要加进去
                if s not in outputs:
                    outputs.append(s)
            else:
                # 没数据,表示这个套接字来自已经断开链接的客户，所以此时可以关闭流
                print(f"链接{s.getpeername()}已经断开")
                if s in outputs:
                    outputs.remove(s)
                inputs.remove(s)
                s.close()
                # 移除消息队列
                del message_queue[s]
 
    # 对于可写，情况要少一些。如果一个连接的相应队列中有数据，那么就发送下一个消息。
    # 否则，将这个链接从输出链接列表当中删除，这样下一次循环的时候，select函数不再指示这个套接字已经准备好发送数据
 
    # 处理outputs
    for s in writable:
        # 因为在处理readable时，我们把活跃的链接添加到了outputs里面
        try:
            next_msg = message_queue[s].get_nowait()
        except queue.Empty:
            # 没有数据了
            print(f"{s.getpeername()}对应队列为空")
            outputs.remove(s)
        else:
            print(f"发送{next_msg}给{s.getpeername()}")
            s.send(next_msg + bytes("我收到了", encoding="utf-8"))
 
    # 最后一种情况，exceptional列表中的套接字会关闭
    for s in exceptional:
        print(f"{s.getpeername()}出现异常")
        # 停止inputs中对这个套接字的监听
        inputs.remove(s)
        if s in outputs:
            outputs.remove(s)
        # 移除消息队列
        del message_queue[s]

客户端

import socket
import time
 
client = socket.socket()
print("连接了")
client.connect(("localhost", 8888))
 
for i in range(1, 4):
    print(f"第{i}次循环")
    client.send(b"number %d" % i)
    data = client.recv(1024)
    print(str(data, encoding="utf-8"))
    time.sleep(1)

selectors：I/O多路复用抽象

selectors模块在select中平台特定的I/O监视函数之上提供了一个平台独立的函数

平台模型

selectors中的api是基于时间的，与select中的poll类似。它有多个实现，并且这个模块会自动设置别名DefaultSelector来指示当前系统配置最高的一个实现选择器对象提供了一些方法，可以指定在一个套接字是哪个查找哪些事件，然后以一种平台独立的方式让调用者等待事件。注册对事件会创建一个SelectorKey，其中包含套接字、所注册事件的有关信息，其中还有可选的应用数据。选择器的所有者调用它的select方法来了解事件。返回值是一个键对象序列和一个指示发生了那些事件的位掩码。使用选择器的程序要反复调用select，然后适当地处理事件

实例

import socket
from selectors import DefaultSelector, EVENT_READ, EVENT_WRITE
  
# 会根据当前的操作系统选择一个合适的文件描述符，Linux下是epoll，Windows下则是select
selector = DefaultSelector()
  
  
class Fetcher:
  
    def __init__(self, host):
        self.client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self.client.setblocking(False)  # 设置非阻塞socket
        self.host = host
  
    def get_url(self):
        try:
            self.client.connect((self.host, 80))
        except BlockingIOError:
            # 也可以做一些其他的事
            pass
  
        # 将socket注册一个事件，self.client.fileno()是当前socket的文件描述符,
        # EVENT_WRITE表示可写,因为我们连接建立好之后要send一个请求(相当于写数据)，所以等待一个可写状态
        # 并且绑定一个回调函数，如果连接建立好了，说明可以写了，那么就调用相应的回调函数
        selector.register(self.client.fileno(), EVENT_WRITE, self.send_request)
  
    def send_request(self, key):
        # 这里的参数key指的便是调用当前函数的socket
        # 调用了之后，我们就不再监视了，因此要取消注册，key.fd指的便是key(当前socket)的fd(file descriptor)
        selector.unregister(key.fd)
        self.client.send("GET {} HTTP/1.1
Host:{}
Connection:close

".format("/", self.host).encode("utf-8"))
        # 这时候便将我们的请求发出去了，但是我们还要继续注册新的socket
        # EVENT_READ指的是可读，我们发送一个请求之后要等待服务器返回数据(相当于读数据)，所以等待一个可读状态
        # 当满足可读状态，那么调用相应的回调函数读取数据
        selector.register(self.client.fileno(), EVENT_READ, self.read_data)
  
    def read_data(self, key):
        data = b""
        recv = self.client.recv(1024)
        if recv:
            data += recv
        else:
            # 值获取完毕，那么取消注册
            selector.unregister(key.fd)
  
        print(str(data, encoding="utf-8", errors="ignore"))
  
  
def loop_forever():
    # 事件循环，为什么需要事件循环
    # 因为我们上面绑定的回调函数没办法自动调用，所以我们必须创建一个事件循环，不断地从里面取出满足状态的socket
    while 1:
        ready = selector.select()
        for key, mask in ready:
            # key.data就是我们注册的回调的函数
            call_back = key.data
            # 将key也就是相应的socket主动传进去
            call_back(key)
  
  
if __name__ == '__main__':
    fetcher = Fetcher("www.baidu.com")
    fetcher.get_url()
    loop_forever()

socketserver：创建网络服务器

服务端

import socketserver
 
 
class Myserver(socketserver.BaseRequestHandler):
    '''
    定义一个类，该类必须继承socketserver下的BaseRequestHandler
    '''
    def handle(self):
        # 重写其内部的handle方法
        # 内部封装了self.request,就相当于socket当中的conn
        while True:
            recv = self.request.recv(1024)  # 接受到字节形式的内容
            if not recv:
                break
            print(str(recv, encoding="utf-8"))
            self.request.send(recv+bytes("我是你爸", encoding="utf-8"))
 
 
# 创建多任务server，每来一个连接我就创建一个线程与其交互
server = socketserver.ThreadingTCPServer(("localhost", 8080), Myserver)
server.serve_forever()

客户端1

import socket
 
client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
client.connect(("localhost", 8080))
 
while True:
    inp = input("请输入你要发送的内容：")
    client.send(bytes(inp, encoding="utf-8"))
    recv = client.recv(1024)
    print(str(recv, encoding="utf-8"))

客户端2

import socket
 
client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
client.connect(("localhost", 8080))
 
while True:
    inp = input("请输入你要发送的内容：")
    client.send(bytes(inp, encoding="utf-8"))
    recv = client.recv(1024)
    print(str(recv, encoding="utf-8"))

客户端3

import socket
 
client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
client.connect(("localhost", 8080))
 
while True:
    inp = input("请输入你要发送的内容：")
    client.send(bytes(inp, encoding="utf-8"))
    recv = client.recv(1024)
    print(str(recv, encoding="utf-8"))

多年以前写的博客，重新拷贝过来了。那时候很中二，原谅我的素质极差。

利用socketserver传输文件

服务端

import socketserver
 
 
class Myserver(socketserver.BaseRequestHandler):
    def handle(self):
        while True:
            try:
                print(f'--------------连接开始----------------')
                # 在发送文件之前，做一次交互进行确认
                recv = self.request.recv(1024)
                if not recv:
                    break
                print(str(recv, encoding="utf-8"))
                self.request.send(bytes("来自服务端的回复：我是loser，我准备好了，请发吧", encoding="utf-8"))
 
                file_size = self.request.recv(1024)  # 获取文件的大小
                if not file_size:
                    break
                print("文件大小:", file_size)
 
                file_name = str(self.request.recv(1024), encoding="utf-8")
                if not file_name:
                    break
                print("文件名为:", file_name)
 
                self.request.send(bytes("文件名和文件路径都已经接受完毕，请发送文件内容吧！！！", encoding="utf-8"))
 
                data_size = 0
 
                # 创建一个文件，用于写入客户端发来的文件
                f = open(file_name, "wb")
 
                while True:
                    recv_data = self.request.recv(1024)
                    f.write(recv_data)
                    data_size += len(recv_data)
                    if data_size >= int(file_size):
                        f.close()
                        break
 
                print("文件已经传输完毕，向客户端发送信息，告知客户端")
                self.request.send(bytes("文件收完了，断开连接吧！", encoding="utf-8"))
                # 传输完毕，主动断开连接
                print("服务端已经主动断开连接······")
                break
            except Exception:
                # 如果客户端断开连接，直接break，打印断开的连接
                print(f"出现异常，已经和{self.request}断开连接")
                break
            finally:
                print(f"和{self.request}之间的连接已经结束·······")
                print(f'--------------此次连接结束，等待下一个连接----------------')
 
 
# 创建多任务server，每来一个连接我就创建一个线程与其交互
server = socketserver.ThreadingTCPServer(("localhost", 8080), Myserver)
server.serve_forever()

客户端

import socket
import os
 
 
client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
client.connect(("localhost", 8080))
 
while True:
    try:
        client.send(bytes("客户端：loser，我要给你发送文件了，准备好了吗？", encoding="utf-8"))
        recv = client.recv(1024)
        print(str(recv, encoding="utf-8"))
        print("那个loser准备好了，那么我开始发吧")
        inp = input("请输入你要发送的文件的路径：")
        # 打开文件准备发送
        f = open(inp, "rb")
        data = f.read()
        file_size = len(data)  # 获取文件的大小
        file_name = os.path.basename(inp)   # 获取不包括路径的文件名
 
        # 将文件的大小发送给对方
        client.send(bytes(str(file_size), encoding="utf-8"))
 
        # 将文件名发送给对方
        client.send(bytes(file_name, encoding="utf-8"))
 
        # 等待服务端回复，发送文件
        recv = client.recv(1024)
        print(str(recv, encoding="utf-8"))
        if not recv:
            break
 
        client.send(data)
        recv = client.recv(1024)
        print(str(recv, encoding="utf-8"))
        break
    except Exception:
        print("已经断开连接······")
        break

运行结果

服务端运行结果
  
--------------连接开始----------------
客户端：loser，我要给你发送文件了，准备好了吗？
文件大小: b'18964'
文件名为: mashiro.jpg
文件已经传输完毕，向客户端发送信息，告知客户端
服务端已经主动断开连接······
和<socket.socket fd=336, family=AddressFamily.AF_INET, type=SocketKind.SOCK_STREAM, proto=0, laddr=('127.0.0.1', 8080), raddr=('127.0.0.1', 56121)>之间的连接已经结束·······
--------------此次连接结束，等待下一个连接----------------


客户端运行结果
  
来自服务端的回复：我是loser，我准备好了，请发吧
那个loser准备好了，那么我开始发吧
请输入你要发送的文件的路径：C:UsersAdministratorDesktopmashiro.jpg
文件名和文件路径都已经接受完毕，请发送文件内容吧！！！
文件收完了，断开连接吧