Python自动化开发学习10

2023-01-31 05:01:27 开发 学习 自动化

上次讲了由于GIL锁的存在，python的多线程是假的，用的还是CPU的单核。Python的多线程只是利用了CPU的上下文切换，上下分切换也是占用CPU的。那么什么时候用多行程？
Python的多线程，适合io密集型的任务，不适合CPU密集型的任务。
IO操作不占用CPU，比如Socket这种网络编程的情景。
计算占用CPU，所以大量计算的情景下多线程反而更慢，额外消耗了CPU切换上下文的计算。

多进程的基本语法和多线程差不多：

import multiprocessing
import time

def show(name):
    time.sleep(2)
    print('hello', name)

if __name__ == '__main__':
    p = multiprocessing.Process(target=show, args=('Jack',))
    p.start()
    p.join()  # join的效果就是等待子进程执行完毕
    print('执行结束')

上面的例子，只模块名变了，其他都和多线程差不多。

父进程和子进程的id号

下面的例子打印了进程的id号：

import multiprocessing
import os, time

def info(title):
    print(title, 'module name:', __name__)  # 模块名
    time.sleep(0.3)  # 加点停顿，可以看出来，所有进程真的是并行处理的
    print(title, 'parent process:', os.getppid())  # 父进程号
    time.sleep(0.3)
    print(title, 'process id:', os.getpid())  # 进程号

def f(title):
    info(title)

if __name__ == '__main__':
    info('main')
    for i in range(10):  # 这次起10个进程
        p = multiprocessing.Process(target=f, args=('p%s' % i,))
        p.start()

可以适当修改加长info的延时，，可以去系统里查看一下所有进程的情况，如下图：
Python自动化开发学习10
上面起了10个子进程，加上主进程，一个11个python进程。
我是用PyCharm执行的代码，主进程的ID是8036，主进程的父进程是pycharm7832。
然后，所有的子进程，都是通过8036这个python的父进程开启的。8036就是这些子进程的父进程。

进程间通讯

进程间的内存是独立的，如果进程间需要交换数据，就需要借助其他方法

通过队列 - Queue

下面的例子通过Queue实现了进程间的通信，

from multiprocessing import Process, Queue
import time

def f(q):
    q.put([42, None, 'hello'])
    time.sleep(30)

if __name__ == '__main__':
    q = Queue()
    p = Process(target=f, args=(q,))
    p.start()
    print(q.get())  # prints "[42, None, 'hello']"
    time.sleep(30)

这里的Queue是 multiprocessing 多线程模块里的，不是之前的独立的queue模块。

通过管道 - Pipe

from multiprocessing import Process, Pipe

def f(conn):
    conn.send([42, None, 'hello'])
    conn.close()

if __name__ == '__main__':
    parent_conn, child_conn = Pipe()  # 实例化管道后会生成2个对象，这2个对象是一样的
    p = Process(target=f, args=(child_conn,))  # 把管道的1头交给子进程，自己操作另外一头
    p.start()
    print(parent_conn.recv())  # prints "[42, None, 'hello']"

上面实例化管道之后的两个对象是一样的，不要被名字误导，这么取名是随进程的关系。
实例化后的两个对象就是管道的两头。通讯的双方任意各取一头操作，就能实现管道两头的通讯。这里是父进程留下一头，把另一头传递给子进程操作。
这里管道的操作使用send和recv，类似socket（不过没有黏包）。这边send一次对端就recv一次获取数据，如果一边send多次，那么对端也只能一次recv取到一次的数据，所以也得recv多次才能取到全部的数据。如果数据取完了，再recv则会阻塞，等待管道对端send数据进来。

通过 Manager 共享

上面的2个方法，只是实现了数据的传递。
通过Manager生成的数据对象，可以在多个进程间共享。以下的数据类型都可以通过Manager来生成，实现共享：
list, dict, Namespace, Lock, RLock, Semaphore, BoundedSemaphore, Condition, Event, Barrier, Queue, Value and Array
举例说明：

from multiprocessing import Process, Manager
import os

def f(d, l):
    pid = os.getpid()
    d['pid'] = pid  # 每次这个key都会被重写，最后打印的一定是最后一个操作的进程的结果
    d[pid] = pid  # 每次都是生成一个新的key，添加到字典里
    l.append(pid)  # 每次往列表中添加一个元素

if __name__ == '__main__':
    with Manager() as manager:  # 等于manager = Manager()，最后还省了一个manager.close()
        d = manager.dict()  # 生成一个共享的字典
        l = manager.list()  # 生成一个共享的列表
        p_list = []
        for i in range(10):
            p = Process(target=f, args=(d, l))
            p.start()
            p_list.append(p)
        for res in p_list:
            res.join()
        print(d)
        print(l)

进程锁 Lock

线程有线程锁，进程也有进程锁。用法也和线程锁一样，如下：

from multiprocessing import Process, Lock
import time

def f(l, n):
    l.acquire()
    print('hello world', n, end='', flush=True)
    time.sleep(0.1)
    print(" Finished", n)
    l.release()
    time.sleep(0.1)

if __name__ == '__main__':
    lock = Lock()
    for n in range(10):
        Process(target=f, args=(lock, n)).start()

有锁，但是我们要锁什么？上面的代码就算不加锁应该也不一定会有什么问题。线程锁我们锁的是内存，因为线程共享的是同一块内存空间。进程锁锁的是资源，进程就是各种资源的集合。比如例子中的print用到的就是屏幕输出的资源，如果不加锁，可能几个进程同时想操作屏幕输出内容，那么就有可能会造成最终输出的字符错乱。不过打印数据应该是一串一次性进缓存的，应该也不会出现被插队吧。
虽然试不出字符错乱的情况，但是对两次print之间不要插入别的进程的内容。比如例子中去掉进程锁就会出现两段内容混乱的情况。
和线程锁的情况一样，加了锁之后，其实就应该是暂时变成了串行执行了。

进程池

每起一个进程，就是克隆一份父进程的基本数据给子进程用。这样开销会很大（比如内存），系统资源是有限的，无限起进程，可能会导致系统瘫痪。所以要有进程池。
学习线程的时候也有类似的问题，不过线程占用资源小，不容易导致系统瘫痪，但是一定会导致CPU频繁切换上下文导致效率反而会降低。所以要有信号量。
设定进程池，可以限制一次起的进程的数量。这点有点像信号量
进程池有两个方法：

  • apply ：还是串行
  • apply_async ：这个是并行

