Python全栈开发-Day10-进程/协程/异步IO/IO多路复用

本节内容

  1. 多进程multiprocessing
  2. 进程间的通讯
  3. 协程
  4. 论事件驱动与异步IO
  5. SelectPollEpoll——IO多路复用

 

1、多进程multiprocessing

Python的线程用的是操作系统的原生线程,同样python的进程用的是操作系统的原生进程。

多进程之间没有锁的概念,多进程之间数据不能互相访问,所以不存在互斥锁。GIL问题又是仅仅出现在多线程中。

所以如果我们启动8个进程,每个进程有一个主线程,即8个线程,分别运行在8个CPU上,就可以充分利用多核的优势了。

在多进程充分利用多核的优势下,唯一的坏处是这8个进程之间数据无法共享。传递数据需要找媒介。

8进程表示,同一时间最多只能干8件事情。

所以多进程可以解决多核的问题

同时每个进程里又可以写多个线程,启动并执行。

每一个进程都是由它的父进程启动的。

os.getppid()  #获得父进程的id

os.getpid()  #获得自己进程的id

1
2
3
4
5
6
7
8
9
10
from multiprocessing import Process
import time
def f(name):
    time.sleep(2)
    print('hello', name)
 
if __name__ == '__main__':
    = Process(target=f, args=('bob',))
    p.start()
    p.join()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
from multiprocessing import Process
import os
 
def info(title):
    print(title)
    print('module name:', __name__)
    print('parent process:', os.getppid())
    print('process id:', os.getpid())
    print(" ")
 
def f(name):
    info('33[31;1mfunction f33[0m')
    print('hello', name)
 
if __name__ == '__main__':
    info('33[32;1mmain process line33[0m')
    = Process(target=f, args=('bob',))
    p.start()
    p.join()

2、进程间的通讯  

不同进程间内存是不共享的,要想实现两个进程间的数据交换,可以用以下方法:

Queues(进程队列)——数据传递

使用方法跟threading里的queue差不多,

原理上是两个进程各有一个queue队列,通过pickle序列化的方式,实现两个队列之间的交互,从而看起来像两个进程之间的交互。

所以实际上进程队列是2个Queues,而不是一个共享queue队列。

1
2
3
4
5
6
7
8
9
10
11
12
from multiprocessing import Process, Queue
 
def f(q):
    q.put([42None'hello'])
 
if __name__ == '__main__':
    = Queue()
    = Process(target=f, args=(q,))
    p.start()
    print(q.get())    # prints "[42, None, 'hello']"
    p.join()

Pipes(管道)——数据传递

pipe管道就相当于拿了一根电话线,两头分别连接了两个进程。原理是利用socket网络协议进行周转。

pipe实例一生成会产生两个返回对象,一个是管道的一头,另一个是管道的另一头。

1
2
3
4
5
6
7
8
9
10
11
12
from multiprocessing import Process, Pipe
 
def f(conn):
    conn.send([42None'hello'])
    conn.close()
 
if __name__ == '__main__':
    parent_conn, child_conn = Pipe()
    = Process(target=f, args=(child_conn,))
    p.start()
    print(parent_conn.recv())   # prints "[42, None, 'hello']"
    p.join()
 
Managers——数据共享

Managers已经自动加锁,所以不需要手动加锁。

A manager object returned by Manager() controls a server process which holds Python objects and allows other processes to manipulate them using proxies.

A manager returned by Manager() will support types listdictNamespaceLockRLockSemaphoreBoundedSemaphoreConditionEventBarrierQueueValue and Array. For example,

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
from multiprocessing import Process, Manager
 
