Python基础之并发编程

迈不过友情╰ 2022-03-06 08:40 180阅读 0赞

在 Python 学习的过程中，并发编程有些概念我们可能在理解上有些模糊，这些概念却很重要。下面我讲逐一解释一下并发和并行，同步和异步，阻塞和非阻塞。

## 并发&并行 ##

并发：在 OS 中，是指一个时间段中有几个程序都处于已启动运行到运行完毕之间，且这几个程序都是在同一个处理机上运行，但任一个时刻点上只有一个程序在处理机上运行。简言之，是指系统具有处理多个任务的能力。

并行：当系统有一个以上 CPU 时，则线程的操作有可能非并发。当一个 CPU 执行一个线程时，另一个 CPU 可以执行另一个线程，两个线程互不抢占 CPU 资源，可以同时进行，这种方式我们称之为并行(Parallel)。简言之，是指系统具有同时处理多个任务的能力。

实例1：

import time
    
    
    def music():
        print('Begin music time:{}'.format(time.ctime()))
        time.sleep(3)
        print('Stop music time:{}'.format(time.ctime()))
    
    
    def game():
        print('Begin game time:{}'.format(time.ctime()))
        time.sleep(5)
        print('Stop game time:{}'.format(time.ctime()))
    
    
    if __name__ == '__main__':
        start_time = time.ctime()
        print('Start time:{}'.format(start_time))
    
        music()
        game()
    
        end_time = time.ctime()
        print('End time:{}'.format(time.ctime()))

运行：

> Start time:Mon Mar 18 12:35:23 2019  
> Begin music time:Mon Mar 18 12:35:23 2019  
> Stop music time:Mon Mar 18 12:35:26 2019  
> Begin game time:Mon Mar 18 12:35:26 2019  
> Stop game time:Mon Mar 18 12:35:31 2019  
> End time:Mon Mar 18 12:35:31 2019
> 
> Process finished with exit code 0

music 的时间为3秒，game 的时间为5秒，如果按照我们正常的执行，直接执行函数，那么将按顺序顺序执行，整个过程8秒。

实例2：

import threading                                         
    import time                                              
                                                             
                                                             
    def music():                                             
        print('Begin music time:{}'.format(time.ctime()))    
        time.sleep(3)                                        
        print('Stop music time:{}'.format(time.ctime()))     
                                                             
                                                             
    def game():                                              
        print('Begin game time:{}'.format(time.ctime()))     
        time.sleep(5)                                        
        print('Stop game time:{}'.format(time.ctime()))      
                                                             
                                                             
    if __name__ == '__main__':                               
        start_time = time.ctime()                            
        print('Start time:{}'.format(start_time))            
                                                             
        t1 = threading.Thread(target=music)                  
        t2 = threading.Thread(target=game)                   
                                                             
        t1.start()                                           
        t2.start()                                           
                                                             
        t2.join()  # 等待子线程执行完毕                               
                                                             
        end_time = time.ctime()                              
        print('End time:{}'.format(time.ctime()))

运行：

> Start time:Mon Mar 18 12:36:38 2019  
> Begin music time:Mon Mar 18 12:36:38 2019  
> Begin game time:Mon Mar 18 12:36:38 2019  
> Stop music time:Mon Mar 18 12:36:41 2019  
> Stop game time:Mon Mar 18 12:36:43 2019  
> End time:Mon Mar 18 12:36:43 2019
> 
> Process finished with exit code 0

在这个例子中，我们开了两个线程，将 music 和 game 两个函数分别通过线程执行，运行结果显示两个线程同时开始，由于听音乐时间3秒，玩游戏时间5秒，所以整个过程完成时间为5秒。我们发现，通过开启多个线程，原本8秒的时间缩短为5秒，原本顺序执行现在是不是看起来好像是并行执行的？看起来好像是这样，听音乐的同时在玩游戏，整个过程的时间随最长的任务时间变化。但真的是这样吗？那么下面我来提出一个 GIL 锁的概念。

## GIL 锁 ##

