Python编程-网络编程进阶（IO复用、Socketserver）

时间：2023-03-10 03:34:42
Python编程-网络编程进阶（IO复用、Socketserver）

一、认证客户端的链接合法性

如果你想在分布式系统中实现一个简单的客户端链接认证功能，又不像SSL那么复杂，那么利用hmac+加盐的方式来实现。

服务端

from socket import *

import hmac,os

secret_key=b'linhaifeng bang bang bang'

def conn_auth(conn):

    '''

    认证客户端链接

    :param conn:

    :return:

    '''

    print('开始验证新链接的合法性')

    msg=os.urandom(32)

    conn.sendall(msg)

    h=hmac.new(secret_key,msg)

    digest=h.digest()

    respone=conn.recv(len(digest))

    return hmac.compare_digest(respone,digest)

def data_handler(conn,bufsize=1024):

    if not conn_auth(conn):

        print('该链接不合法,关闭')

        conn.close()

        return

    print('链接合法,开始通信')

    while True:

        data=conn.recv(bufsize)

        if not data:break

        conn.sendall(data.upper())

def server_handler(ip_port,bufsize,backlog=5):

    '''

    只处理链接

    :param ip_port:

    :return:

    '''

    tcp_socket_server=socket(AF_INET,SOCK_STREAM)

    tcp_socket_server.bind(ip_port)

    tcp_socket_server.listen(backlog)

    while True:

        conn,addr=tcp_socket_server.accept()

        print('新连接[%s:%s]' %(addr[0],addr[1]))

        data_handler(conn,bufsize)

if __name__ == '__main__':

    ip_port=('127.0.0.1',9999)

    bufsize=1024

    server_handler(ip_port,bufsize)

客户端(合法)

from socket import *

import hmac,os

secret_key=b'linhaifeng bang bang bang'

def conn_auth(conn):

    '''

    验证客户端到服务器的链接

    :param conn:

    :return:

    '''

    msg=conn.recv(32)

    h=hmac.new(secret_key,msg)

    digest=h.digest()

    conn.sendall(digest)

def client_handler(ip_port,bufsize=1024):

    tcp_socket_client=socket(AF_INET,SOCK_STREAM)

    tcp_socket_client.connect(ip_port)

    conn_auth(tcp_socket_client)

    while True:

        data=input('>>: ').strip()

        if not data:continue

        if data == 'quit':break

        tcp_socket_client.sendall(data.encode('utf-8'))

        respone=tcp_socket_client.recv(bufsize)

        print(respone.decode('utf-8'))

    tcp_socket_client.close()

if __name__ == '__main__':

    ip_port=('127.0.0.1',9999)

    bufsize=1024

    client_handler(ip_port,bufsize)

客户端(非法:不知道加密方式)

from socket import *

def client_handler(ip_port,bufsize=1024):

    tcp_socket_client=socket(AF_INET,SOCK_STREAM)

    tcp_socket_client.connect(ip_port)

    while True:

        data=input('>>: ').strip()

        if not data:continue

        if data == 'quit':break

        tcp_socket_client.sendall(data.encode('utf-8'))

        respone=tcp_socket_client.recv(bufsize)

        print(respone.decode('utf-8'))

    tcp_socket_client.close()

if __name__ == '__main__':

    ip_port=('127.0.0.1',9999)

    bufsize=1024

    client_handler(ip_port,bufsize)

客户端(非法:不知道secret_key)

from socket import *

import hmac,os

secret_key=b'linhaifeng bang bang bang1111'

def conn_auth(conn):

    '''

    验证客户端到服务器的链接

    :param conn:

    :return:

    '''

    msg=conn.recv(32)

    h=hmac.new(secret_key,msg)

    digest=h.digest()

    conn.sendall(digest)

def client_handler(ip_port,bufsize=1024):

    tcp_socket_client=socket(AF_INET,SOCK_STREAM)

    tcp_socket_client.connect(ip_port)

    conn_auth(tcp_socket_client)

    while True:

        data=input('>>: ').strip()

        if not data:continue

        if data == 'quit':break

        tcp_socket_client.sendall(data.encode('utf-8'))

        respone=tcp_socket_client.recv(bufsize)

        print(respone.decode('utf-8'))

    tcp_socket_client.close()