apply

先看apply的例子：

from multiprocessing import Pool
import time

def Foo(i):
    time.sleep(1)
    print("In Foo", i)

if __name__ == '__main__':
    pool = Pool(5)
    for i in range(10):
        pool.apply(func=Foo, args=(i,))
    print('end')
    pool.close()  # 老师说进程池这里要先close再join，至于为啥不知道
    pool.join()  # 进程池中进程执行完毕后再关闭，如果注释，那么程序直接关闭。

执行的时候就是同时起5个进程（Pool中设置了进程池的大小），但是是一个一个依次执行的。执行完毕后会退出然后可以将之后的进程放入进程池等待执行。
注意最后要先close再join。可以试验一下，如果不执行close，但是直接执行join的话会报错。

apply_rsync

使用这个方法就是并行效果了，一次并行执行5个。每完成1个会再将下一个放入进程池马上执行：

from multiprocessing import Pool
import time

def Foo(i):
    time.sleep(1)
    print("In Foo", i)

if __name__ == '__main__':
    pool = Pool(5)
    for i in range(10):
        pool.apply_async(func=Foo, args=(i,))
    print('end')
    pool.close()
    pool.join()  # 进程池中进程执行完毕后再关闭，如果注释，那么程序直接关闭。

注意：还是close和join的问题，刚才异步的时候其实不写也没问题。但是这里由于是并行的，主进程执行完毕之后如果没有join就会直接关闭了。子进程也不会再执行了，就像守护线程那样。所以这里一定要加上join，等待pool里的进程都执行完毕。然后join前必须要close。

回调函数

callback参数 ：这里还有一个callback参数，函数执行完成后可以调用另一个函数，这个叫回调函数。
回调函数 ：就是前面的方法执行完之后，就会自动对用执行这个回调函数。并且是由主进程调用执行的。
举例说明：

from multiprocessing import Pool
import os
import time

def Foo(i):
    time.sleep(1)
    # raise
    return os.getpid(), os.getppid()

def Bar(arg):
    print(os.getpid(), 'Foo 执行完毕，结果：', arg)

if __name__ == '__main__':
    pool = Pool(5)
    for i in range(10):
        pool.apply_async(func=Foo, args=(i,), callback=Bar)
    print('end')
    pool.close()
    pool.join()  # 进程池中进程执行完毕后再关闭，如果注释，那么程序直接关闭。

回调函数只有在函数正常执行完之后才会被调用。Foo中有一句raise，主动抛出一个错误，如果去掉注释导致函数没有正常执行完成，rais之前的print还是会正常执行，但是不会调用callback的函数执行。
另外，这里打印了每个进程id，从id中可以看到，Foo函数是由主进程启动的子进程执行的。而callback的函数是由主进程来执行的。Foo的父进程id就是Bar的进程id
回调函数的意义，主要就是因为回调函数是由主进程执行的。如果子进程的执行结果需要记录保留，那么这部分工作就通过调用回调函数，由回调函数在主进程中来处理。比如将结果写入数据库，我们就要让每个子进程都连接数据库写入数据，而是在主进程里建立一个与数据库的连接，统一将执行结果写入数据库。虽然调用的是同一个函数，但是通过回调函数调用在主进程中执行效率会更高。比如例子中的做法，Foo负责返回数据，回调函数统一打印Foo的执行结果。

协程，又称微线程，纤程。英文名Coroutine。一句话说明：协程是一种用户态的轻量级线程。
协程的好处：

  • 无需线程上下文切换的开销：就是没有CPU层面的上下文切换参考多线程的实现，我们在代码中实现切换
  • 无需原子操作锁定及同步的开销：就是不用加线程锁
  • 方便切换控制流，简化编程模型
  • 高并发+高扩展性+低成本：一个CPU支持上万的协程都不是问题。所以很适合用于高并发处理。

所谓原子操作是指不会被线程调度机制打断的操作；这种操作一旦开始，就一直运行到结束，中间不会有任何

协程的缺点：

  • 无法利用多核资源：协程的本质是个单线程，它不能同时将CPU 的多个核用上
  • 进行阻塞（Blocking）操作（如IO时）会阻塞掉整个程序

之前学习生成器的时候，通过yield实现了单线程下多并发。但是那也不是真正的协程。
协程的定义：

  1. 必须在只有一个单线程里实现并发
  2. 修改共享数据不需加锁
  3. 用户程序里自己保存多个控制流的上下文栈
  4. 一个协程遇到IO操作自动切换到其它协程

总结：使用协程就是为了高效。协程如何实现高效？一遇到IO操作就切换，因为IO操作耗时但是不占用CPU，此时切换到另一个协程，高效的利用CPU。
问题：何时切换回来？IO操作结束了就可以切换回来。如何知道IO操作结束了？往下学 ...

yield

这里写个例子，回顾一下yield的用法。但是yield并不满足我们前面对协程的定义。
下面的例子会先启动B，B会启动A。B中打印后切换到A执行，A返回后循环。A中打印后通过yield返回，循环。A和B之间通过yield和send来传递count的值，每次都自增1。

import time

def print_A():
    count = 0
    while True:
        print('A'.center(9, '-'), count)
        time.sleep(0.1)
        count = yield count+1  # 从B那里send过来的值，赋值给count。然后回到开头执行打印，把count值自增1后再返回给B

def print_B(func):
    count = next(func)  # 要先next一下，启动A，这样A会先运行一次到yield的地方返回
    while True:
        print('B'.center(9, '-'), count)
        time.sleep(0.1)
        count = func.send(count+1)  # 将值传递给A的yield，并获取A的返回值

if __name__ == '__main__':
    a = print_A()
    print_B(a)

greenlet 模块

greenlet是一个用C实现的协程模块，相比与python自带的yield，它可以使你在任意函数之间随意切换，而不需把这个函数先声明为generator。
这个是第三方模块，所以需要安装。安装的话直接安装下面要讲的gevent模块就好了。
greenlet 模块实现的是协程的手动切换，其实就和yield差不多。不过用起来更好理解了。
gevent 模块才能实现我们要的自动切换，但是gevent是在greenlet的基础上进行了封装，实现了自动切换。所以安装的时候顺便把有依赖关系的greenlet模块一起装好了。现在自然也是先看一下greenlet模块的用法。

