今日目录：
- 上周回顾
- 分享一个小知识点
- Python多继承知识补充
- socketserver的源码剖析
- 并发处理
IO多路复用(单独知识点)
多线程,多进程
一. 上周回顾:
1. socket：
- 1. 导入模块(import socket)
- 2. 创建socket
- 3. send 和 sendall
sendall本质上也是调用send
但是在内部做了一个while循环，拿到消息，循环发，直到消息完全发送为止
#sendall: while True: 10 = send('adkflskjf023jlskdf') 10 = send('adkflskjf023jlskdf') ...
send，一次不一定能完全发送
10 = send('adkflskjf023jlskdf')
PS: 所以以后使用都应该用sendall来发送
2. Python 3中发送 的是 字节， Python2中发送的是 字符串
recv(2048) 表示最多接受2048， 源码中最大接受8192字节
3. 解决粘包:
先发送ack确认消息
4. socketserver：
- 1. 自定义类
- 2. 继承方法
二. 分享一个小知识点
1. 某司的一个面试题：
l1 = [ x+100 for x in range(10)] #循环0-9, 每次的值x并+100,而后生成一个列表 print(l1) #执行结果： [100, 101, 102, 103, 104, 105, 106, 107, 108, 109]
l1 = [x + 100 for x in range(10) if x > 5] #当然也可以有判断条件, 如x>5的条件满足后, x才+100 print(l1) #代码执行结果: [106, 107, 108, 109]
结合lambda表达式来一个例子：
l1 = [lambda :x for x in range(10)] res = l1[0]() #加括号是因为调用列表中现存的值是函数，调用函数不就应该加()么？ print(res) #执行结果是9， 为啥呢？ 9
你瞅准了, 函数在不调用之前内部代码是不执行的。也就是说循环后，现在列表中有10个x，每个x都是一个未执行的函数：
l1 = [lambda :x for x in range(10)] # print(l1) [<function <listcomp>.<lambda> at 0x101bef1e0>, <function <listcomp>.<lambda> at 0x101bef268>, <function <listcomp>.<lambda> at 0x101bef2f0>, <function <listcomp>.<lambda> at 0x101bef378>, <function <listcomp>.<lambda> at 0x101bef400>, <function <listcomp>.<lambda> at 0x101bef488>, <function <listcomp>.<lambda> at 0x101bef510>, <function <listcomp>.<lambda> at 0x101bef598>, <function <listcomp>.<lambda> at 0x101bef620>, <function <listcomp>.<lambda> at 0x101bef6a8>]
看下上面的lambda表达式分开来写，是不是就是下面的代码：
l1 = [] for x in range(10): def f1(): return x #循环把x赋值，最后的值也是10个9 l1.append(f1) for i in l1: print(i())
循环调用验证下（很明显循环到最后一次时x被重新赋值9了）：
l1 = [lambda :x for x in range(10)] for i in l1: print(i()) #执行结果： 9 9 9 9 9 9 9 9 9 9
l1 = [] for x in range(10): def f1(): return x l1.append(f1) for i in l1: print(i()) #结果一样： 9 9 9 9 9 9 9 9 9 9
2. Python中的作用域
(1) Python中无块级作用域
这个块可以理解为代码块，如下代码：
if 1 == 1: name = 'Tom' print(name) #结果是可以打印出name的， 但是，条件必须为真了。 Tom
循环体也一样：
for i in range(10): name = i print(name) #执行结果： 9
但是在Java、c#中是有块级作用域的，不会打印出name的值。
(2) Python中以函数为作用域
def func(): name = 'Tom' func() print(name) #在函数外是不能使用的, 执行代码报错： NameError: name 'name' is not defined
(3) 作用域链
找的时候，由内往外找，直到找不到报错
name = 'Sam' def f1(): print(name) def f2(): name = 'Jack' f1() f2()
那么上面代码会打印出“Sam” 还是“Jack” ？？？ 嘿嘿， 是“Sam”了， 因为解释器解释的时候是从上到下解释， 还是上面的话：
f2调用f1函数， f1中print name， f1中没有，而后从内往外找，在上面找到了Sam
f2中的name在f2函数内部没有调用，和f1半毛钱关系都没有啦， so，当然是Sam!
三. Python中的多继承补充
PS: 本节主要是Python2.7多继承， 回顾下Python3.x的多继承
- Python3.x 中所有的类都默认继承 object类
- Python2.7中所有的类默认都不继承 object类
Python2.7中：
- 默认不支持， 不继承object的类 叫经典类；
- 添加object的，继承object的类，叫新式类；
Python3中：
- 所有的都是新式类，也就是默认的都继承object类
实例说话：
class M: pass class D(M): pass class E(M): pass class B(D): pass class C(E): pass class A(B,C): pass
Python3的查找图例:
Python2.7中的查找图：
(经典类)
新式类:
代码：（让M继承object类）
class M(object): pass class D(M): pass class E(M): pass class B(D): pass class C(E): pass class A(B,C): pass
新式类查找顺序图：
四. socketserver源码剖析：
继续上一篇博客socketserver往下说，上一篇只是知道了具体的用法，但是不知道什么意思，下面来深入源码剖析，先看下面代码：
import subprocess class MySocketServer(socketserver.BaseRequestHandler): def handle(self): self.request.sendall(bytes("Hello",encoding="utf-8")) while True: data=self.request.recv(1024) if len(data) ==0:break print("[%s] says:%s" %(self.client_address,data.decode())) cmd= subprocess.Popen(data.decode(),shell=True,stdout=subprocess.PIPE,stderr=subprocess.PIPE) cmd_res=cmd.stdout.read() if not cmd_res: cmd_data=bytes(cmd.stderr.read()) if len(cmd_res)==0: cmd_data=bytes("err cmd",encoding="utf8") self.request.send(cmd_res) if __name__ == '__main__': server=socketserver.ThreadingTCPServer(('127.0.0.1',8001),MyServer) server.serve_forever()
ThreadingTCPServer的类继承图：
上述代码流程：
启动服务端程序
- 执行 TCPServer.__init__ 方法，创建服务端Socket对象并绑定 IP 和 端口
- 执行 BaseServer.__init__ 方法，将自定义的继承自socketserver.BaseRequestHandler 的类MySocketServer赋值给 self.RequestHandlerClass
- 执行 BaseServer.server_forever 方法，While 循环一直监听是否有客户端请求到达 ...
- 执行 handle_request_noblock()
当客户端连接到达服务器
- 执行 ThreadingMixIn.process_request 方法，创建一个 “线程” 用来处理请求
- 执行 ThreadingMixIn.process_request_thread 方法
- 执行 BaseServer.finish_request 方法，
- 执行 self.RequestHandlerClass() 即：执行 自定义 MyServer 的构造方法（自动调用基类BaseRequestHandler的构造方法，在该构造方法中又会调用 MyServer的handle方法）
源码：
class BaseServer: """Base class for server classes. Methods for the caller: - __init__(server_address, RequestHandlerClass) - serve_forever(poll_interval=0.5) - shutdown() - handle_request() # if you do not use serve_forever() - fileno() -> int # for select() Methods that may be overridden: - server_bind() - server_activate() - get_request() -> request, client_address - handle_timeout() - verify_request(request, client_address) - server_close() - process_request(request, client_address) - shutdown_request(request) - close_request(request) - handle_error() Methods for derived classes: - finish_request(request, client_address) Class variables that may be overridden by derived classes or instances: - timeout - address_family - socket_type - allow_reuse_address Instance variables: - RequestHandlerClass - socket """ timeout = None 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 server_activate(self): """Called by constructor to activate the server. May be overridden. """ pass def serve_forever(self, poll_interval=0.5): """Handle one request at a time until shutdown. Polls for shutdown every poll_interval seconds. Ignores self.timeout. If you need to do periodic tasks, do them in another thread. """ self.__is_shut_down.clear() try: while not self.__shutdown_request: # XXX: Consider using another file descriptor or # connecting to the socket to wake this up instead of # polling. Polling reduces our responsiveness to a # shutdown request and wastes cpu at all other times. r, w, e = _eintr_retry(select.select, [self], [], [], poll_interval) if self in r: self._handle_request_noblock() finally: self.__shutdown_request = False self.__is_shut_down.set() def shutdown(self): """Stops the serve_forever loop. Blocks until the loop has finished. This must be called while serve_forever() is running in another thread, or it will deadlock. """ self.__shutdown_request = True self.__is_shut_down.wait() # The distinction between handling, getting, processing and # finishing a request is fairly arbitrary. Remember: # # - handle_request() is the top-level call. It calls # select, get_request(), verify_request() and process_request() # - get_request() is different for stream or datagram sockets # - process_request() is the place that may fork a new process # or create a new thread to finish the request # - finish_request() instantiates the request handler class; # this constructor will handle the request all by itself def handle_request(self): """Handle one request, possibly blocking. Respects self.timeout. """ # Support people who used socket.settimeout() to escape # handle_request before self.timeout was available. timeout = self.socket.gettimeout() if timeout is None: timeout = self.timeout elif self.timeout is not None: timeout = min(timeout, self.timeout) fd_sets = _eintr_retry(select.select, [self], [], [], timeout) if not fd_sets[0]: self.handle_timeout() return self._handle_request_noblock() def _handle_request_noblock(self): """Handle one request, without blocking. I assume that select.select has returned that the socket is readable before this function was called, so there should be no risk of blocking in get_request(). """ try: request, client_address = self.get_request() except socket.error: return if self.verify_request(request, client_address): try: self.process_request(request, client_address) except: self.handle_error(request, client_address) self.shutdown_request(request) def handle_timeout(self): """Called if no new request arrives within self.timeout. Overridden by ForkingMixIn. """ pass def verify_request(self, request, client_address): """Verify the request. May be overridden. Return True if we should proceed with this request. """ return True def process_request(self, request, client_address): """Call finish_request. Overridden by ForkingMixIn and ThreadingMixIn. """ self.finish_request(request, client_address) self.shutdown_request(request) def server_close(self): """Called to clean-up the server. May be overridden. """ pass def finish_request(self, request, client_address): """Finish one request by instantiating RequestHandlerClass.""" self.RequestHandlerClass(request, client_address, self) def shutdown_request(self, request): """Called to shutdown and close an individual request.""" self.close_request(request) def close_request(self, request): """Called to clean up an individual request.""" pass def handle_error(self, request, client_address): """Handle an error gracefully. May be overridden. The default is to print a traceback and continue. """ print '-'*40 print 'Exception happened during processing of request from', print client_address import traceback traceback.print_exc() # XXX But this goes to stderr! print '-'*40
class TCPServer(BaseServer): """Base class for various socket-based server classes. Defaults to synchronous IP stream (i.e., TCP). Methods for the caller: - __init__(server_address, RequestHandlerClass, bind_and_activate=True) - serve_forever(poll_interval=0.5) - shutdown() - handle_request() # if you don't use serve_forever() - fileno() -> int # for select() Methods that may be overridden: - server_bind() - server_activate() - get_request() -> request, client_address - handle_timeout() - verify_request(request, client_address) - process_request(request, client_address) - shutdown_request(request) - close_request(request) - handle_error() Methods for derived classes: - finish_request(request, client_address) Class variables that may be overridden by derived classes or instances: - timeout - address_family - socket_type - request_queue_size (only for stream sockets) - allow_reuse_address Instance variables: - server_address - RequestHandlerClass - socket """ address_family = socket.AF_INET socket_type = socket.SOCK_STREAM request_queue_size = 5 allow_reuse_address = False def __init__(self, server_address, RequestHandlerClass, bind_and_activate=True): """Constructor. May be extended, do not override.""" BaseServer.__init__(self, server_address, RequestHandlerClass) self.socket = socket.socket(self.address_family, self.socket_type) if bind_and_activate: try: self.server_bind() self.server_activate() except: self.server_close() raise def server_bind(self): """Called by constructor to bind the socket. May be overridden. """ if self.allow_reuse_address: self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) self.socket.bind(self.server_address) self.server_address = self.socket.getsockname() def server_activate(self): """Called by constructor to activate the server. May be overridden. """ self.socket.listen(self.request_queue_size) def server_close(self): """Called to clean-up the server. May be overridden. """ self.socket.close() def fileno(self): """Return socket file number. Interface required by select(). """ return self.socket.fileno() def get_request(self): """Get the request and client address from the socket. May be overridden. """ return self.socket.accept() def shutdown_request(self, request): """Called to shutdown and close an individual request.""" try: #explicitly shutdown. socket.close() merely releases #the socket and waits for GC to perform the actual close. request.shutdown(socket.SHUT_WR) except socket.error: pass #some platforms may raise ENOTCONN here self.close_request(request) def close_request(self, request): """Called to clean up an individual request.""" request.close() TCPServer
class ThreadingMixIn: """Mix-in class to handle each request in a new thread.""" # Decides how threads will act upon termination of the # main process daemon_threads = False def process_request_thread(self, request, client_address): """Same as in BaseServer but as a thread. In addition, exception handling is done here. """ try: self.finish_request(request, client_address) self.shutdown_request(request) except: self.handle_error(request, client_address) self.shutdown_request(request) def process_request(self, request, client_address): """Start a new thread to process the request.""" t = threading.Thread(target = self.process_request_thread, args = (request, client_address)) t.daemon = self.daemon_threads t.start()
class ThreadingTCPServer(ThreadingMixIn, TCPServer): pass
class BaseRequestHandler: """Base class for request handler classes. This class is instantiated for each request to be handled. The constructor sets the instance variables request, client_address and server, and then calls the handle() method. To implement a specific service, all you need to do is to derive a class which defines a handle() method. The handle() method can find the request as self.request, the client address as self.client_address, and the server (in case it needs access to per-server information) as self.server. Since a separate instance is created for each request, the handle() method can define arbitrary other instance variariables. """ 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
五. 并发处理
1. IO多路复用
I/O多路复用(又被称为“事件驱动”)，首先要理解的是，操作系统为你提供了一个功能，当你的某个socket可读或者可写的时候，它可以给你一个通知。这样当配合非阻塞的socket使用时，只有当系统通知我哪个描述符可读了，我才去执行read操作，可以保证每次read都能读到有效数据而不做纯返回-1和EAGAIN的无用功。写操作类似。操作系统的这个功能通过select/poll/epoll/kqueue之类的系统调用函数来使用，这些函数都可以同时监视多个描述符的读写就绪状况，这样，多个描述符的I/O操作都能在一个线程内并发交替地顺序完成，这就叫I/O多路复用，这里的“复用”指的是复用同一个线程。
PS: Linux中的 select，poll，epoll 都是IO多路复用的机制。
select
- select最早于1983年出现在4.2BSD中，它通过一个select()系统调用来监视多个文件描述符的数组，当select()返回后，该数组中就绪的文件描述符便会被内核修改标志位，使得进程可以获得这些文件描述符从而进行后续的读写操作。
- select目前几乎在所有的平台上支持，其良好跨平台支持也是它的一个优点，事实上从现在看来，这也是它所剩不多的优点之一。
- select的一个缺点在于单个进程能够监视的文件描述符的数量存在最大限制，在Linux上一般为1024，不过可以通过修改宏定义甚至重新编译内核的方式提升这一限制。
- 另外，select()所维护的存储大量文件描述符的数据结构，随着文件描述符数量的增大，其复制的开销也线性增长。同时，由于网络响应时间的延迟使得大量TCP连接处于非活跃状态，但调用select()会对所有socket进行一次线性扫描，所以这也浪费了一定的开销。
poll
- poll在1986年诞生于System V Release 3，它和select在本质上没有多大差别，但是poll没有最大文件描述符数量的限制。
- poll和select同样存在一个缺点就是，包含大量文件描述符的数组被整体复制于用户态和内核的地址空间之间，而不论这些文件描述符是否就绪，它的开销随着文件描述符数量的增加而线性增大。
- 另外，select()和poll()将就绪的文件描述符告诉进程后，如果进程没有对其进行IO操作，那么下次调用select()和poll()的时候将再次报告这些文件描述符，所以它们一般不会丢失就绪的消息，这种方式称为水平触发（Level Triggered）。
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()时便得到通知。
用select实现socket多线路连接
server端：
import socket import select sk=socket.socket() sk.bind(("127.0.0.1",8001)) sk.listen(5) while True: rlist,wlist,elist=select.select([sk,],[],[],1) print(rlist) #rlist中是socket 对象列表 【sk】 for i in rlist: conn,addr=i.accept() conn.sendall(bytes("hello",encoding="utf8"))
Client端
import socket sk=socket.socket() sk.connect(("127.0.0.1",8001)) data=sk.recv(1024) print(data.decode()) while True: input(">>>>") sk.close()
server端参数详解：
- 句柄列表rlist, 句柄列表wlist, 句柄列表elist = select.select([sk], 句柄序列2, 句柄序列3, 1是超时时间)
- 参数： 可接受四个参数（前三个必须）
- 返回值：三个列表
- select方法用来监视文件句柄，如果句柄发生变化，则获取该句柄。
- 1、当 参数sk序列中的句柄发生可读时（accetp和read），则获取发生变化的句柄并添加到 返回值1 序列中
- 2、当 参数2 序列中含有句柄时，则将该序列中所有的句柄添加到 返回值2 序列中
- 3、当 参数3 序列中的句柄发生错误时，则将该发生错误的句柄添加到 返回值3 序列中
- 4、当 超时时间 未设置，则select会一直阻塞，直到监听的句柄发生变化
- 当 超时时间 ＝ 1时，那么如果监听的句柄均无任何变化，则select会阻塞 1 秒，之后返回三个空列表，如果监听的句柄有变化，则直接执行。
实例二：
import socket import select sk=socket.socket() sk.bind(("127.0.0.1",8001)) sk.listen(5) inputs=[sk,] outputs=[] while True: rlist,wlist,e=select.select(inputs,outputs,[],1) print(len(inputs),len(rlist),len(outputs),len(wlist)) #rlist中是socket 对象列表 【sk】 #打印的第一个参数是 公有多少个连接，2，变化的rlist个数， 3，有操作变化个数，4，也是有操作变化个数 for i in rlist: #循环这个句柄，只要inputs,有变化，rlist就能取到 if i == sk: #判断是否相等sk，相等话证明就会有新的连接 conn,addr=i.accept() #然后建立连接 inputs.append(conn) #把此次连接的线路也进行监听 conn.sendall(bytes("hello",encoding="utf8")) #发送给客户端一个信息验证 else: #如果不等，就有可能是线路发生了变化 try: #程序运行正确 data=i.recv(1024) #接收客户端信息 print(data.decode()) if not data: # raise Exception("断开连接") else: #一些运行正常 ，把此次线路变化加到outputs列表中 outputs.append(i) except Exception as e: #如果有程序退出， 也需要把此次线路变化从监听中删除 print(e) inputs.remove(i) for w in wlist: #wlist 有变化证明接收到信息， w.sendall(bytes("response",encoding="utf8")) # 给客户端回个信息 outputs.remove(w) #然后删除此次监听，实现读写分离作用
import socket sk=socket.socket() sk.connect(("127.0.0.1",8001)) data=sk.recv(1024) print(data.decode()) while True: intt=input(">>>>") sk.sendall(bytes(intt,encoding="utf8")) print(sk.recv(1024)) sk.close()
其实select, 并不是真正实现并发，一直循环着（串行）在监听数据是否有变化，并把数据处理完毕之后才会去处理新的请求数据。如果每个请求的耗时比较长时，select版本的服务器端也无法完成同时操作，这种模式称之为伪并发。
PS: 能检测所有的IO操作，不单纯是socket， 但是文件操作除外。
select 系统内核底层维护了一个for循环，检测变化。在windows、linux平台都可以用，支持跨平台。有C10K问题，最大支持1024个
poll， 底层也是通过for循环实现。 但是没有个数限制了。但是for循环效率不高。
epoll，谁变化(回调函数)，告诉epoll谁发生了变化。
2. 多线程、多进程
- 1. 一个应用程序可以有多进程、多线程
- 2. 默认是单进程、单线程
- 3. 单进程，多线程，在Python中不会性能提升，在Java和C#中可以提升。
提高并发：
- 多线程： IO操作，一般不会用到CPU，效率提升是可以的
- 多进程：计算型操作， 需要占用CPU，因此性能不会有提升
线程和进程是一个很重要的概念：
进程（process）：
是计算机中已运行程序的实体。进程为曾经是分时系统的基本运作单位。在面向进程设计的系统（如早期的UNIX，Linux 2.4及更早的版本）中，进程是程序的基本执行实体；在面向线程设计的系统（如当代多数操作系统、Linux 2.6及更新的版本）中，进程本身不是基本运行单位，而是线程的容器。程序本身只是指令、数据及其组织形式的描述，进程才是程序（那些指令和数据）的真正运行实例。
线程（thread）：
是操作系统能够进行运算调度的最小单位。它被包含在进程之中，是进程中的实际运作单位。一条线程指的是进程中一个单一顺序的控制流，一个进程中可以并发多个线程，每条线程并行执行不同的任务
概念太吓人了，先来看一下进程，这个相对于线程来说还是稍微好理解一点的。进程，是程序运行的实体，这句话的意思是，程序是存放在硬盘中的，当这个程序运行时，就会产生若干个进程，并且这个进程是可见的
那么什么是 线程呢，线程是一个任务流，它被包含在进程之中
看一下python中线程：
import threading import time def f1(arg): time.sleep(5) #等待5秒钟 print(arg) #执行 t = threading.Thread(target=f1,args=(123,)) t.setDaemon(True) #True 表示主线程不等此子线程 t.start() #不代表当前线程会被立即执行 t.join(6) #表示主线程到此，等待。。直到子线程执行完毕 # 参数6，表示主线程在此最多等待6秒 print(33) print(33)
PS: 单进程，多线程，在Python中不会性能提升，在Java和C#中可以提升。可以用 多进程
因为Python(CPython)中有GIL，全局解释器锁的存在而导致的
创建线程知识点：
- obj.setDaemon(False) #True, 表示主线程不等子线程
- obj.join(num) #表示主线程到此等待。。。，直到子线程执行结束； num是指最多等待num秒
- obj.start() #不代表当前县城内会被立即执行