if __name__ == '__main__':

    ip_port=('127.0.0.1',9999)

    bufsize=1024

    client_handler(ip_port,bufsize)

二、IO多路复用

I/O多路复用指：通过一种机制，可以监视多个描述符，一旦某个描述符就绪（一般是读就绪或者写就绪），能够通知程序进行相应的读写操作。

1.Linux

Linux中的 select，poll，epoll 都是IO多路复用的机制。

（1）select

select最早于1983年出现在4.2BSD中，它通过一个select()系统调用来监视多个文件描述符的数组，当select()返回后，该数组中就绪的文件描述符便会被内核修改标志位，使得进程可以获得这些文件描述符从而进行后续的读写操作。

select目前几乎在所有的平台上支持，其良好跨平台支持也是它的一个优点，事实上从现在看来，这也是它所剩不多的优点之一。

select的一个缺点在于单个进程能够监视的文件描述符的数量存在最大限制，在Linux上一般为1024，不过可以通过修改宏定义甚至重新编译内核的方式提升这一限制。

另外，select()所维护的存储大量文件描述符的数据结构，随着文件描述符数量的增大，其复制的开销也线性增长。同时，由于网络响应时间的延迟使得大量TCP连接处于非活跃状态，但调用select()会对所有socket进行一次线性扫描，所以这也浪费了一定的开销。

（2）poll

poll在1986年诞生于System V Release 3，它和select在本质上没有多大差别，但是poll没有最大文件描述符数量的限制。

poll和select同样存在一个缺点就是，包含大量文件描述符的数组被整体复制于用户态和内核的地址空间之间，而不论这些文件描述符是否就绪，它的开销随着文件描述符数量的增加而线性增大。

另外，select()和poll()将就绪的文件描述符告诉进程后，如果进程没有对其进行IO操作，那么下次调用select()和poll()的时候将再次报告这些文件描述符，所以它们一般不会丢失就绪的消息，这种方式称为水平触发（Level Triggered）。

（3）epoll

直到Linux2.6才出现了由内核直接支持的实现方法，那就是epoll，它几乎具备了之前所说的一切优点，被公认为Linux2.6下性能最好的多路I/O就绪通知方法。

epoll可以同时支持水平触发和边缘触发（Edge Triggered，只告诉进程哪些文件描述符刚刚变为就绪状态，它只说一遍，如果我们没有采取行动，那么它将不会再次告知，这种方式称为边缘触发），理论上边缘触发的性能要更高一些，但是代码实现相当复杂。

epoll同样只告知那些就绪的文件描述符，而且当我们调用epoll_wait()获得就绪文件描述符时，返回的不是实际的描述符，而是一个代表就绪描述符数量的值，你只需要去epoll指定的一个数组中依次取得相应数量的文件描述符即可，这里也使用了内存映射（mmap）技术，这样便彻底省掉了这些文件描述符在系统调用时复制的开销。

另一个本质的改进在于epoll采用基于事件的就绪通知方式。在select/poll中，进程只有在调用一定的方法后，内核才对所有监视的文件描述符进行扫描，而epoll事先通过epoll_ctl()来注册一个文件描述符，一旦基于某个文件描述符就绪时，内核会采用类似callback的回调机制，迅速激活这个文件描述符，当进程调用epoll_wait()时便得到通知。

2.Python

Python中有一个select模块，其中提供了：select、poll、epoll三个方法，分别调用系统的 select，poll，epoll 从而实现IO多路复用。

  • Windows Python：

    提供： select
  • Mac Python：

    提供： select
  • Linux Python：

    提供： select、poll、epoll

注意：网络操作、文件操作、终端操作等均属于IO操作，对于windows只支持Socket操作，其他系统支持其他IO操作，但是无法检测普通文件操作，自动上次读取是否已经变化。

3.select方法

句柄列表11, 句柄列表22, 句柄列表33 = select.select(句柄序列1, 句柄序列2, 句柄序列3, 超时时间)

参数： 可接受四个参数（前三个必须）

返回值：三个列表