from greenlet import greenlet

def test1():
    print('A1')
    gr2.switch()
    print('A2')
    gr2.switch()

def test2():
    print("B1")
    gr1.switch()
    print('B2')

if __name__ == '__main__':
    gr1 = greenlet(test1)  # 启动一个协程
    gr2 = greenlet(test2)  # 再启动一个协程
    gr1.switch()  # 手动切换一下

执行一下switch，就完成了切换的操作。比yield更直观。不过重点是学下一个模块。

gevent 模块

Gevent 是一个第三方库，可以轻松通过gevent实现并发同步或异步编程，在gevent中用到的主要模式是Greenlet，它是以C扩展模块形式接入Python的轻量级协程。Greenlet全部运行在主程序操作系统进程的内部，但它们被协作式地调度。

import gevent

def func1():
    print("This is in func1")
    gevent.sleep(2)
    print("End of func1")

def func2():
    print("This is in func2")
    gevent.sleep(1)
    print("End of func2")

if __name__ == '__main__':
    gevent.joinall([gevent.spawn(func1),
                    gevent.spawn(func2)])
    # 上面是把要启动的方法加到列表里一起处理了，推荐就这么做
    # 其实也可以使用start和join一个一个启动，
    f1 = gevent.spawn(func1)
    f2 = gevent.spawn(func2)
    f1.start()  # 没有start，貌似也一样，注释掉试一下
    f2.start()
    f1.join()  # 没有join主线程会直接退出，就不会等待上面的协程的执行结果了
    f2.join()

要启动所有的协程，通过gevent.joinall。参数是一个列表，列表中依次启动需要进行自动切换操作的协程。并且阻塞等待所有协程处理完毕。相当于start和join
上面的gevent.sleep是一个模拟IO操作，不会像time.sleep那样停在那里，而是会当做有一个几秒的IO操作。上面的输出结果是：

 This is in func1
This is in func2
End of func2
End of func1

首先执行了func1，打印了第一行。然后之后是一个IO操作，所以切换到了下一个协程。
切换到func2，打印了一行，之后又是一个IO操作，此时再切换。不过此时已经没有可操作的协程了。没别的协程了，fun1也没好。
之后是fun2的IO操作先执行完毕，所以最终切换到fun2的时候，打印了func2的第二行，打印前会顿1秒。
最后func1的IO操作也结束了，于是切换到fun1，打印fun1的第二行，打印前会再顿1秒。

并发爬网页

这根本就不是爬虫，这里先讲如何将一个网页保存到本地，因为这就是一个比较耗时的IO操作。刚才的例子中我们是用sleep来模拟的。直接上代码就好了：

from urllib import request

def f(url):
    print('GET: %s' % url)
    resp = request.urlopen(url)  # 给网址发一个请求
    data = resp.read()  # 读取到的就是整个网页的内容
    with open('url.html', 'wb') as file:  # 将网页保存下来
        file.write(data)
    print('%d betes received from %s' % (len(data), url))

if __name__ == '__main__':
    f('Http://www.python.org/')

然后我们来多爬几个网页，看下协程的效果。这次计算一下整个过程的时间：

from urllib import request
import gevent
import time

def f(url):
    print('GET: %s' % url)
    resp = request.urlopen(url)  # 给网址发一个请求
    data = resp.read()  # 读取到的就是整个网页的内容
    # with open('url.html', 'wb') as file:  # 保存网页的操作就先不用了
    #     file.write(data)
    print('%d betes received from %s' % (len(data), url))

if __name__ == '__main__':
    g_list = []
    url_list = ['http://www.python.org/', 'http://www.yahoo.com/', 'http://GitHub.com/']
    for url in url_list:
        g_list.append(gevent.spawn(f, url))
    start_time = time.time()
    gevent.joinall(g_list)
    print('运行时间：', time.time()-start_time)

上面这段虽然用了gevent，但是还是串行的。到底是串行还是并行，只要看f函数开始的时候的第一句print是什么时候出现的就知道了。这里之所以是串行，是因为，并没有看到IO切换的命令，就是f函数里没有类似gevent.sleep这样的切换命令。但是，其实gevent是可以自动判断是否有IO操作的。所以这里的问题是gevent发现不了urllib模块里的IO操作。
所以真正要做的是在开头加上一句，让gevent能够发现这些IO操作。

from urllib import request
import gevent
import time

# 导入模块，添加下面这2句
from gevent import monkey
monkey.patch_all()  # 把当前程序的所有的IO操作给我单独的做上标记

def f(url):
    print('GET: %s' % url)
    resp = request.urlopen(url)  # 给网址发一个请求
    data = resp.read()  # 读取到的就是整个网页的内容
    # with open('url.html', 'wb') as file:  # 保存网页的操作就先不用了
    #     file.write(data)
    print('%d betes received from %s' % (len(data), url))

if __name__ == '__main__':
    g_list = []
    url_list = ['http://www.python.org/', 'http://www.yahoo.com/', 'http://github.com/']
    for url in url_list:
        g_list.append(gevent.spawn(f, url))
    start_time = time.time()
    gevent.joinall(g_list)
    print('运行时间：', time.time()-start_time)

高并发的socket_server

gevent里有专门的socket模块，当然其中大部分都是import原生的socket模块。我们可以使用gevent的socket实现一个单线程下高并发的socket_server。
服务端：

# 就算这里不导入socket，也会在gevent模块里导入。但是pycharm里下面的socket.socket会显示个错误，不影响运行
# 而且到下面也是要导入的，这里显示的声明一下，提前导入也不会影响效率
# import socket  # 建议导入
# 下面两句也可以不要，这里不加也能识别到socket的IO操作
# 只所以能识别，是因为下面的socket已经不是原生的socket了，而是gevent修改后的socket
# from gevent import monkey
# monkey.patch_all()

import gevent
from gevent import socket

def server(port):
    # 这里pycharm里会显示个错误，不影响运行。因为gevent的socket模块里没有socket这个方法
    # 但是其实gevent会把原生的socket全部导入的。就是运行的时候会有socket.socket这个方法
    server = socket.socket()
    server.bind(('localhost', port))
    server.listen(500)
    print("监听已经开始")
    while True:
        conn, addr = server.accept()
        print("发现连接请求：\n%s\n%s" % (conn, addr))
        gevent.spawn(handle_request, conn)