def f(d, l):
    d[1= '1'
    d['2'= 2
    d[0.25= None
    l.append(1)
    print(l)
 
if __name__ == '__main__':
    with Manager() as manager:
        = manager.dict()
 
        = manager.list(range(5))
        p_list = []
        for in range(10):
            = Process(target=f, args=(d, l))
            p.start()
            p_list.append(p)
        for res in p_list:
            res.join()
 
        print(d)
        print(l)

  

进程锁

虽然本身由于进程间不能共享数据,所以不需要锁。但是当进程往屏幕上打印数据时,对各个进程而言屏幕是共享的。所以这个锁的目的是锁住一个时间只能一个进程去打印数据。保证打印数据的完整性,前面正在打印的数据,不被后面要打印的数据打断。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
from multiprocessing import Process, Lock
 
def f(l, i):
    l.acquire()
    try:
        print('hello world', i)
    finally:
        l.release()
 
if __name__ == '__main__':
    lock = Lock()
 
    for num in range(10):
        Process(target=f, args=(lock, num)).start()

  

进程池  

进程池内部维护一个进程序列,当使用时,则去进程池中获取一个进程,如果进程池序列中没有可供使用的进程,那么程序就会等待,直到进程池中有可用进程为止。

进程池中有两个方法:

  • apply  #进程池串行
  • apply_async  #进程池并行

对于进程池中进程的启动是要使用pool.apply()或pool.apply_async()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
from  multiprocessing import Process,Pool,freeze_support
import time
 
def Foo(i):
    time.sleep(2)
    return i+100
 
def Bar(arg):
    print('-->exec done:',arg)
if __name__ == '__main__':
  freeze_support()  #在windows上必须添加这行和上行代码
 
  pool = Pool(5)  #允许进程池里同时放入5个进程
 
  for in range(10):
    pool.apply_async(func=Foo, args=(i,),callback=Bar)
    #pool.apply_async(func=Foo, args=(i,))
    #pool.apply(func=Foo, args=(i,))
 
  print('end')
  pool.close()
  pool.join()#进程池中进程执行完毕后再关闭,如果注释,那么程序直接关闭。

  这里不知何种原因,必须是先pool.close(),然后再是pool.join()

  如果最后不加pool.join(),主程序不会等待进程池执行完毕,会直接关闭。

if __name__ == '__main__':  #这句话是为了区分,主动启动脚本还是把它当成一个模块,从别的地方去调用。

如果主动启动该脚本,则该段代码下面部分代码会被执行。如果从别的地方调用,则下面代码不会被执行。

主动执行该py文件时,把这句话当成主程序的入口。

上面代码中含有pool.apply_async(func=Foo, args=(i,),callback=Bar),callback叫做回调,意思是执行完这个语句后再回调Bar函数。这里需要注意,callback方法的调用进程是主进程,而不是子进程。

回调函数的意义,如果开了100个子进程,在每个子进程结束时往数据库中插入数据,需要建立100个连接,但如果先把100个子进程的结果保存到变量里,让父进程连接数据库一次,然后一口气全部插入,就大大提高了程序的运行速度。此时的回调函数,应运而生。

3、协程

协程,又称微线程。英文名Coroutine。一句话说明什么是线程:协程是一种用户态的轻量级线程

CPU只认识线程,并不知道协程的存在。协程是跑在线程中的。

协程拥有自己的寄存器上下文和栈。协程调度切换时,将寄存器上下文和栈保存到其他地方,在切回来的时候,恢复先前保存的寄存器上下文和栈。因此:

协程能保留上一次调用时的状态(即所有局部状态的一个特定组合),每次过程重入时,就相当于进入上一次调用的状态,换种说法:进入上一次离开时所处逻辑流的位置。

换句话说:在单线程下,实现并发的效果,就是协程。例如,之前使用yield做的生产者消费者模型。

协程的好处:

  • 无需线程上下文切换的开销
  • 无需原子操作锁定及同步的开销
    • "原子操作(atomic operation)是不需要synchronized",所谓原子操作是指不会被线程调度机制打断的操作;这种操作一旦开始,就一直运行到结束,中间不会有任何 context switch (切换到另一个线程)。原子操作可以是一个步骤,也可以是多个操作步骤,但是其顺序是不可以被打乱,或者切割掉只执行部分。视作整体是原子性的核心。
  • 方便切换控制流,简化编程模型
  • 高并发+高扩展性+低成本:一个CPU支持上万的协程都不是问题。所以很适合用于高并发处理。

缺点:

  • 无法利用多核资源:协程的本质是个单线程,它不能同时将单个CPU 的多个核用上,协程需要和进程配合才能运行在多CPU上.当然我们日常所编写的绝大部分应用都没有这个必要,除非是cpu密集型应用。
  • 进行阻塞(Blocking)操作(如IO时)会阻塞掉整个程序

使用yield实现协程操作的例子    

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
import time
import queue
def consumer(name):
    print("--->starting eating baozi...")
    while True:
        new_baozi = yield  #yield这里本身可以返回数据,也可接受数据
        print("[%s] is eating baozi %s" % (name,new_baozi))
        #time.sleep(1)
 
def producer():
 
    = con.__next__()
    = con2.__next__()
    = 0
    while n < 5:
        +=1
        con.send(n)  #激活yield的阻塞状态,同时传入一个数据
        con2.send(n)
        print("33[32;1m[producer]33[0m is making baozi %s" %n )
 
 
if __name__ == '__main__':
    con = consumer("c1")
    con2 = consumer("c2")
    = producer()

协程之所以能处理大并发,就是把I/O操作给挤掉了,即一旦遇到I/O操作就切换。使得整个程序变成了只有CPU运算,大大提高了效率。

只要I/O操作一完成,CPU就可以切换回去了。这样就把I/O操作完全挤出去了。程序会自动检测I/O是否完成,不需要我们关心。

协程的标准定义,即符合什么条件就能称之为协程:

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

Greenlet

yield是自己写的协程,greenlet是一个封装好的协程。

greenlet是一个用C实现的协程模块,相比与python自带的yield,它可以使你在任意函数之间随意切换,而不需把这个函数先声明为generator

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
# -*- coding:utf-8 -*-
 
 
from greenlet import greenlet
 
 
def test1():
    print(12)
    gr2.switch()  #gr2.switch()的意思是切换到gr2
    print(34)
    gr2.switch()  #同上
 
 
def test2():
    print(56)
    gr1.switch()
    print(78)
 
 
gr1 = greenlet(test1)  #启动一个协程
gr2 = greenlet(test2)
gr1.switch()  #gr1.switch()的意思是切换到gr1

感觉确实用着比generator还简单了呢,但好像还没有解决一个问题,就是遇到IO操作,自动切换,对不对?

 greenlet只能手动切换,就相当于汽车的手动档。

Gevent 

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

【注意】上文说greenlet手动切换,相当于手动档汽车,而gevent是自动切换,相当于自动档汽车。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
import gevent
 
def func1():
    print('33[31;1m李闯在跟海涛搞...33[0m')
    gevent.sleep(2)
    print('33[31;1m李闯又回去跟继续跟海涛搞...33[0m')
 
def func2():
    print('33[32;1m李闯切换到了跟海龙搞...33[0m')
    gevent.sleep(1)
    print('33[32;1m李闯搞完了海涛,回来继续跟海龙搞...33[0m')
 
 
gevent.joinall([
    gevent.spawn(func1),  #生成
    gevent.spawn(func2),
    #gevent.spawn(func3),
])

通过gevent实现单线程下的多socket并发

server side 

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
import sys
import socket
import time
import gevent
 
from gevent import socket,monkey
monkey.patch_all()
 
 
def server(port):
    = socket.socket()
    s.bind(('0.0.0.0', port))
    s.listen(500)
    while True:
        cli, addr = s.accept()
        gevent.spawn(handle_request, cli)
 
 
 
def handle_request(conn):
    try:
        while True:
            data = conn.recv(1024)
            print("recv:", data)
            conn.send(data)
            if not data:
                conn.shutdown(socket.SHUT_WR)
 
    except Exception as  ex:
        print(ex)
    finally:
        conn.close()
if __name__ == '__main__':
    server(8001)

  

client side   

1
2
3
4
5
6
7
8
9
10
11
12
13
14
import socket
 
HOST = 'localhost'    # The remote host
PORT = 8001           # The same port as used by the server
= socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.connect((HOST, PORT))
while True:
    msg = bytes(input(">>:"),encoding="utf8")
    s.sendall(msg)
    data = s.recv(1024)
    #print(data)
 
    print('Received'repr(data))
s.close()

上述代码是一个非常牛逼的socket服务端和客户端,使用异步IO得以实现。效率高。

4、论事件驱动与异步IO

通常,我们写服务器处理模型的程序时,有以下几种模型:
(1)每收到一个请求,创建一个新的进程,来处理该请求;
(2)每收到一个请求,创建一个新的线程,来处理该请求;
(3)每收到一个请求,放入一个事件列表,让主进程通过非阻塞I/O方式来处理请求
上面的几种方式,各有千秋,
第(1)中方法,由于创建新的进程的开销比较大,所以,会导致服务器性能比较差,但实现比较简单。
第(2)种方式,由于要涉及到线程的同步,有可能会面临死锁等问题。
第(3)种方式,在写应用程序代码时,逻辑比前面两种都复杂。
综合考虑各方面因素,一般普遍认为第(3)种方式是大多数网络服务器采用的方式
 

看图说话讲事件驱动模型

在UI编程中,常常要对鼠标点击进行相应的操作,首先如何获得鼠标点击呢?
方式一:创建一个线程,该线程一直循环检测是否有鼠标点击,那么这个方式有以下几个缺点
1. CPU资源浪费,可能鼠标点击的频率非常小,但是扫描线程还是会一直循环检测,这会造成很多的CPU资源浪费;如果扫描鼠标点击的接口是阻塞的呢?
2. 如果是堵塞的,又会出现下面这样的问题,如果我们不但要扫描鼠标点击,还要扫描键盘是否按下,由于扫描鼠标时被堵塞了,那么可能永远不会去扫描键盘;
3. 如果一个循环需要扫描的设备非常多,这又会引来响应时间的问题;
所以,该方式是非常不好的。

方式二:就是事件驱动模型
目前大部分的UI编程都是事件驱动模型,如很多UI平台都会提供onClick()事件,这个事件就代表鼠标按下事件。事件驱动模型大体思路如下:
1. 有一个事件(消息)队列;
2. 鼠标按下时,往这个队列中增加一个点击事件(消息);
3. 有个循环,不断从队列取出事件,根据不同的事件,调用不同的函数,如onClick()、onKeyDown()等;
4. 事件(消息)一般都各自保存各自的处理函数指针,这样,每个消息都有独立的处理函数;

 上文中提出的问题:在异步I/O操作中,当CPU遇见I/O操作时,就会让操作系统通过自己的文件接口进行I/O操作,而CPU自己则切换到其他的程序处执行。在切换之前,CPU会注册一个回调函数。作用是,当操作系统完成I/O操作后,调用回调函数,来主动的告诉CPU我完成了,你可以切换回来了。这个过程就是事件驱动。

 

事件驱动编程是一种编程范式,这里程序的执行流由外部事件来决定。它的特点是包含一个事件循环,当外部事件发生时使用回调机制来触发相应的处理。另外两种常见的编程范式是单线程以及多线程编程。

让我们用例子来比较和对比一下单线程、多线程以及事件驱动编程模型。下图展示了随着时间的推移,这三种模式下程序所做的工作。这个程序有3个任务需要完成,每个任务都在等待I/O操作时阻塞自身。阻塞在I/O操作上所花费的时间已经用灰色框标示出来了。

 

在单线程同步模型中,任务按照顺序执行。如果某个任务因为I/O而阻塞,其他所有的任务都必须等待,直到它完成之后它们才能依次执行。这种明确的执行顺序和串行化处理的行为是很容易推断得出的。如果任务之间并没有互相依赖的关系,但仍然需要互相等待的话这就使得程序不必要的降低了运行速度。

在多线程版本中,这3个任务分别在独立的线程中执行。这些线程由操作系统来管理,在多处理器系统上可以并行处理,或者在单处理器系统上交错执行。这使得当某个线程阻塞在某个资源的同时其他线程得以继续执行。与完成类似功能的同步程序相比,这种方式更有效率,但程序员必须写代码来保护共享资源,防止其被多个线程同时访问。多线程程序更加难以推断,因为这类程序不得不通过线程同步机制如锁、可重入函数、线程局部存储或者其他机制来处理线程安全问题,如果实现不当就会导致出现微妙且令人痛不欲生的bug。

在事件驱动版本的程序中,3个任务交错执行,但仍然在一个单独的线程控制中。当处理I/O或者其他昂贵的操作时,注册一个回调到事件循环中,然后当I/O操作完成时继续执行。回调描述了该如何处理某个事件。事件循环轮询所有的事件,当事件到来时将它们分配给等待处理事件的回调函数。这种方式让程序尽可能的得以执行而不需要用到额外的线程。事件驱动型程序比多线程程序更容易推断出行为,因为程序员不需要关心线程安全问题。

当我们面对如下的环境时,事件驱动模型通常是一个好的选择:

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

当应用程序需要在任务间共享可变的数据时,这也是一个不错的选择,因为这里不需要采用同步处理。

网络应用程序通常都有上述这些特点,这使得它们能够很好的契合事件驱动编程模型。

此处要提出一个问题,就是,上面的事件驱动模型中,只要一遇到IO就注册一个事件,然后主程序就可以继续干其它的事情了,只到io处理完毕后,继续恢复之前中断的任务,这本质上是怎么实现的呢?下面我们就来一起揭开这神秘的面纱。。。。

5、SelectPollEpoll——IO多路复用 

Select Poll Epoll 属于IO多路复用

虽然IO多路复用的效果不如异步IO好,但是由于异步IO实现起来较复杂,所以一般情况下还是IO多路复用用的多一些。同时在内核层面,对异步IO支持也不是特别好。

所以我们一般在市面上见到的所谓的异步IO,比如Nginx等其实本质上是IO多路复用。

http://www.cnblogs.com/alex3714/p/4372426.html 

番外篇 http://www.cnblogs.com/alex3714/articles/5876749.html 

由于在实际中,IO多路复用一般都是做服务器端的Web开发,或者是异步爬虫等,这种十分复杂的程序开发的。与我自己学Python做量化交易的初衷,相差太远,这里就不写相关的知识了。

selectors模块

selectors模块是对SelectPollEpoll的高级封装

如果不指定,selectors默认使用最高级的Epoll,但由于windows系统不支持epoll,所以它就会用select

This module allows high-level and efficient I/O multiplexing, built upon the select module primitives. Users are encouraged to use this module instead, unless they want precise control over the OS-level primitives used.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
import selectors
import socket
 
sel = selectors.DefaultSelector()
 
def accept(sock, mask):
    conn, addr = sock.accept()  # Should be ready
    print('accepted', conn, 'from', addr)
    conn.setblocking(False)
    sel.register(conn, selectors.EVENT_READ, read)
 
def read(conn, mask):
    data = conn.recv(1000)  # Should be ready
    if data:
        print('echoing'repr(data), 'to', conn)
        conn.send(data)  # Hope it won't block
    else:
        print('closing', conn)
        sel.unregister(conn)
        conn.close()
 
sock = socket.socket()
sock.bind(('localhost'10000))
sock.listen(100)
sock.setblocking(False)
sel.register(sock, selectors.EVENT_READ, accept)
 
while True:
    events = sel.select()
    for key, mask in events:
        callback = key.data
        callback(key.fileobj, mask)

  

原文地址:https://www.cnblogs.com/GavinSimons/p/7991040.html