select方法用来监视文件句柄，如果句柄发生变化，则获取该句柄。

  • 当 参数1 序列中的句柄发生可读时（accetp和read），则获取发生变化的句柄并添加到 返回值1 序列中
  • 当 参数2 序列中含有句柄时，则将该序列中所有的句柄添加到 返回值2 序列中
  • 当 参数3 序列中的句柄发生错误时，则将该发生错误的句柄添加到 返回值3 序列中
  • 当 超时时间 未设置，则select会一直阻塞，直到监听的句柄发生变化
  • 当 超时时间 ＝ 1时，那么如果监听的句柄均无任何变化，则select会阻塞 1 秒，之后返回三个空列表，如果监听的句柄有变化，则直接执行。

利用select监听终端操作实例

import select

import threading

import sys

while True:

    readable, writeable, error = select.select([sys.stdin,],[],[],1)

    if sys.stdin in readable:

        print 'select get stdin',sys.stdin.readline()

利用select实现伪同时处理多个Socket客户端请求：服务端

import socket

import select

sk1 = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

sk1.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)

sk1.bind(('127.0.0.1',8002))

sk1.listen(5)

sk1.setblocking(0)

inputs = [sk1,]

while True:

    readable_list, writeable_list, error_list = select.select(inputs, [], inputs, 1)

    for r in readable_list:

        # 当客户端第一次连接服务端时

        if sk1 == r:

            print 'accept'

            request, address = r.accept()

            request.setblocking(0)

            inputs.append(request)

        # 当客户端连接上服务端之后，再次发送数据时

        else:

            received = r.recv(1024)

            # 当正常接收客户端发送的数据时

            if received:

                print 'received data:', received

            # 当客户端关闭程序时

            else:

                inputs.remove(r)

sk1.close()

利用select实现伪同时处理多个Socket客户端请求：客户端

import socket

ip_port = ('127.0.0.1',8002)

sk = socket.socket()

sk.connect(ip_port)

while True:

    inp = raw_input('please input:')

    sk.sendall(inp)

sk.close()

此处的Socket服务端相比与原生的Socket，他支持当某一个请求不再发送数据时，服务器端不会等待而是可以去处理其他请求的数据。但是，如果每个请求的耗时比较长时，select版本的服务器端也无法完成同时操作。

基于select实现socket服务端

'''

 服务器的实现 采用select的方式

'''

import select

import socket

import sys

import Queue

#创建套接字并设置该套接字为非阻塞模式

server = socket.socket(socket.AF_INET,socket.SOCK_STREAM)

server.setblocking(0)

#绑定套接字

server_address = ('localhost',10000)

print >>sys.stderr,'starting up on %s port %s'% server_address

server.bind(server_address)

#将该socket变成服务模式

#backlog等于5，表示内核已经接到了连接请求，但服务器还没有调用accept进行处理的连接个数最大为5

#这个值不能无限大，因为要在内核中维护连接队列

server.listen(5)

#初始化读取数据的监听列表,最开始时希望从server这个套接字上读取数据

inputs = [server]

#初始化写入数据的监听列表，最开始并没有客户端连接进来，所以列表为空

outputs = []

#要发往客户端的数据

message_queues = {}