# 上面的函数建立了连接后，就将连接作为handle_request的参数，启动一个gevent的协程
# 下面的方法是通过协程启动的，是协程并发运行的
def handle_request(conn):
    while True:
        data = conn.recv(1024)
        if not data: break
        print("recv:", data.decode('utf-8'))
        conn.send(data.upper())
    conn.close()
    print("断开连接：", conn)

if __name__ == '__main__':
    server(8002)

客户端只需要之前的客户端就可以了。这里测试一下效率，起100个线程，每个线程发送100条消息：

import socket
import threading

HOST = 'localhost'  # The remote host
PORT = 8002  # The same port as used by the server
def client(i):
    client = socket.socket()
    client.connect((HOST, PORT))
    for j in range(100):
        msg = "hello %s %s" % (i, j)
        client.send(msg.encode('utf-8'))
        data = client.recv(1024)
        print('Received:', data.decode('utf-8'))
    client.close()

if __name__ == '__main__':
    for i in range(100):
        t = threading.Thread(target=client, args=(i,))
        t.start()

这里有一个问题，记一下。 gevent.spawn()

通常，我们写服务器处理模型的程序时，有以下几种模型：

  • 每收到一个请求，创建一个新的进程，来处理该请求
  • 每收到一个请求，创建一个新的线程，来处理该请求
  • 每收到一个请求，放入一个事件列表，让主进程通过非阻塞I/O方式来处理请求。就是协程

三种方法各有千秋，之前应该都说过，这里就当总结一下：

  • 进程，由于创建新的进程的开销比较大，所以，会导致服务器性能比较差,但实现比较简单。
  • 线程，由于要涉及到线程的同步，有可能会面临死锁等问题。
  • 协程，在写应用程序代码时，逻辑比前面两种都复杂。

事件驱动模型，目前大部分的UI编程都是事件驱动模型，如很多UI平台都会提供onClick()事件（比如WEB页面），这个事件就代表鼠标按下事件。事件驱动模型大体思路如下：

  1. 有一个事件（消息）队列
  2. 鼠标按下时，往这个队列中增加一个点击事件（消息）
  3. 有个循环，不断从队列取出事件，根据不同的事件，调用不同的函数，如onClick()、onKeyDown()等
  4. 事件（消息）一般都各自保存各自的处理函数指针，这样，每个消息都有独立的处理函数

事件驱动编程范式

事件驱动编程是一种编程范式，这里程序的执行流由外部事件来决定。它的特点是包含一个事件循环，当外部事件发生时使用回调机制来触发相应的处理。另外两种常见的编程范式是（单线程）同步以及多线程编程。
在面对如下的环境时，事件驱动模型通常是一个好的选择：

  1. 程序中有许多任务
  2. 任务之间高度独立（因此它们不需要互相通信，或者等待彼此）
  3. 在等待事件到来时，某些任务会阻塞

此处重申一下协程开篇提出的问题，只要一遇到IO就注册一个事件，然后主程序就可以继续干其它的事情了，直到IO处理完毕后，继续恢复之前中断的任务，这本质上是怎么实现的呢？

五种IO模式

用户空间与内核空间：现在操作系统都是采用虚拟存储器，那么对32位操作系统而言，它的寻址空间（虚拟存储空间）为4G（2的32次方）。操作系统的核心是内核，独立于普通的应用程序，可以访问受保护的内存空间，也有访问底层硬件设备的所有权限。为了保证用户进程不能直接操作内核（kernel），保证内核的安全，操作系统将虚拟空间划分为两部分，一部分为内核空间，一部分为用户空间。针对linux操作系统而言，将最高的1G字节（从虚拟地址0xC0000000到0xFFFFFFFF），供内核使用，称为内核空间，而将较低的3G字节（从虚拟地址0x00000000到0xBFFFFFFF），供各个进程使用，称为用户空间。
缓存 I/O ：又被称作标准 I/O，大多数文件系统的默认 I/O 操作都是缓存 I/O。在 Linux 的缓存 I/O 机制中，操作系统会将 I/O 的数据缓存在文件系统的页缓存（ page cache ）中，也就是说，数据会先被拷贝到操作系统内核的缓冲区中，然后才会从操作系统内核的缓冲区拷贝到应用程序的地址空间。

IO模式

对于一次IO访问（以read举例），数据会先被拷贝到操作系统内核的缓冲区中，然后才会从操作系统内核的缓冲区拷贝到应用程序的地址空间。所以说，当一个read操作发生时，它会经历两个阶段：

  1. 等待数据准备 (Waiting for the data to be ready)
  2. 将数据从内核拷贝到进程中 (Copying the data from the kernel to the process)

正是因为上面的两个阶段，linux系统产生了下面五种网络模式的方案：

  • 阻塞 I/O（blocking IO） ：就是最最简单的串行，效率低实现简单
  • 非阻塞 I/O（nonblocking IO） ：第一阶段不会阻塞，需要反复主动发起询问数据是否准备好，第二阶段会阻塞
  • I/O 多路复用（ IO multiplexing） ：还是完全的阻塞，没什么优越性，但是可以同时处理多个连接。后面详讲
  • 信号驱动 I/O（ signal driven IO） ：略过没讲，第一阶段不阻塞，等待告知数据已经准备好，第二阶段会阻塞
  • 异步 I/O（asynchronous IO） ：完全没有阻塞，看似是最好的方式。但是用的不多。asyncio模块在python3.4引入标准库，支持异步IO

五种模型的比较：
Python自动化开发学习10
用的最多的是IO多路复用。虽然看似异步IO更好，反正用的不多。另外多线程+阻塞模式也是一个方案，但是多线程的开销较大（相对于单线程），更适合处理少量的并发（多少算少？看你系统能起多少个线程，不过和进程比线程的开销还不算特别大）。要处理高并发，推荐还是使用IO多路复用。

IO多路复用

