Python 7 并发编程

刺骨的言语ヽ痛彻心扉 2021-12-22 15:39 219阅读 0赞

**第一部分：进程**

进程就是在运行着的程序。

相关概念：

1、同步：指同一时间只能做一件事，即使需要等待，也一直等到第一件事结束，才开始第二件事。

2、异步：指同一时间可以做多件事情。

3、阻塞：指需要等待时等待

4、非阻塞：指需要等待时不等待

5、同步阻塞形式：遇到阻塞只能等待

6、异步阻塞形式：遇到阻塞可以做别的事，但是要定时查看阻塞是否结束

7、同步非阻塞：遇到阻塞需要等待，但是可以做别的事，只是两件事需要来回切换

8、异步非阻塞：遇到阻塞直接去做别的事，当阻塞结束被通知，立即回来做这件事。

**一、多进程基础**

**1、多进程创建**

在多进程中，如果主进程先执行完，会等待子进程执行完毕再关闭，如果子进程先执行完，那么主进程结束后立即关闭。

创建多进程的基本方法：

from multiprocessing import Process
    
    def func(arg):
        print(arg)
    
    if __name__ == "__main__":
        for i in range(10):
            p = Process(target=func, args=(i,))    #这里是创建了一个p进程，但是还没有执行
            p.start()    #只有start后才执行，相当于是开启了一个新的子进程，func里面的代码执行和当前进程无关
        print(执行完毕)    #这句话会在一开始打印，因为子进程的代码和主进程的代码是异步的，同时执行，由于子进程会有等待的时间，所以会先打印出这句话

第二种方式：

from multiprocessing import Process
    
    class MyProcess(Process):
        def run(self):
            print(111)
    
    if __name__ == "__main__":
        for i in range(10):
            p = MyProcess()
            p.start()        #p.start自动执行run()方法中的代码
        print("执行完毕")

**2、join方法**

join方法会将这个对象和他之后的代码变成同步执行，如果想在所有子进程结束后再打印执行完毕，可以这样：

from multiprocessing import Process
    
    def func(arg):
        print(arg)
    
    if __name__ == "__main__":
        li = []
        for i in range(10):
            p = Process(target=func, args=(i,))
            p.start()
            li.append(p)
        [p.join() for p in li]
        print("执行完毕")

**3、守护进程**

守护进程也是子进程，但是守护进程会在主进程的代码执行结束时自行结束。

p = Process(xxxx)
    p.daemon = True
    p.start()

**二、进程同步控制**

**1、进程锁**

锁用来保护数据安全，当一段代码加了锁时，每次只能有一个进程执行这段代码，当这个进程执行完这段代码后，其他进程才能执行。

from multiprocessing import Lock
    lock = Lock()
    lock.acquire()  #获取钥匙
    xxxxxxxxx
    lock.release()  #还钥匙

**2、信号量**

信号量其实就是锁的升级，每次可以有多个进程执行某段代码

![ContractedBlock.gif][] ![ExpandedBlockStart.gif][]

import time
    from multiprocessing import Process
    from multiprocessing import Semaphore
    
    def mod(i, sem):
        sem.acquire()
        print("{}过马路".format(i))
        time.sleep(2)
        sem.release()
    
    if __name__ == "__main__":
        sem = Semaphore(4)
        for i in range(20):
            p = Process(target=mod, args=(i, sem))
            p.start()

**3、事件**

事件会一直处于监听状态，当e.is\_set()为False时，e.wait()会阻塞，反之不阻塞

![ContractedBlock.gif][] ![ExpandedBlockStart.gif][]

import time
    from multiprocessing import Process, Event
    
    def light(e):
        while True:
            if e.is_set():    #结果为True不阻塞
                print("红灯亮了")
                e.clear()  #将is_set()设置为False
            else:    #结果为False阻塞
                print("绿灯亮了")
                e.set()    #将is_set()设置成True
            time.sleep(2)
    
    def cars(i, e):
        if not e.is_set():
            print("{}正在等待".format(i))
            e.wait()
        print("{}通过了".format(i))
    
    if __name__ == "__main__":
        e = Event()
        l = Process(target=light, args=(e,))
        l.start()
        time.sleep(0.1)
        for i in range(20):
            car = Process(target=cars, args=(i, e))
            car.start()
            time.sleep(1)