while inputs:

    print >>sys.stderr,'waiting for the next event'

    #调用select监听所有监听列表中的套接字，并将准备好的套接字加入到对应的列表中

    readable,writable,exceptional = select.select(inputs,outputs,inputs)#列表中的socket 套接字  如果是文件呢？

    #监控文件句柄有某一处发生了变化 可写 可读  异常属于Linux中的网络编程

    #属于同步I/O操作，属于I/O复用模型的一种

    #rlist--等待到准备好读

    #wlist--等待到准备好写

    #xlist--等待到一种异常

    #处理可读取的套接字

    '''

        如果server这个套接字可读，则说明有新链接到来

        此时在server套接字上调用accept,生成一个与客户端通讯的套接字

        并将与客户端通讯的套接字加入inputs列表，下一次可以通过select检查连接是否可读

        然后在发往客户端的缓冲中加入一项，键名为:与客户端通讯的套接字，键值为空队列

        select系统调用是用来让我们的程序监视多个文件句柄(file descrīptor)的状态变化的。程序会停在select这里等待，

        直到被监视的文件句柄有某一个或多个发生了状态改变

        '''

    '''

        若可读的套接字不是server套接字,有两种情况:一种是有数据到来，另一种是链接断开

        如果有数据到来,先接收数据,然后将收到的数据填入往客户端的缓存区中的对应位置，最后

        将于客户端通讯的套接字加入到写数据的监听列表:

        如果套接字可读.但没有接收到数据，则说明客户端已经断开。这时需要关闭与客户端连接的套接字

        进行资源清理

        '''

    for s in readable:

        if s is server:

            connection,client_address = s.accept()

            print >>sys.stderr,'connection from',client_address

            connection.setblocking(0)#设置非阻塞

            inputs.append(connection)

            message_queues[connection] = Queue.Queue()

        else:

            data = s.recv(1024)

            if data:

                print >>sys.stderr,'received "%s" from %s'% \

                (data,s.getpeername())

                message_queues[s].put(data)

                if s not in outputs:

                    outputs.append(s)

            else:

                print >>sys.stderr,'closing',client_address

                if s in outputs:

                    outputs.remove(s)

                inputs.remove(s)

                s.close()

                del message_queues[s]

    #处理可写的套接字

    '''

        在发送缓冲区中取出响应的数据，发往客户端。

        如果没有数据需要写，则将套接字从发送队列中移除，select中不再监视

        '''

    for s in writable:

        try:

            next_msg = message_queues[s].get_nowait()

        except Queue.Empty:

            print >>sys.stderr,'  ',s,getpeername(),'queue empty'

            outputs.remove(s)

        else:

            print >>sys.stderr,'sending "%s" to %s'% \

            (next_msg,s.getpeername())

            s.send(next_msg)

    #处理异常情况

    for s in exceptional:

        for s in exceptional:

            print >>sys.stderr,'exception condition on',s.getpeername()

            inputs.remove(s)

            if s in outputs:

                outputs.remove(s)

            s.close()

            del message_queues[s]

三、SOCKETSERVER

1.SocketServer模块

SocketServer内部使用 IO多路复用 以及 “多线程” 和 “多进程” ，从而实现并发处理多个客户端请求的Socket服务端。即：每个客户端请求连接到服务器时，Socket服务端都会在服务器是创建一个“线程”或者“进程” 专门负责处理当前客户端的所有请求。

Python编程-网络编程进阶（IO复用、Socketserver）

（1）ThreadingTCPServer

ThreadingTCPServer实现的Soket服务器内部会为每个client创建一个 “线程”，该线程用来和客户端进行交互。

1）ThreadingTCPServer基础

使用ThreadingTCPServer:

创建一个继承自 SocketServer.BaseRequestHandler 的类

类中必须定义一个名称为 handle 的方法

启动ThreadingTCPServer

SocketServer实现服务器

import SocketServer

class MyServer(SocketServer.BaseRequestHandler):

    def handle(self):

        # print self.request,self.client_address,self.server

        conn = self.request

        conn.sendall('欢迎致电 10086，请输入1xxx,0转人工服务.')

        Flag = True

        while Flag:

            data = conn.recv(1024)

            if data == 'exit':

                Flag = False

            elif data == '0':

                conn.sendall('通过可能会被录音.balabala一大推')

            else:

                conn.sendall('请重新输入.')

if __name__ == '__main__':

    server = SocketServer.ThreadingTCPServer(('127.0.0.1',8009),MyServer)

    server.serve_forever()

客户端

import socket

ip_port = ('127.0.0.1',8009)

sk = socket.socket()

sk.connect(ip_port)

sk.settimeout(5)

while True:

    data = sk.recv(1024)

    print 'receive:',data

    inp = raw_input('please input:')

    sk.sendall(inp)

    if inp == 'exit':

        break

sk.close()

2）ThreadingTCPServer源码剖析

ThreadingTCPServer的类图关系如下：

Python编程-网络编程进阶（IO复用、Socketserver）

内部调用流程为：

  • 启动服务端程序
  • 执行 TCPServer.__init__ 方法，创建服务端Socket对象并绑定 IP 和 端口
  • 执行 BaseServer.__init__ 方法，将自定义的继承自SocketServer.BaseRequestHandler 的类 MyRequestHandle赋值给 self.RequestHandlerClass
  • 执行 BaseServer.server_forever 方法，While 循环一直监听是否有客户端请求到达 ...
  • 当客户端连接到达服务器
  • 执行 ThreadingMixIn.process_request 方法，创建一个 “线程” 用来处理请求
  • 执行 ThreadingMixIn.process_request_thread 方法
  • 执行 BaseServer.finish_request 方法，执行 self.RequestHandlerClass() 即：执行自定义 MyRequestHandler 的构造方法（自动调用基类BaseRequestHandler的构造方法，在该构造方法中又会调用 MyRequestHandler的handle方法）