GIL(全局解释器锁）：无论你启多少个线程，你有多少个 CPU, Python 在执行的时候会淡定的在同一时刻只允许一个线程运行。

下面我们对比实例3和实例4：

实例3：

import time
    
    
    def add():
        res = 0
        i = 1
        while i <= 1000000:
            res += i
            i += 1
        print('Add res:', res)
    
    
    def mul():
        res = 1
        i = 1
        while i <= 100000:
            res = res * i
            i += 1
        print('Mul res:', res)
    
    
    if __name__ == '__main__':
        restart = time.time()
    
        add()
        mul()
    
        print('Running time:%s s' % (time.time() - restart))

运行：

> 略

实例4：

import time
    import threading
    
    
    def add():
        res = 0
        i = 1
        while i <= 1000000:
            res += i
            i += 1
        print('Add res:', res)
    
    
    def mul():
        res = 1
        i = 1
        while i <= 100000:
            res = res * i
            i += 1
        print('Mul res:', res)
    
    
    if __name__ == '__main__':
        restart = time.time()
    
        t1 = threading.Thread(target=add)
        t2 = threading.Thread(target=mul)
    
        lst = [t1, t2]
        for t in lst:
            t.start()
    
        for t in lst:
            t.join()
    
        print('Running time:%s s' % (time.time() - restart))

运行：

> 略

哎吆，这是怎么回事，串行执行比多线程还快？不符合常理呀。是不是颠覆了你的人生观，这个就和 GIL 锁有关，同一时刻，系统只允许一个线程执行，那么，就是说，本质上我们之前理解的多线程的并行是不存在的，那么之前的例子为什么时间确实缩短了呢？这里有涉及到一个任务的类型。

> 任务类型：
> 
> 1.  IO 密集型（会有 CPU 空闲的时间）  注：sleep 等同于 IO 操作， socket 通信也是 IO 
> 2.  计算密集型

而之前那个例子恰好是IO密集型的例子，后面这个由于涉及到了加法和乘法，属于计算密集型操作，那么，就产生了如下结论：

*  多线程对于 IO 密集型任务有作用，而计算密集型任务不推荐使用多线程。
 *  由于GIL锁，多线程不可能真正实现并行，所谓的并行也只是宏观上并行微观上并发，本质上是由于遇到 IO 操作不断的 CPU 切换所造成并行的现象。由于 CPU 切换速度极快，所以看起来就像是在同时执行。

结果就是也就是说没有利用多核的优势，这就造成了多线程不能同时执行，并且增加了切换的开销，串行的效率可能更高。

## 同步&异步 ##

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

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

同步：当进程执行 IO等（待外部数据）的时候必须等待。例如打电话的时候必须等。

异步：当进程执行 IO等（待外部数据的）时候不需要等待，去执行其他任务，一直等到数据接收成功，再回来处理。例如发短信的时候。

当我们去爬取一个网页的时候，要爬取多个网站，有些人可能会发起多个请求，然后通过函数顺序调用。执行顺序也是先调用先执行，效率非常低。下面我们看一下异步的一个例子：

import socket
    import select
    
    """
    ########http请求本质，IO阻塞########
    sk = socket.socket()
    #1.连接
    sk.connect(('www.baidu.com',80,)) #阻塞
    print('连接成功了')
    #2.连接成功后发送消息
    sk.send(b"GET / HTTP/1.0\r\nHost: baidu.com\r\n\r\n")
    
    #3.等待服务端响应
    data = sk.recv(8096)#阻塞
    print(data) #\r\n\r\n区分响应头和影响体
    
    #关闭连接
    sk.close()
    """
    """
    ########http请求本质，IO非阻塞########
    sk = socket.socket()
    sk.setblocking(False)
    #1.连接
    try:
        sk.connect(('www.baidu.com',80,)) #非阻塞，但会报错
        print('连接成功了')
    except BlockingIOError as e:
        print(e)
    
    #2.连接成功后发送消息
    sk.send(b"GET / HTTP/1.0\r\nHost: baidu.com\r\n\r\n")
    
    #3.等待服务端响应
    data = sk.recv(8096)#阻塞
    print(data) #\r\n\r\n区分响应头和影响体
    
    #关闭连接
    sk.close()
    """
    
    
    class HttpRequest:
        def __init__(self, sk, host, callback):
            self.socket = sk
            self.host = host
            self.callback = callback
    
        def fileno(self):
            return self.socket.fileno()
    
    
    class HttpResponse:
        def __init__(self, recv_data):
            self.recv_data = recv_data
            self.header_dict = {}
            self.body = None
    
            self.initialize()
    
        def initialize(self):
            headers, body = self.recv_data.split(b'\r\n\r\n', 1)
            self.body = body
            header_list = headers.split(b'\r\n')
            for h in header_list:
                h_str = str(h, encoding='utf-8')
                v = h_str.split(':', 1)
                if len(v) == 2:
                    self.header_dict[v[0]] = v[1]
    
    
    class AsyncRequest:
        def __init__(self):
            self.conn = []
            self.connection = []  # 用于检测是否已经连接成功
    
        def add_request(self, host, callback):
            try:
                sk = socket.socket()
                sk.setblocking(0)
                sk.connect((host, 80))
            except BlockingIOError as e:
                pass
            request = HttpRequest(sk, host, callback)
            self.conn.append(request)
            self.connection.append(request)
    
        def run(self):
    
            while True:
                rlist, wlist, elist = select.select(self.conn, self.connection, self.conn, 0.05)
                for w in wlist:
                    print(w.host, '连接成功...')
                    # 只要能循环到，表示socket和服务器端已经连接成功
                    tpl = "GET / HTTP/1.0\r\nHost:%s\r\n\r\n" % (w.host,)
                    w.socket.send(bytes(tpl, encoding='utf-8'))
                    self.connection.remove(w)
                for r in rlist:
                    # r,是HttpRequest
                    recv_data = bytes()
                    while True:
                        try:
                            chunck = r.socket.recv(8096)
                            recv_data += chunck
                        except Exception as e:
                            break
                    response = HttpResponse(recv_data)
                    r.callback(response)
                    r.socket.close()
                    self.conn.remove(r)
                if len(self.conn) == 0:
                    break
    
    
    def f1(response):
        print('保存到文件', response.header_dict)
    
    
    def f2(response):
        print('保存到数据库', response.header_dict)
    
    
    url_list = [
        {'host': 'www.youku.com', 'callback': f1},
        {'host': 'v.qq.com', 'callback': f2},
        {'host': 'www.cnblogs.com', 'callback': f2},
    ]
    
    req = AsyncRequest()
    for item in url_list:
        req.add_request(item['host'], item['callback'])
    
    req.run()

运行：

> 略

我们可以看到，三个请求发送顺序与返回顺序，并不一样，这样就体现了异步请求。即我同时将请求发送出去，哪个先回来我先处理哪个。

即我们可以理解为：我打电话的时候只允许和一个人通信，和这个人通信结束之后才允许和另一个人开始。这就是同步。

我们发短信的时候发完可以不去等待，去处理其他事情，当他回复之后我们再去处理，这样就大大解放了我们的时间。这就是异步。

体现在网页请求上面就是我请求一个网页时候等待他回复，否则不接收其它请求，这就是同步。另一种就是我发送请求之后不去等待他是否回复，而去处理其它请求，当处理完其他请求之后，某个请求说，我的回复了，然后程序转而去处理他的回复数据。这就是异步请求。所以，异步可以充分 CPU 的效率。

## 阻塞&非阻塞 ##

阻塞：从调用者的角度出发，如果在调用的时候，被卡住，不能再继续向下运行，需要等待，就说是阻塞。

非阻塞： 从调用者的角度出发， 如果在调用的时候，没有被卡住，能够继续向下运行，无需等待，就说是非阻塞。

调用 blocking IO 会一直 block 住对应的进程直到操作完成，而 non-blocking IO 在 kernel 还准备数据的情况下会立刻返回。

服务端：

from socket import *
    import subprocess
    import struct
    
    ip_port = ('127.0.0.1', 8080)
    buffer_size = 1024
    backlog = 5
    
    tcp_server = socket(AF_INET, SOCK_STREAM)
    tcp_server.setsockopt(SOL_SOCKET, SO_REUSEADDR, 1)  # 就是它，在bind前加
    tcp_server.bind(ip_port)
    tcp_server.listen(backlog)
    
    while True:
        conn, addr = tcp_server.accept()
        print('新的客户端链接：', addr)
        while True:
            try:
                cmd = conn.recv(buffer_size)
                print('收到客户端命令:', cmd.decode('utf-8'))
    
                # 执行命令cmd，得到命令的结果cmd_res
                res = subprocess.Popen(cmd.decode('utf-8'), shell=True,
                                       stderr=subprocess.PIPE,
                                       stdout=subprocess.PIPE,
                                       stdin=subprocess.PIPE,
                                       )
                err = res.stderr.read()
                if err:
                    cmd_res = err
                else:
                    cmd_res = res.stdout.read()
                if not cmd_res:
                    cmd_res = '执行成功'.encode('gbk')
                # 解决粘包
                length = len(cmd_res)
                data_length = struct.pack('i', length)
                conn.send(data_length)
                conn.send(cmd_res)
            except Exception as e:
                print(e)
                break
        conn.close()

客户端：

from socket import *
    import struct
    
    PORT = ('127.0.0.1', 8080)
    BUFFER_SIZE = 1024
    BACKLOG = 5
    
    tcp_client = socket(AF_INET, SOCK_STREAM)
    tcp_client.connect(PORT)
    
    while True:
        cmd = input('>>:').strip()
        if not cmd:
            continue
        if cmd == 'quit':
            break
        tcp_client.send(cmd.encode('utf-8'))
    
        # 解决粘包
        length = tcp_client.recv(4)
        length = struct.unpack('i', length)[0]
    
        recv_size = 0
        recv_msg = b''
        while recv_size < length:
            recv_msg += tcp_client.recv(BUFFER_SIZE)
            recv_size = len(recv_msg)
    
        print(recv_msg.decode('gbk'))

运行：

> >>:dir  
>  驱动器 F 中的卷是 Project  
>  卷的序列号是 D05A-A7B8
> 
>  F:\\Python\\并发编程\\阻塞 的目录
> 
> 2019/03/18  14:32    <DIR>          .  
> 2019/03/18  14:32    <DIR>          ..  
> 2019/03/18  14:14               626 cli.py  
> 2019/03/18  14:32             1,401 ser.py  
>                2 个文件          2,027 字节  
>                2 个目录 168,398,262,272 可用字节
> 
> >>:

开启了服务器和一个客户端之后，我们在客户端输入一些命令，然后正确显示，功能实现。这是在我再打开一个客户端，输入命令，发现服务器迟迟没有响应。

这个就是当一个客户端在请求的时候，当这个客户端没有结束的时候，服务器不会去处理其他客户端的请求。这时候就阻塞了。如何让服务器同时处理多个客户端请求呢？

服务端：

import socketserver
    
    
    class Myserver(socketserver.BaseRequestHandler):
        """socketserver内置的通信方法"""
    
        def handle(self):
            print('conn is:', self.request)  # conn
            print('addr is:', self.client_address)  # addr
    
            while True:
                try:
                    # 发消息
                    data = self.request.recv(1024)
                    if not data: break
                    print('收到的客户端消息是:', data.decode('utf-8'), self.client_address)
    
                    # 发消息
                    self.request.sendall(data.upper())
                except Exception as e:
                    print(e)
                    break
    
    
    if __name__ == '__main__':
        s = socketserver.ThreadingTCPServer(('127.0.0.1', 8000), Myserver)  # 通信循环
        # s = socketserver.ForkingTCPServer(('127.0.0.1',8000), Myserver)  #通信循环
        print(s.server_address)
        print(s.RequestHandlerClass)
        print(Myserver)
        print(s.socket)
        s.serve_forever()

客户端：

from socket import *
    
    ip_port = ('127.0.0.1', 8000)
    buffer_size = 1024
    backlog = 5
    
    tcp_client = socket(AF_INET, SOCK_STREAM)
    tcp_client.connect(ip_port)
    
    while True:
        msg = input('>>:').strip()
        if not msg: continue
        if msg == 'quit': break
    
        tcp_client.send(msg.encode('utf-8'))
    
        data = tcp_client.recv(buffer_size)
        print(data.decode('utf-8'))
    
    tcp_client.close()

运行：

> 略

这段代码通过 socketserver 模块实现了 socket 的并发。这个过程中，当一个客户端在向服务器请求的时候，另一个客户端也可以正常请求。服务器在处理一个客户端请求的时候，另一个请求没有被阻塞。

总结：

只要有阻塞，就是阻塞 IO异；步 IO 的特点就是全程无阻塞。

有些人常把同步阻塞和异步非阻塞联系起来，但实际上经过分析，阻塞与同步，非阻塞和异步的定义是不一样的。同步和异步的区别是遇到IO请求是否等待。阻塞和非阻塞的区别是数据没准备好的情况下是否立即返回。同步可能是阻塞的，也可能是非阻塞的，而非阻塞的有可能是同步的，也有可能是异步的。

详情代码：[https://github.com/vadonical/python/tree/master/%E5%B9%B6%E5%8F%91%E7%BC%96%E7%A8%8B][https_github.com_vadonical_python_tree_master_E5_B9_B6_E5_8F_91_E7_BC_96_E7_A8_8B]

[https_github.com_vadonical_python_tree_master_E5_B9_B6_E5_8F_91_E7_BC_96_E7_A8_8B]: https://github.com/vadonical/python/tree/master/%E5%B9%B6%E5%8F%91%E7%BC%96%E7%A8%8B