**三、进程通信**

**1、队列**

队列里的数据都是安全的，同一时刻只能有一个进程操作某个数据，不存在多个进程操作同一个数据的情况，相当于时每次操作都加了锁。

生产者消费者模型用来解决数据供需不平衡问题。

![ContractedBlock.gif][] ![ExpandedBlockStart.gif][]

import time
    from multiprocessing import Process, Queue
    
    def consumers(name, q):
        while True:
            f = q.get()
            if f == None:
                break
            print("{}吃了{}".format(name, f))
            time.sleep(1)
    
    def producer(name, food, q):
        for i in range(1,5):
            time.sleep(1)
            f = "{}{}".format(food, i)
            print("{}制作了{}".format(name, f))
            q.put(f)
    
    if __name__ == "__main__":
        q = Queue(10)
        p = Process(target=producer, args=("师傅", "面包", q))
        p.start()
        c = Process(target=consumers, args=("小明", q))
        c.start()
        p.join()
        q.put(None)

模型一Queue

![ContractedBlock.gif][] ![ExpandedBlockStart.gif][]

import time
    from multiprocessing import Process, JoinableQueue
    
    def consumers(name, q):
        while True:
            f = q.get()
            print("{}吃了{}".format(name, f))
            time.sleep(1)
            q.task_done()   #告诉队列，已取走一个数据
    
    def producer(name, food, q):
        for i in range(1,5):
            time.sleep(1)
            f = "{}{}".format(food, i)
            print("{}制作了{}".format(name, f))
            q.put(f)
        q.join()    #队列不再放入数据，如果所有数据被取完则结束
    
    if __name__ == "__main__":
        q = JoinableQueue(10)
        p = Process(target=producer, args=("师傅", "面包", q))
        p.start()
        c = Process(target=consumers, args=("小明", q))
        c.daemon = True
        c.start()
        p.join()

模型二JoinableQueue

**2、管道**

管道中的数据是不安全的，可能会有一个数据同时被两个进程取到。解决办法是加锁。

队列实际上就是管道+锁实现的。

![ContractedBlock.gif][] ![ExpandedBlockStart.gif][]

from multiprocessing import Process, Pipe
    
    def func(conn):
        print(conn.recv())
    
    if __name__ == "__main__":
        conn1, conn2 = Pipe()  #返回一个元祖
        conn1.send("hello")
        Process(target=func, args=(conn2,)).start()

基本使用

![ContractedBlock.gif][] ![ExpandedBlockStart.gif][]

import time
    import random
    from multiprocessing import Process, Pipe
    
    def consumers(conn1,conn2,name):
        conn1.close()
        while True:
            try:
                time.sleep(random.randint(1,3))
                food = conn2.recv()
                print("{}吃了{}".format(name,food))
            except EOFError:
                conn2.close()
                break
    
    def productor(conn1,conn2,name,food):
        conn2.close()
        for i in range(5):
            time.sleep(random.randint(1,3))
            f = "{}{}".format(food,i)
            print("{}制作了{}".format(name,f))
            conn1.send(f)
        conn1.close()
    
    if __name__ == "__main__":
        conn1, conn2 = Pipe()  #返回一个元祖
        Process(target=productor, args=(conn1,conn2,"王师傅","面包")).start()
        Process(target=consumers, args=(conn1,conn2,"小明")).start()
        conn1.close()
        conn2.close()

生产者消费者模型

如果在主进程和其他子进程中的所有管道均关闭，那么仅剩的那个管道如果取不到值，则会报EOFError。

**四、数据共享**

数据共享的数据也不安全，需要通过加锁的方式解决。

![ContractedBlock.gif][] ![ExpandedBlockStart.gif][]