2.实例

服务端

import SocketServer

class MyServer(SocketServer.BaseRequestHandler):

    def handle(self):

        # print self.request,self.client_address,self.server

        conn = self.request

        conn.sendall('欢迎致电 10086，请输入1xxx,0转人工服务.')

        Flag = True

        while Flag:

            data = conn.recv(1024)

            if data == 'exit':

                Flag = False

            elif data == '0':

                conn.sendall('通过可能会被录音.balabala一大推')

            else:

                conn.sendall('请重新输入.')

if __name__ == '__main__':

    server = SocketServer.ThreadingTCPServer(('127.0.0.1',8009),MyServer)

    server.serve_forever()

客户端

import socket

ip_port = ('127.0.0.1',8009)

sk = socket.socket()

sk.connect(ip_port)

sk.settimeout(5)

while True:

    data = sk.recv(1024)

    print 'receive:',data

    inp = raw_input('please input:')

    sk.sendall(inp)

    if inp == 'exit':

        break

sk.close()

3.ForkingTCPServer

ForkingTCPServer和ThreadingTCPServer的使用和执行流程基本一致，只不过在内部分别为请求者建立 “线程” 和 “进程”。

基本使用：

服务端

import SocketServer

class MyServer(SocketServer.BaseRequestHandler):

    def handle(self):

        # print self.request,self.client_address,self.server

        conn = self.request

        conn.sendall('欢迎致电 10086，请输入1xxx,0转人工服务.')

        Flag = True

        while Flag:

            data = conn.recv(1024)

            if data == 'exit':

                Flag = False

            elif data == '0':

                conn.sendall('通过可能会被录音.balabala一大推')

            else:

                conn.sendall('请重新输入.')

if __name__ == '__main__':

    server = SocketServer.ForkingTCPServer(('127.0.0.1',8009),MyServer)

    server.serve_forever()

客户端

import socket

ip_port = ('127.0.0.1',8009)

sk = socket.socket()

sk.connect(ip_port)

sk.settimeout(5)

while True:

    data = sk.recv(1024)

    print 'receive:',data

    inp = raw_input('please input:')

    sk.sendall(inp)

    if inp == 'exit':

        break

sk.close()

以上ForkingTCPServer只是将 ThreadingTCPServer 实例中的代码：

server = SocketServer.ThreadingTCPServer(('127.0.0.1',8009),MyRequestHandler)

变更为：

server = SocketServer.ForkingTCPServer(('127.0.0.1',8009),MyRequestHandler)

SocketServer的ThreadingTCPServer之所以可以同时处理请求得益于 select 和 os.fork 两个东西，其实本质上就是在服务器端为每一个客户端创建一个进程，当前新创建的进程用来处理对应客户端的请求，所以，可以支持同时n个客户端链接（长连接）。

4.socketserver实现并发

基于tcp的套接字，关键就是两个循环，一个链接循环，一个通信循环

socketserver模块中分两大类：server类（解决链接问题）和request类（解决通信问题）

server类：

Python编程-网络编程进阶（IO复用、Socketserver）

request类：

Python编程-网络编程进阶（IO复用、Socketserver）

继承关系:

Python编程-网络编程进阶（IO复用、Socketserver）

Python编程-网络编程进阶（IO复用、Socketserver）

Python编程-网络编程进阶（IO复用、Socketserver）

以下述代码为例，分析socketserver源码：

ftpserver=socketserver.ThreadingTCPServer(('127.0.0.1',8080),FtpServer)

ftpserver.serve_forever()