文件描述符（File descriptor） ：是计算机科学中的一个术语，是一个用于表述指向文件的引用的抽象化概念。文件描述符在形式上是一个非负整数。实际上，它是一个索引值，指向内核为每一个进程所维护的该进程打开文件的记录表。当程序打开一个现有文件或者创建一个新文件时，内核向进程返回一个文件描述符。在程序设计中，一些涉及底层的程序编写往往会围绕着文件描述符展开。但是文件描述符这一概念往往只适用于UNIX、Linux这样的操作系统。
I/O多路复用就是通过一种机制，一个进程可以监视多个描述符，一旦某个描述符就绪（一般是读就绪或者写就绪），就通知程序进行相应的读写操作。
IO多路复用的三种机制：
select ：最早出现，有些缺点，优势就是几乎在所有平台上都支持。
pool ：解决了部分缺点，但是本质上没多大差别，可以认为是个过度阶段。
epool ：现在用这个，性能最好的。但是不是所有系统都支持，windows就不支持。

写一个非阻塞的socket

Python的select()方法直接调用操作系统的IO接口，它监控sockets，open files，and pipes（所有带fileno()方法的文件句柄）何时变成readable 和writeable，或者通信错误，select()使得同时监控多个连接变的简单，并且这比写一个长循环来等待和监控多客户端连接要高效，因为select直接通过操作系统提供的C的网络接口进行操作，而不是通过Python的解释器。
写一个socket的例子来理解
注意：socket必须得运行在非阻塞模式下。之前用的都是默认的阻塞模式，阻塞模式下，如果没有数据会等待。非阻塞模式下，如果没有数据就会抛出异常。所以我们需要用select来帮我们监视

开启监听

先写到accept之前，accept之前只是设置，执行到accept在阻塞模式下会进入阻塞。我们暂时只要能收到客户端请求建立起连接就好。现在是阻塞模式，所以直接accept没有数据就会报错。所以就需要select来解决了，监视到有活动的连接再返回并继续执行accept。

import select
import socket

server = socket.socket()
server.bind(('localhost', 9000))
server.listen()
print("监听已经开启")
server.setblocking(False)  # 设置为非阻塞
inputs = [server, ]
outputs = []
# 3个参数，读列表，写列表，异常列表，就是你想让内核监视哪些链接
# 异常列表监视的还是链接返回的内容，那么还是在inputs里，所以第三个参数还是填inputs
# 开始什么都没有，只有server，先把server加到inputs里
# 就是select监视到server活动了，就可以返回了
# 返回3个数据，监视到有活动的3个列表（读列表，写列表，异常列表）
readable, writeable, exceptional = select.select(inputs, outputs, inputs)
# 这里就是阻塞的，直到select监视到列表中有活动的链接，才会继续
# 非阻塞socket就是通过select来实现阻塞
print(readable, writeable, exceptional)  # 下面有打印的内容
# [<socket.socket fd=472, family=AddressFamily.AF_INET, type=SocketKind.SOCK_STREAM, proto=0, laddr=('127.0.0.1', 9000)>] [] []
# 第一个fd，就是文件描述符，非负整数。
conn, addr = server.accept()
print(conn, addr)

测试的客户端，只发不收： 

import socket
HOST, PORT = "localhost", 9000
client = socket.socket()
client.connect((HOST, PORT))
while True:
    msg = input(">>:").strip()
    if len(msg) == 0: break
    client.send(msg.encode('utf-8'))
    # 下面2行是接收服务的返回是要用的，暂时先注释掉，到后面再启用测试返回数据
    # data = client.recv(1024)
    # print('Received:', data.decode('utf-8'))
client.close()

接收一条数据

上面链接已经可以建立起来了，那么来接收一条数据吧。这里recv之前当然还是要用select来监视是否有数据进来，有才会执行到recv。

import select
import socket

server = socket.socket()
server.bind(('localhost', 9000))
server.listen()
print("监听已经开启")
server.setblocking(False)  # 设置为非阻塞

inputs = [server, ]
outputs = []
# 3个参数，读列表，写列表，异常列表，就是你想让内核监视哪些链接
# 异常列表监视的还是链接返回的内容，那么还是在inputs里，所以第三个参数还是填inputs
# 如果inputs里是读的消息，会返回到readable列表里，如果是异常消息，就返回到exceptional列表里
# 开始什么都没有，只有server，先把server加到inputs里
# 就是select监视到server活动了，就可以返回了
# 返回3个数据，监视到活动的3个列表（读列表，写列表，异常列表）
readable, writeable, exceptional = select.select(inputs, outputs, inputs)
# 这里就是阻塞的，直到select监视到列表中有活动的链接，才会继续
# 非阻塞socket就是通过select来实现阻塞
print(readable, writeable, exceptional)  # 打印一下监视到的活动链接
for r in readable:  # 可能一次返回多个连接啊，所以得写个循环
    conn, addr = server.accept()  # 已经是非阻塞了，不经过select监视就会报错。可以注释掉select试一下
    print(conn, addr)
    inputs.append(conn)  # 把链接加入监视列表
    # 现在conn加到了inputs的监视列表里了，就可以通过select监视conn是否有数据进来了
    readable, writeable, exceptional = select.select(inputs, outputs, inputs)
    print(readable, writeable, exceptional)  # 看看打印内容，inputs里的活动链接是conn，其实也可能是server
    data = conn.recv(1024)
    print(data.decode('utf-8'))

这里可以再试一下，应该可以接收到数据。但是这里有个问题。因为现在inputs列表里监视的内容是2个了，一个是server，一个是conn。如果再监视到server活动，说明又有新链接进来，如果监视到的是conn的活动，那么才是收到数据了。所以我们要在for循环里判断收到的活动链接是谁的。

可以接收多个连接请求

再多做一步，加上一层while循环，让服务端始终处于这么一个循环之中：select返回活动链接 ==> for循环处理所有的活动链接 循环继续。

import select
import socket

server = socket.socket()
server.bind(('localhost', 9000))
server.listen()
print("监听已经开启")
server.setblocking(False)  # 设置为非阻塞