from multiprocessing import Process, Manager
    
    def func(dic):
        dic["k1"] -= 1
    
    if __name__ == "__main__":
        m = Manager()
        dic = m.dict({
         "k1":100})
        p = Process(target=func, args=(dic,))
        p.start()
        p.join()
        print(dic["k1"])

**五、进程池**

由于开启进程需要消耗很多时间，如果开启较多的进程，则效率低下。所以创建进程池，规定数量的进程，有任务进来就去进程池中取一个进程，超出数量的进程会阻塞等待。

一般进程池中进程的数量是CPU个数+1

**1、map，默认异步**

自带close和join，返回值是一个列表，包含所有结果

![ContractedBlock.gif][] ![ExpandedBlockStart.gif][]

from multiprocessing import Pool
    
    def func(n):
        print(n)
    
    if __name__ == "__main__":
        p = Pool(5)
        p.map(func,range(10))

**2、apply，默认同步**

返回值是func函数的return值

![ContractedBlock.gif][] ![ExpandedBlockStart.gif][]

import time
    from multiprocessing import Pool
    
    def func(n):
        time.sleep(1)
        print(n)
    
    if __name__ == "__main__":
        p = Pool(5)
        for i in range(10):
            p.apply(func, args=(i,))

**3、apply\_async，默认异步**

返回值是一个对象，对象.get()取到值，get会阻塞直到取到值。所以当使用get时，程序变成同步状态。

![ContractedBlock.gif][] ![ExpandedBlockStart.gif][]

import time
    from multiprocessing import Pool
    
    def func(n):
        time.sleep(1)
        print(n)
    
    if __name__ == "__main__":
        p = Pool(5)
        for i in range(10):
            pro = p.apply_async(func, args=(i,))
        p.close()   #结束进程池接收任务
        p.join()    #感知进程池中的任务执行结束

**关于回调函数：**

回调函数是在主进程执行的。

使用场景：当任务数量比较多时，使用回调函数可以减少子进程执行的时间，以便让后面的任务快点被执行。

**第二部分：线程**

**为什么会有线程？**

1.进程只能同时做一件事。

2.进程遇到阻塞只能等待或者再开启一个进程。

3.进程间的数据不共享。

**线程的特点：**

1.轻量级，线程的开启和切换消耗的资源少。

2.线程间共享进程的数据

3.是cpu调度的最小单元

4.可并发运行

**GIL全局解释器锁**

在python的一个进程中，任何时刻都只能有一个线程在调用cpu，相当于每个进程中都有一把锁，每次只能有一个线程拿到钥匙，不能真正实现并行，不管电脑有多火cpu，一个进程中都只能有一个线程在同一时刻被执行。加上了这把锁，导致了python在进行多线程任务时比单线程慢，多花费了时间在线程的切换上。

产生原因：1.python诞生时还没有多核的概念。2.python是一门解释型语言，解释型语言是一句一句执行的，python并不知道下一句会开启一个线程，不能帮助我们规避数据不安全问题，而多线程中全局变量是共享的，有可能造成数据不安全，所以加了一把大锁，防止数据不安全。

注意：GIL并不是python的特性，而是CPython解释器提供的。

在实际的运行中，一个线程执行制定数量字节码，或者不间断运行15毫秒，或者遇到阻塞，python会释放GIL锁，各个线程竞争，抢到的线程再执行100字节码，以此类推。

解决方式：用多进程代替多线程，采用其他解释器。

**1.守护线程**

守护线程的使用方式和守护进程一样，设置daemon=True，但是守护进程在主进程代码执行完毕后立即结束，守护线程则是在主线程代码执行完毕后等待其他非守护线程也运行完毕才结束。

守护进程在执行完毕后不立即结束是在等待子进程结束后回收子进程的资源，而主线程在执行完毕后不立即结束是因为主线程的结束代表着进程的结束，进程结束则子线程不能共享进程中的资源。

**2.互斥锁和递归锁**