查找属性的顺序：ThreadingTCPServer->ThreadingMixIn->TCPServer->BaseServer

  • 实例化得到ftpserver，先找类ThreadingTCPServer的__init__,在TCPServer中找到，进而执行server_bind,server_active
  • 找ftpserver下的serve_forever,在BaseServer中找到，进而执行self._handle_request_noblock()，该方法同样是在BaseServer中
  • 执行self._handle_request_noblock()进而执行request, client_address = self.get_request()（就是TCPServer中的self.socket.accept()），然后执行self.process_request(request, client_address)
  • 在ThreadingMixIn中找到process_request，开启多线程应对并发，进而执行process_request_thread，执行self.finish_request(request, client_address)
  • 上述四部分完成了链接循环，本部分开始进入处理通讯部分，在BaseServer中找到finish_request,触发我们自己定义的类的实例化，去找__init__方法，而我们自己定义的类没有该方法，则去它的父类也就是BaseRequestHandler中找....

源码分析总结：

基于tcp的socketserver我们自己定义的类中的

  • self.server即套接字对象
  • self.request即一个链接
  • self.client_address即客户端地址

基于udp的socketserver我们自己定义的类中的

  • self.request是一个元组（第一个元素是客户端发来的数据，第二部分是服务端的udp套接字对象），如(b'adsf', <socket.socket fd=200, family=AddressFamily.AF_INET, type=SocketKind.SOCK_DGRAM, proto=0, laddr=('127.0.0.1', 8080)>)
  • self.client_address即客户端地址

FtpServer

import socketserver

import struct

import json

import os

class FtpServer(socketserver.BaseRequestHandler):

    coding='utf-8'

    server_dir='file_upload'

    max_packet_size=1024

    BASE_DIR=os.path.dirname(os.path.abspath(__file__))

    def handle(self):

        print(self.request)

        while True:

            data=self.request.recv(4)

            data_len=struct.unpack('i',data)[0]

            head_json=self.request.recv(data_len).decode(self.coding)

            head_dic=json.loads(head_json)

            # print(head_dic)

            cmd=head_dic['cmd']

            if hasattr(self,cmd):

                func=getattr(self,cmd)

                func(head_dic)

    def put(self,args):

        file_path = os.path.normpath(os.path.join(

            self.BASE_DIR,

            self.server_dir,

            args['filename']

        ))

        filesize = args['filesize']

        recv_size = 0

        print('----->', file_path)

        with open(file_path, 'wb') as f:

            while recv_size < filesize:

                recv_data = self.request.recv(self.max_packet_size)

                f.write(recv_data)

                recv_size += len(recv_data)

                print('recvsize:%s filesize:%s' % (recv_size, filesize))

ftpserver=socketserver.ThreadingTCPServer(('127.0.0.1',8080),FtpServer)

ftpserver.serve_forever()

FtpClient

import socket

import struct

import json

import os

class MYTCPClient:

    address_family = socket.AF_INET

    socket_type = socket.SOCK_STREAM

    allow_reuse_address = False

    max_packet_size = 8192

    coding='utf-8'

    request_queue_size = 5

    def __init__(self, server_address, connect=True):

        self.server_address=server_address

        self.socket = socket.socket(self.address_family,

                                    self.socket_type)

        if connect:

            try:

                self.client_connect()

            except:

                self.client_close()

                raise

    def client_connect(self):

        self.socket.connect(self.server_address)

    def client_close(self):

        self.socket.close()

    def run(self):

        while True:

            inp=input(">>: ").strip()

            if not inp:continue

            l=inp.split()

            cmd=l[0]

            if hasattr(self,cmd):

                func=getattr(self,cmd)

                func(l)

    def put(self,args):

        cmd=args[0]

        filename=args[1]

        if not os.path.isfile(filename):

            print('file:%s is not exists' %filename)

            return

        else:

            filesize=os.path.getsize(filename)

        head_dic={'cmd':cmd,'filename':os.path.basename(filename),'filesize':filesize}

        print(head_dic)

        head_json=json.dumps(head_dic)

        head_json_bytes=bytes(head_json,encoding=self.coding)

        head_struct=struct.pack('i',len(head_json_bytes))

        self.socket.send(head_struct)

        self.socket.send(head_json_bytes)

        send_size=0

        with open(filename,'rb') as f:

            for line in f:

                self.socket.send(line)

                send_size+=len(line)

                print(send_size)

            else:

                print('upload successful')

client=MYTCPClient(('127.0.0.1',8080))

client.run()

相关文章