inputs = [server, ]
outputs = []
# 3个参数，读列表，写列表，异常列表，就是你想让内核监视哪些链接
# 异常列表监视的还是链接返回的内容，那么还是在inputs里，所以第三个参数还是填inputs
# 开始什么都没有，只有server，先把server加到inputs里
# 就是select监视到server活动了，就可以返回了
# 返回3个数据，监视到活动的3个列表（读列表，写列表，异常列表）
while True:  # select返回活动链接==>for循环处理所有的活动链接，循环往复
    readable, writeable, exceptional = select.select(inputs, outputs, inputs)
    # 这里就是阻塞的，直到select监视到列表中有活动的链接，才会继续
    # 非阻塞socket就是通过select来实现阻塞
    print(readable, writeable, exceptional)  # 下面有打印的内容
    # [<socket.socket fd=472, family=AddressFamily.AF_INET, type=SocketKind.SOCK_STREAM, proto=0, laddr=('127.0.0.1', 9000)>] [] []
    # 第一个fd，就是文件描述符，非负整数。
    for r in readable:  # 可能一次返回多个连接啊，所以得写个循环
        # 现在inputs里有server 和 conn了，
        # 如果readable返回的是server的活动，表示来了一个新链接
        # 如果readable返回的是conn的活动，表示收到了conn发来的数据
        if r is server:
            conn, addr = server.accept()  # 已经是非阻塞了，不经过select监视就会报错。可以注释掉select试一下
            print(conn, addr)
            inputs.append(conn)  # 把链接加入监视列表
        else:
            data = conn.recv(1024)
            print(data.decode('utf-8'))

现在新链接也能再连上来了，但是问题是旧的连接没有另外保存，最新连上的链接会再次赋值给conn。就链接发来的数据，导致select返回，但是会用conn去尝试recv。现在conn是新的连接，所以是空的，于是就报错。那么解决这个事情就是要保存每一个conn，就是说要再用一个列表保存所有的conn，再写一个for循环？我一开始是这么想的。
其实所有的连接都保存在inputs里了，for循环的时候就是取出每一个链接，在for循环里面应该使用变量r，而不是conn。难怪之前用conn的时候，pycharm会提示 ‘name can not be defined’ 。第一次循环的时候没有conn这个变量，不过也不会进到那个if里

将收到的数据发回给客户端

接下来继续，我们可以直接把数据发回去，这里没有阻塞的问题，就不搞了。换个方法，不直接发回去，先把消息保存到队列里。然后统一发送，这样就是发消息也是多路复用的形式。

import select
import socket
import queue

server = socket.socket()
server.bind(('localhost', 9000))
server.listen()
print("监听已经开启")
server.setblocking(False)  # 设置为非阻塞

inputs = [server, ]
outputs = []
data_queue = {}  # 存放返回给客户端消息的队列，每个客户端连接一个队列，就是一个item
# 3个参数，读列表，写列表，异常列表，就是你想让内核监视哪些链接
# 异常列表监视的还是链接返回的内容，那么还是在inputs里，所以第三个参数还是填inputs
# 开始什么都没有，只有server，先把server加到inputs里
# 就是select监视到server活动了，就可以返回了
# 返回3个数据，监视到活动的3个列表（读列表，写列表，异常列表）
while True:  # select返回活动链接==>for循环处理所有的活动链接，循环往复
    readable, writeable, exceptional = select.select(inputs, outputs, inputs)
    # 这里就是阻塞的，直到select监视到列表中有活动的链接，才会继续
    # 非阻塞socket就是通过select来实现阻塞
    print('select返回：\n', readable, writeable, exceptional)  # 下面有打印的内容
    # [<socket.socket fd=472, family=AddressFamily.AF_INET, type=SocketKind.SOCK_STREAM, proto=0, laddr=('127.0.0.1', 9000)>] [] []
    # 第一个fd，就是文件描述符，非负整数。
    for r in readable:  # 可能一次返回多个连接啊，所以得写个循环
        # 现在inputs里有server 和 conn了，
        # 如果readable返回的是server的活动，表示来了一个新链接
        # 如果readable返回的是conn的活动，表示收到了conn发来的数据
        if r is server:
            conn, addr = r.accept()  # r就是server
            print('接收到新的客户端连接：\n', conn, addr)
            inputs.append(conn)
            data_queue[conn] = queue.Queue()  # 创建连接的消息队列
        else:
            data = r.recv(1024)  # r就是conn
            print(data.decode('utf-8'))
            # 接下来处理发数据
            # 把有数据返回的连接当道outputs列表里，这样下次循环select就会监视到
            # 要发的数据也得保存，这里用队列来保存准备发送的数据
            # 要为每个连接分别建一个队列，不能把消息搞混，这里用字典
            # 字典的key就是连接，value就是该连接的消息队列
            outputs.append(r)
            # data_queue = {} 在使用前，要先建立一个空字典。这句放到While循环外
            # data_queue[conn] = queue.Queue  # 在客户端建立连接时，就创建好连接的消息队列。这句放在上面处理server.accept()里面
            data_queue[r].put(data.upper())  # 然后就先不管了。等到select再监视的时候，会返回到writeable列表里。
            # 所以后面还要写一个writeable的for循环
    # 虽然是在上面的for循环里添加的，但是要等到在执行一次select后才会在writeable里有返回值
    for w in writeable:
        data = data_queue[w].get()  # 从队列里取出数据。这里get了之后，这条消息就从队列里移除了
        w.send(data)  # 发数据，注意data的数据类型
        outputs.remove(w)  # 从outputs里移除这个活动的连接，否则下次过来还有尝试在发数据，但是消息队列里是空的

这里把之前客户端注释掉的内容去掉测试一下收数据。并且已经可以接入多个客户端了，开2个试一下就好了。

处理客户端断开的问题

剩下就是客户端断开的问题了。断开有两种情况：
一种是正常断开，客户端close()，会发送一个空给服务的，那么要在 data = r.recv(1024) 之后判断一下是不是空，就和之前写的socket服务器一样。
还有一种是强行关闭客户端，这时inputs仍然会收到活跃连接，但是recv的时候会抛出异常“ConnectionResetError”，这里大概要把recv放到try里，如果捕获到异常，就断开客户端。
另外还有一个exceptional异常列表有返回的情况，这里也粗暴的断开客户端处理了好了。
客户端断开就是要清除掉字典和列表中的这个连接的信息。

import select
import socket
import queue

server = socket.socket()
server.bind(('localhost', 9000))
server.listen()
print("监听已经开启")
server.setblocking(False)  # 设置为非阻塞