虽然存在全局解释器锁，但是线程间仍然会造成数据不安全现象，所以还是需要用到锁。

互斥锁是Lock，相当于是一把把独立的钥匙，可能存在线程1拿了锁A，线程2拿了锁B，过了一会线程1要拿锁B，线程2要拿锁A，双方都阻塞，造成死锁现象。

递归锁是RLock，是用来规避死锁现象，相当于是一串绑定在一起的钥匙，只要一个线程拿到了一个递归锁，那么其他线程就不能再拿任何锁，直到第一个线程释放锁。

使用方法和进程锁一致，lock=Lock(), lock.acquire(), lock.release()

**3.信号量和事件**

线程中的信号量和进程中一样，相当于是多把锁，Semaphore。事件也和进程中一样。

**4.线程队列**

虽然多线程中全局变量数据是共享的，但是还是会造成数据不安全问题，所以需要队列这种数据安全的东西来操作数据。

线程队列就是普通的队列，直接import queue

一共有三种队列：队列Queue（先进先出），栈LifoQueue（先进后出），优先级队列（PriorityQueue）。优先级队列put一个元祖，第一个元素是数字，数字越小优先级越高。

操作：put，get，put\_nowait，get\_nowait

**第三部分 协程**

协程是一个线程，能够在多个任务间来回切换，可以减少高IO的阻塞时间。

一个进程可以开20个线程，一个线程可以开500个协程

线程是cpu调度的最小单位，协程的切换是由greenlet模块实现的。线程是操作系统级别的任务切换，协程是代码级别的切换，协程的速度要快的多。

**一、greenlet模块**

当第一次switch时，从函数第一句开始执行，当遇到switch时，记录当前位置，跳转到另一个函数。

![ContractedBlock.gif][] ![ExpandedBlockStart.gif][]

from greenlet import greenlet
    
    def func1():
        print(111)
        g2.switch()
        print(111)
    
    def func2():
        print(222)
        g1.switch()
    
    g1 = greenlet(func1)
    g2 = greenlet(func2)
    g1.switch()

**二、gevent模块**

一般都使用gevent模块，他封装了greenlet的方法，当检测到阻塞时自动切换，但是他默认只能在遇到自己认识的阻塞时切换，如果想让所有阻塞均切换，需要在程序第一行引用：

from gevent import monkey;monkey.patch_all()

from gevent import monkey;monkey.patch_all()
    import time
    import gevent
    
    def func1():
        print("func1 start")
        time.sleep(2)
        print("func1 end")
    
    def func2():
        print("func2 start")
        time.sleep(2)
        print("func2 end")
    
    g1 = gevent.spawn(func1)
    g2 = gevent.spawn(func2)
    g1.join()    #如果没有join，则程序一运行就立刻关闭，没有任何输出
    g2.join()

**第四部分：IO模型**

**一、阻塞IO**

在通常情况下，等待数据会经历两个阶段，一是等待接收数据，数据会先被操作系统接收，二是从操作系统中将数据拷贝到程序中。这两个阶段都会造成阻塞，这就是最普通的阻塞IO模型。

**二、非阻塞IO**

socket提供了实现非阻塞的工具，sk.setblocking(False)，此后只要遇到阻塞就不等待。两种情况，没有收到连接或没有收到消息，都会报BlockingIOError。

但是这只能解决等待连接和消息的阻塞，不能解决从操作系统拷贝数据到程序时的阻塞。

![ContractedBlock.gif][] ![ExpandedBlockStart.gif][]

import socket
    sk = socket.socket()
    sk.bind(('127.0.0.1',8080))
    sk.listen()
    sk.setblocking(False)    #将socket变成非阻塞
    
    conn_del_list = []  #将所有已经断开的连接放到这个列表
    conn_list = []   #将接收到的连接都放到这个列表中
    while True:
        try:
            conn, addr = sk.accept()    #如果没有接收到链接会报错
            print("建立了连接")
            conn_list.append(conn)
        except BlockingIOError:
            for conn in conn_list:
                try:
                    ret = conn.recv(1024).decode("utf-8")     #如果没有收到消息也会报错
                    print(ret)
                    conn.send(b'byebye')
                except BlockingIOError:
                    pass
                except ConnectionAbortedError:    #如果对方断开连接报的错
                    conn.close()
                    # conn_list.remove(conn)    不能直接在for循环中删除列表的值，会造成索引错乱！
                    conn_del_list.append((conn))
            for conn in conn_del_list:
                conn_list.remove(conn)
            conn_del_list.clear()
    sk.close()