inputs = [server, ]
outputs = []
data_queue = {}  # 存放返回给客户端消息的队列，每个客户端连接一个队列，就是一个item
while True:  # select返回活动链接==>for循环处理所有的活动链接，循环往复
    # 返回3个数据，监视到活动的3个列表（读列表，写列表，异常列表）
    readable, writeable, exceptional = select.select(inputs, outputs, inputs)
    # 这里就是阻塞的，直到select监视到列表中有活动的链接，才会继续
    print('select返回：\n', readable, writeable, exceptional)
    for r in readable:  # 可能一次返回多个连接啊，所以得写个循环
        # 现在inputs里有server 和 conn了，
        # 如果readable返回的是server的活动，表示来了一个新链接
        # 如果readable返回的是conn的活动，表示收到了conn发来的数据
        if r is server:
            conn, addr = r.accept()  # r就是server
            print('接收到新的客户端连接：\n', conn, addr)
            inputs.append(conn)  # 将新连接加入到select监视列表中
            data_queue[conn] = queue.Queue()  # 创建新连接的消息队列
        else:
            # 这里有3中情况，有数据，有空数据（正常断开），无数据（一旦recv就报错）
            try:
                data = r.recv(1024)
            except Exception as error:
                print("recv时捕获到异常：%s" % error)
                # 清除连接的4个操作，这段代码重复用了3次，应该专门写个函数引用
                # 1 从读列表中清除，这里其实不用判断，但是后面的for循环里可能会尝试重复remove
                # 2 如果还有没发出去的消息，把连接从写列表中清除
                # 3 关闭连接
                # 4 如果还有没发出去的消息，把消息队列的对象从字典里清除
                if r in inputs:
                    inputs.remove(r)
                if r in outputs:
                    outputs.remove(r)
                r.close()
                if r in data_queue:
                    del data_queue[r]
            else:
                # 这里处理能recv到数据，收到数据就加入outputs列表。为空就清除连接
                if data:
                    print(data.decode('utf-8'))
                    outputs.append(r)
                    data_queue[r].put(data.upper())
                else:
                    print("客户端已断开：\n", r)
                    inputs.remove(r)
                    if r in outputs:
                        outputs.remove(r)
                    if r in data_queue:
                        del data_queue[r]
    # 虽然是上面的for循环里添加的，但是要等到在执行一次select后才会在writeable里有返回值
    for w in writeable:
        data = data_queue[w].get()  # 从队列里取出数据。这里get了之后，这条消息就从队列里移除了
        w.send(data)  # 发数据，注意data的数据类型
        outputs.remove(w)  # 从outputs里移除这个活动的连接，否则下次过来还有尝试在发数据，但是消息队列里是空的
    # 还有一个exceptional没处理，还是和上面一样，再写一个for循环
    # 异常处理这里还是简单粗暴把异常列表中的连接清除就好了
    for e in exceptional:
        print("异常列表有返回：", e)
        if e in inputs:
            inputs.remove(e)
        if e in outputs:
            outputs.remove(e)
        e.close()
        if e in data_queue:
            del data_queue[e]

现在就可以处理高并发了，写一个多线程的客户端，每个线程循环发数据测试一下。
测试客户端：

import socket
import threading

HOST = 'localhost'
PORT = 9000
def client(i):
    client = socket.socket()
    client.connect((HOST, PORT))
    for j in range(100):
        msg = "hello %s %s" % (i, j)
        client.send(msg.encode('utf-8'))
        data = client.recv(1024)
        print('Received:', data.decode('utf-8'))
    client.close()

if __name__ == '__main__':
    for i in range(100):
        t = threading.Thread(target=client, args=(i,))
        t.start()

到这里感觉好了，但是还有3地方有问题，下面一个一个说明。上面的代码经不起强行断开客户端的考验。

send 和 recv 异常的问题

非阻塞模式下，如果调用recv()没有发现任何数据，或send()调用无法立即发送数据，那么将引起异常。我这里看到的是 "ConnectionResetError" 。
产生的原因是强行断开客户端，导致这个连接已经失效，但是连接还在select返回的列表里。这时之后的for循环里还会尝试去send或recv这个连接就会抛出异常。
解决办法 ：send 和 recv 的时候都得用try，然后捕获到异常后，就把这个连接清理掉

改进writeable循环

data = data_queue[w].get() 这里有时候会报错 "KeyError" ，就是字典里已经没有这个key了。那就是在另外2个for循环里已经将这个key清除了。这里清除的时候没清writeable列表，所以在清除连接的时候加一句：

if r in writeable:
    writeable.remove(r)

或者是用字典的get方法读取，这样在读取不存在的key的时候，会返回None：

data = data_queue.get(w)
if data:
    data = data.get()
else:
    if w in outputs:
        outputs.remove(w)
    continue

remove队列的问题

上面有3处地方会清除连接，这里可以另外写一个函数，需要清除的时候调用函数就好了。另外可能会再两个for循环里同时会要清除同一个连接，这样在第二次清除的时候如果不做if判断，就会报错无法删除列表里不存在的元素。这里保险起见还是在remove之前都加上if判断。应该只有第一个for循环里的inputs是一定有的不用判断，别的地方都可能会报错。

writeable的另外一个写法

说好了3个问题，这里还有一个暂时没事出问题，就是可能发数据的时候一次发不完。现在的写法都是默认一次send就是发完的，send之后直接从outputs里remove掉。
也可以换一种方法，send之后，不从移出outputs列表。那么下次while循环还会进来，此时get队列的时候要用get_nowait无阻塞模式取队列，如果空会抛出队列空的异常，那么这个一定是发完了。此时再把链接移出outputs列表。如果需要可以把上面字典 "KeyError" 一起处理了，继续用 data = data_queue[w].get() 取列表，放到try里，捕获2种异常分别处理掉。
后面的最终版本没这么用，我怕再踩到坑，先把想法记着。

小结 - 最终版本

下面给出一个所有遇到的问题都解决了的最终版本：

import select
import socket
import queue

server = socket.socket()
server.bind(('localhost', 9000))
server.listen()
print("监听已经开启")
server.setblocking(False)  # 设置为非阻塞