非阻塞IO

**三、IO多路复用**

IO多路复用的原理，是让一个代理帮助程序去检查是否有连接或消息到来，如果有就返回，这个代理就是操作系统。多路复用不需要像非阻塞模型一样一直询问是否有连接，而是如果没有就阻塞，操作系统会帮我们监听是否有连接或消息，如果有就返回给程序。

在windows中IO多路复用依靠select模块实现。

![ContractedBlock.gif][] ![ExpandedBlockStart.gif][]

import select
    import socket
    
    sk = socket.socket()
    sk.bind(('127.0.0.1',8080))
    sk.listen()
    sk.setblocking(False)
    
    read_list = [sk,]  #需要监听读操作的对象列表
    while True:
        rl, wl, xl = select.select(read_list, [], [])   #监听，没有信号就阻塞
        for i in rl:
            if i is sk:   #如果这是sk对象，代表有连接
                conn, addr = i.accept()
                read_list.append(conn)    #将这个conn也加入监听对象列表
            else:   #否则就是由消息
                try:
                    data = i.recv(1024).decode("utf-8")
                    print(data)
                    i.send(b'byebye')
                except ConnectionAbortedError:
                    i.close()
                    read_list.remove(i)

select模块实现IO多路复用

但是select模块并不是效率最高的，还有poll和epoll，这些在linux上有，可以使用selectors模块，自动依据操作系统选择最佳的模块。

![ContractedBlock.gif][] ![ExpandedBlockStart.gif][]

import socket
    
    sk = socket.socket()
    sk.bind(('127.0.0.1',8080))
    sk.listen()
    sk.setblocking(False)
    sel = selectors.DefaultSelector()   #selectors模块自动帮我们选择最合适的监听方式
    
    def accept(sk, mask):
        conn, addr = sk.accept()
        sel.register(conn, selectors.EVENT_READ, read)
    
    def read(conn, mask):
        try:
            data = conn.recv(1024).decode('utf-8')
            print(data)
            conn.send(b'byebye')
        except ConnectionAbortedError:
            conn.close()
            sel.unregister(conn)
    
    sel.register(sk, selectors.EVENT_READ, accept)   #注册要监听sk对象，如果有消息就执行accept回调函数
    
    while True:
        events = sel.select()   #检测所有注册的对象，看是否有消息
        for obj, mask in events:
            callback = obj.data   #取到这个对象绑定的回调函数名
            callback(obj.fileobj, mask)

selectors模块自动选择最合适的监听方式

**四、异步IO**

异步IO可以解决整体所有的阻塞，但是python代码无法提供异步IO功能。

**小结：**阻塞IO就是最正常的操作，会在等待连接和消息时阻塞；非阻塞IO会不停询问是否有连接或消息到来，虽然解决了等待的阻塞问题，但是while True占用太多系统内存，浪费很多资源；IO多路复用解决了非阻塞IO占用太多资源的问题，他找了操作系统做代理，自己不会一直询问，而是让操作系统去监听，当有连接或消息到来时给程序发一个信号，程序再去处理。这三种模型一种比一种好，但是都只能解决等待的阻塞，不能解决拷贝数据的阻塞，无法实现真正的异步。

转载于:https://www.cnblogs.com/yinwenjie/p/10847513.html

[ContractedBlock.gif]: https://images.cnblogs.com/OutliningIndicators/ContractedBlock.gif
[ExpandedBlockStart.gif]: /images/20211220/eec3f6ca59f548b9be7f058b04d21c4a.png