inputs = [server, ]
outputs = []
data_queue = {}  # 存放返回给客户端消息的队列，每个客户端连接一个队列，就是一个item
while True:  # select返回活动链接==>for循环处理所有的活动链接，循环往复
    # 返回3个数据，监视到活动的3个列表（读列表，写列表，异常列表）
    readable, writeable, exceptional = select.select(inputs, outputs, inputs)
    # 这里就是阻塞的，直到select监视到列表中有活动的链接，才会继续
    print('select返回：\n', readable, writeable, exceptional)
    for r in readable:  # 可能一次返回多个连接啊，所以得写个循环
        # 现在inputs里有server 和 conn了，
        # 如果readable返回的是server的活动，表示来了一个新链接
        # 如果readable返回的是conn的活动，表示收到了conn发来的数据
        if r is server:
            conn, addr = r.accept()  # r就是server
            print('接收到新的客户端连接：\n', conn, addr)
            conn.setblocking(False)
            inputs.append(conn)  # 将新连接加入到select监视列表中
            data_queue[conn] = queue.Queue()  # 创建新连接的消息队列
        else:
            # 这里有3中情况，有数据，有空数据（正常断开），无数据（一旦recv就报错）
            try:
                data = r.recv(1024)
            except Exception as error:
                print("recv时捕获到异常：%s" % error)
                # 清除连接的4个操作，这段代码重复用了3次，应该专门写个函数引用
                # 1 从读列表中清除，这里其实不用判断，但是后面的for循环里可能会尝试重复remove
                # 2 如果还有没发出去的消息，把连接从写列表中清除
                # 3 关闭连接
                # 4 如果还有没发出去的消息，把消息队列的对象从字典里清除
                if r in inputs:
                    inputs.remove(r)
                if r in outputs:
                    outputs.remove(r)
                r.close()
                if r in data_queue:
                    del data_queue[r]
            else:
                # 这里处理能recv到数据，收到数据就加入outputs列表。为空就清除连接
                if data:
                    print(data.decode('utf-8'))
                    if r not in outputs: outputs.append(r)
                    data_queue[r].put(data.upper())
                else:
                    print("客户端已断开：\n", r
                    if r in inputs:
                        inputs.remove(r)
                    if r in outputs:
                        outputs.remove(r)
                    r.close()
                    if r in data_queue:
                        del data_queue[r]
    # 虽然是上面的for循环里添加的，但是要等到在执行一次select后才会在writeable里有返回值
    for w in writeable:
        data = data_queue.get(w)  # 先从字典里取出这个队列
        if data:  # 队列存在，取数据
            data = data.get()  # 从队列里取出数据。这里get了之后，这条消息就从队列里移除了
        else:  # 队列不存在，这个连接已经被清除了，remove掉，下一个循环
            if w in outputs:
                outputs.remove(w)
            continue
        try:
            w.send(data)  # 发数据，注意data的数据类型
        except Exception as error:
            print("send时捕获到异常：%s" % error)
            if r in inputs:
                inputs.remove(r)
            if r in outputs:
                outputs.remove(r)
            r.close()
            if r in data_queue:
                del data_queue[r]
        else:
            if w in outputs:
                outputs.remove(w)  # 有可能一次send不完，这里也可以不remove，用另外一个方法
            # 开头用get_nowait无阻塞模式取队列，捕获到队列为空的异常，再remove掉这个连接
    # 还有一个exceptional没处理，还是和上面一样，再写一个for循环
    # 异常处理这里还是简单粗暴把异常列表中的连接清除就好了
    for e in exceptional:
        print("异常列表有返回：", e)
        if e in inputs:
            inputs.remove(e)
        if e in outputs:
            outputs.remove(e)
        e.close()
        if e in data_queue:
            del data_queue[e]

学了那么多都没用，应该也用不到，而且可能还会有别的问题。最可怕的是别人都不这么用，有问题都找不到人解决。主要是通过select的使用来了解IO多路复用。除了select，还有poll和epoll。
epoll 更高效，但是代码也更复杂。不过这些我们都用不到，都太底层了。平时用用已经封装好的模块就好了。这里通过学习大概了解一下底层是怎么实现的。剩下的会用模块就好了，现在至少知道模块中是怎么运作的了。
最后就是学一下下面的已经封装的selectors模块，底层都清楚了，学习使用下面的模块已经没有难度了。

平时直接用已经封装好的模块，简单坑又少。selectors模块，默认会用epoll，如果系统不支持，就用select，完美。
理解了前面select的机制，在使用这个模块就简单了。步骤都一样。步骤都一样，但是都封装好了。

import selectors
import socket

def accept(sock, mask):
    """建立新连接并注册"""
    conn, addr = sock.accept()
    print('接收到新的客户端连接：\n', conn, addr)
    conn.setblocking(False)  # sock里已经设置为False，这里貌似没意义，反正没差
    # 上面已经建立好连接了，把新连接注册到sel里，这是第二次注册了
    # 第一次注册是注册server接受客户端连接请求的连接
    # 这里是连接建立后收发数据的连接，这个连接如果发现是活动的，调用的就是read方法了
    sel.reGISter(conn, selectors.EVENT_READ, read)

def read(conn, mask):
    """接收数据"""
    data = conn.recv(1024)
    if data:
        print(data.decode('utf-8'))
        conn.send(data.upper())
    else:
        print("客户端已断开：\n", conn)
        sel.unregister(conn)  # 注销这个连接
        conn.close()

def server(port):
    sock = socket.socket()
    sock.bind(('localhost', port))
    sock.listen()
    print("监听已经开启")
    sock.setblocking(False)
    # 下面就是select方法，也可能是epoll
    # 注册你的socket，就是让select监视，监视到有活动的连接，就调用accept函数
    sel.register(sock, selectors.EVENT_READ, accept)
    # 上面还没有开始监视，只是先把select准备好
    while True:
        events = sel.select()  # 这里就还是监视了
        # 这里会阻塞，一旦有活动的连接，就会返回给events列表
        for key, mask in events:
            callback = key.data  # key.data就是sel.register里的accpet这个函数
            # 现在callback就是accpet这个函数了，下面加上括号填上参数就执行了
            callback(key.fileobj, mask)  # key.fileobj就是sel.register里的sock

if __name__ == '__main__':
    sel = selectors.DefaultSelector()  # 老套路，用之前先实例化一个对象
    server(10001)

还是有那个老问题，客户端强制断开，服务端会报错 "ConnectionResetError" 。

SELECT版FTP ：

  • SELECT或SELECTORS模块实现并发简单版FTP
  • 允许多用户并发上传下载文件

相关文章