并发编程

目录

并发编程:

  • 程序:

    • 程序就是一堆文件
  • 进程:

    • 进程是分配资源的基本单位,为线程提供资源,一个程序可以开启多个进程

  • 进程被谁运行:

    • CPU最终运行你的程序

      • 操作系统调度作用,将你磁盘上的程序加载到内存,然后交给CPU处理,一个CPU在运行的一个程序,就是开启了一个进程

  • 操作系统:

    • 操作系统定义:
      • 操作系统是存在于硬件与软件之间,管理,协调,控制软件与硬件的交互

    • 操作系统的作用:
      • 如果没有操作系统,去写一个程序,你要完成两层功能:

        • 第一层:你要学会底层硬件:CPU,内存,磁盘是如何工作使用的
        • 第二层:去调度这些底层的硬件
      • 操作系统两个作用:

        • 1,将一些复杂的硬件操作封装成简单的接口,便于使用

        • 2,操作系统可以合理的调度分配多个进程与CPU的关系,让其有序化

    • 操作系统(计算机)的发展史:
      • 第一代电子计算机:操作插线与你的程序结合

      • 第二代计算机:磁带存储,批处理系

      • 第三代计算机:集成电路,多道程序系统

知识点解析:

  • 多道技术解决的问题:

    • 多道技术是在不同任务间切换执行,由于计算机切换速度非常快,用户是无感状态

    • 时间复用:
      • 利用闲置时间,进行复用,一个进程占用cpu时间太长也会切换
    • 空间复用:
      • 一个内存可以加载多个进程,提高内存的利用率
    • 数据隔离:
      • 解决软件之间的隔离,互不影响

进程:

  • 程序就是一堆代码

  • 进程是分配资源的基本单位,为线程提供资源,一个程序可以开启多个进程

  • 概念:
    • 串行:所有的进程有CPU一个一个解决
    • 并行:多个CPU,真正的同时运行多个进程
    • 并发:单个CPU,同时执行多个进程(来回切换),看起来像是同时运行,空间复用
    • 阻塞:遇到IO(recv,input)才会阻塞
    • 非阻塞:没有IO
  • tail -f access.log |grep '404'
    执行程序tail,开启一个子进程,执行程序grep,开启另外一个子进程,两个进程之间基于管道'|'通讯,将tail的结果作为grep的输入。
    进程grep在等待输入(即I/O)时的状态称为阻塞,此时grep命令都无法运行
    

  • 进程的创建:
    • 什么是开启多个进程:socket:server,client两个进程

    • python中,如果一次想开启多个进程,必须是一个主进程,开启多个子进程

    • linux,windows:有主进程开启子进程

    • 相同点:主进程开启子进程,两个进程都有相互隔离的独立空间,互不影响

    • 不同点:

      • linux:子进程空间的初始数据完全是从主(父)进程copy一份

      • windows:子进程空间初始数据完全是从主(父)进程copy一份,但是有所不同

创建进程的两种方法:

  • 函数-创建进程:

  • #这样的实例虽然创建了子进程,但是在生产环境中子进程结束的时间不定
    from multiprocessing import Process
    import time
    #当前py文件就是主进程,先运行主进程
    def task(name):
        print(f"{name}is running")
        time.sleep(3)  #阻塞
        print(f"{name}is done")
    
    if __name__ == '__main__':                  #windows开启必须写在mian下面
         p = Process(target=task,args=("海洋",)) #target要封装的内容,对象args一定是个元祖形式
         p.start()      #子进程 通知操作系统在内存中开辟一个空间,将p这个进程放进去,让cpu执行
         print("___主进程")
    

  • 类-创建进程(了解):

  • from multiprocessing import Process
    import time
    class MyProcess(Process):
    
        def __init__(self,name):
            super().__init__()      #放在最上面,必须要继承父类init
            self.name = name
    
        def run(self):
            print(f"{self.name}is running")
            time.sleep(3)  #阻塞
            print(f"{self.name}is done")
    
    if __name__ == '__main__':
        p = MyProcess("海洋")
        p.start()
        print("====主进程")
    

进程PID:

  • tasklist | findstr pycharm win查看某个进程

  • import os print(os.getpid()) 查看当前的pid

  • import os print(os.getppid()) 查看父进程

进程之间数据隔离:

  • import time
    from  multiprocessing import  Process
    X = 1000
    
    def task():
        global x
        x = 2
    
    if __name__ == '__main__':
        p1 = Process(target = task,)
        p1.start()
        time.sleep(1)
        print(f"主进程{X}")
        print(f"主进程{X}")
    
    import time
    from  multiprocessing import  Process
    X = 256                                #满足小数据池
    
    def task():
       print(f"子进程{id(X)}")
    
    if __name__ == '__main__':
        print(f"主进程{id(X)}")
        p1 = Process(target = task,)
        p1.start()
        time.sleep(1)
        print()
    

join方法:

  • join 主进程等待子进程结束之后,在执行

  • join开启一个进程:

    • from multiprocessing import Process
      import time
      
      def task(name):
          time.sleep(1)
          print(f"{name}is running")
      
      if __name__ == '__main__':
           p = Process(target=task,args=("海洋",))
           p.start()
           p.join()           #告知主进程,p进程结束之后,主进程在结束,join有些阻塞的意思
           print("___主进程")
      
      #      p1.start()
      #      p2.start()       #p1,p2,p3三个子进程先后运行顺序不定,start只是通知一下操作系统
      #      p3.start()       #操作系统调用cpu先运行谁,谁先执行
      
  • join串行:

    • from multiprocessing import Process
      import time
      
      def task(name,sec):
          time.sleep(sec)
          print(f"{name}is running")
      
      if __name__ == '__main__':
           p1 = Process(target=task, args=("海洋",1))
           p2 = Process(target=task, args=("俊丽",2))
           p3 = Process(target=task ,args=("宝宝",3))
           start_time = time.time()
      
           p1.start()
           p1.join()
           p2.start()
           p2.join()
           p3.start()
           p3.join()
      
           print(f"主进程{time.time() - start_time}")
      
  • join并发:

    • from multiprocessing import Process
      import time
      
      def task(sec):
          time.sleep(sec)
          print(f"is running")
      
      if __name__ == '__main__':
           start_time = time.time()
           list = []
      
           for i in range(1,4):
                p = Process(target=task, args=(i,))
                p.start()
                list.append(p)
      
           for i in list:
                i.join()
      
           print(f"主进程{time.time() - start_time}")
      

进程对象的其他属性:

  • 属性:

    • from multiprocessing import Process
      import time
      
      def task(name):
          print(f"{name}is running")
          time.sleep(3)
          print(f"{name}is done")
      
      if __name__ == '__main__':
           p = Process(target=task,args=("海洋",),name="俊丽")  #name给进程对象设置name属性
           p.start()
           # print(p.pid)         #获取到进程号
      
           time.sleep(1)          #睡一秒,子进程已经执行完成
           p.terminate()          #强制结束子进程,强制执行也会有执行时间
                                  #terminate跟start一样工作原理,都要通知操作系统开启子进程
                                  #内存终止或者开启都要需要时间的
                      
           time.sleep(1)          #睡一秒,让terminate杀死
           print(p.is_alive())    #判断子进程是否存活,只是查看内存中p子进程是否还运行
           print("主进程")
      
  • 僵尸进程:

    • init是所有进程的父进程:
      
      僵尸进程,僵尸是什么,死而没有消失
      
      主进程创建多个短暂周期的子进程,当子进程退出,是需要等待父进程处理,而父进程没有及时对子进程回收,那么子进程的进程符仍然保存在系统中,这种进程就是僵死进程
      
      什么进程描述符:每一个进程都有描述符,io请求,数据指针
      
      from multiprocessing import Process
      import time
      import os
      
      def task(name):
          print(f"{name}is running")
          print(f"子进程开始了:{os.getpid()}")
          time.sleep(50)
      
      
      if __name__ == '__main__':
          for i in range(100):
              p = Process(target=task, args=("海洋",))
              p.start()
              print(f"___主进程:{os.getpid()}")
      
      

  • 孤儿进程:

    • 孤儿进程:孤儿进程是因为主进程的退出,他下面的所有子进程都变成孤儿进程了,init会对孤儿进行回收,释		   放掉占用系统的资源,这种回收也是为了节省内存。
      
      孤儿进程无害,如果僵尸进程挂了,init会对孤儿进程回收,init是所有进程的祖进程,linux中为1,0系统
      
      

  • 守护进程:

    • 将一个子进程设置成守护进程,当父进程结束,子进程一定会结束,避免孤儿进程产生,应为回收机制

    • 父进程不能创建子进程

    • #守护进程会在主进程代码执行结束后终止,守护进程内无法在开启子进程
      
      from multiprocessing import Process
      import time
      import os
      
      def task(name):
          print(f"{name}is running")
          print(f"子进程开始了:{os.getpid()}")
          time.sleep(50)
      
      if __name__ == '__main__':
           p = Process(target=task,args=("海洋",))
           p.daemon = True  #将p子进程设置成守护进程,守护子进程,只要主进程结束
                            #子进程无论执行与否都马上结束,daemon,开启在start上面
           p.start()
           print(f"___主进程:{os.getpid()}")
      
      

进程之间的通信方式:

  • 第一种:基于文件+锁的形式:效率低,麻烦

  • 第二种:基于队列,推荐的使用形式

  • 第三种:基于管道,管道自己加锁,底层可能会出现数据丢失损坏,队列和管道都是将数据存放于内存中

互斥锁:

  • 互斥锁保证了每次只有一个线程进行写入操作,只有当这个线程解锁,在运行其他资源,上锁和解锁都需要自己添加

  • 三台电脑同时调用打印机去打印,开启三个进程使用互斥锁,实现公平抢占资源

    • #上锁:
      #一定要是同一把锁:只能按照这个规律,上锁一次,解锁一次
      
      #互斥锁与join区别:
      #共同点:都是完成了进程之间的串行
      #区别:join认为控制进程的串行,互斥锁是解决抢占的资源,保证公平性
      
      from multiprocessing import Process
      from multiprocessing import Lock
      import time
      import os
      import random
      
      def task1(lock):
          print("test1")                     #验证CPU遇到IO切换
          lock.acquire()
          print("task1 开始打印")
          time.sleep(random.randint(1,3))
          print("task1 打印完成")
          lock.release()
      
      def task2(lock):
          print("test2")
          lock.acquire()                      #上锁
          print("task2 开始打印")
          time.sleep(random.randint(1,3))#阻塞,cpu切换任务,别的任务都在锁,回来继续执行这个程序
          print("task2 打印完成")
          lock.release()                      #解锁
      
      def task3(lock):
          print("test2")
          lock.acquire()
          # lock.acquire()                    #死锁错误示例
          print("task3 开始打印")
          time.sleep(random.randint(1,3))
          print("task3 打印完成")
          lock.release()
      
      if __name__ == '__main__':
           lock = Lock()                              #一把锁
      
           p1 = Process(target=task1,args=(lock,))    #三个进程哪个先到先执行
           p2 = Process(target=task2,args=(lock,))
           p3 = Process(target=task3,args=(lock,))
      
           p1.start()
           p2.start()
           p3.start()
      
      
  • 互斥锁买票示例:

    • #买票系统:
      #买票之前先要查票,在你查票的同时,100个人也在查看此票
      #买票时,你要从服务端获取到票数,票数>0 ,买票,然后服务端票数减一,中间有网络延迟
      
      #多进程原则上是不能互相通信的,他们在内存级别是有数据隔离,不代表磁盘上的数据隔离,他们可以共同操作一个文件
      #多个进程抢占同一个资源,要想公平按照顺序,只能串行
      
      from multiprocessing import Process
      from multiprocessing import Lock
      import random
      import json
      import time
      import os
      
      def search():
          time.sleep(random.random())  #一秒之内
          with open("db.json", encoding="utf-8") as f1:
              dic = json.load(f1)
          print(f"剩余票数{dic['count']}")
      
      def get():
          with open("db.json",encoding="utf-8") as f1:
              dic = json.load(f1)
          time.sleep(random.randint(1,3))  #时间延迟
      
          if dic['count'] > 0:
              dic['count'] -= 1
              with open("db.json",encoding="utf-8",mode="w") as f1:
                  json.dump(dic,f1)
              print(f'{os.getpid()}用户购买成功')
          else:
              print("没票了")
      
      def task(lock):
          search()
      
          lock.acquire()
          get()
          lock.release()
      
      if __name__ == '__main__':
          lock = Lock()
          for i in range(5):
              p = Process(target=task,args=(lock,))
              p.start()
      缺点:
      	1.操作文件效率低
          2.自己加锁很麻烦,很容易出现死锁,递归锁
      
      

队列:

  • 进程之间的通信最好的方式是基于队列

  • 队列是实现进程之间通信的工具,存在内存中的一个容器,最大的特点是符合先进先出的原则

  • 队列模式:
    • 多个进程抢占一个资源:串行,有序以及数据安全,买票

    • 多个进程实现并发的效果:生产者消费模型

队列参数:

  • from multiprocessing import Queue
    q = Queue(3)                #可以设置元素个数,当数据已经达到上限,在插入夯住
    
    def func():
        print("in func")
    
    q.put("海洋")               #插入数据
    q.put({"count":1})
    q.put(func)
    q.put("333",block=False)    #默认为True 当你插入的数据超过最大限度,默认阻塞
    # q.put(333,timeout=8)      #超过八秒在put不进数据,就会报错
    
    print(q.get())
    print(q.get())
    ret = q.get()
    ret()
    # q.get()  #当你将数据取完,夯住,等待队列put值,起另一个进程往队列中插入数据
    
    #q.put()
    #1,maxsize()    #数据量不易过大,精简的重要数据
    #2,put bolck    #默认为True阻塞 当你插入的数据超过最大限度,可以设置报错
    #3,put timeout  #延时报错,超过三秒在put不进数据,就会报错
    
    #get
    #2,get bolck    #取值为空报错
    #3,get timeout  #取值超过三秒报错
    
    

抢售模型 (并行示例):

  • #小米:抢手机,预期发售10个,100人去抢
    from multiprocessing import Queue
    from multiprocessing import Process
    import os
    
    def task(q):
        try:
            q.put(f'{os.getpid()}')
        except Exception:
            return
    
    if __name__ == '__main__':
        q = Queue(10)             #创建队列,可以存放十个人
    
        for i in range(100):
            p = Process(target=task,args=(q,))
            p.start()
    
        for i in range(1,11):  #数量超过队列会取
            print(f'排名第{i}的用户:{q.get()}') #获取队列中的信息,先进来的先取出来
    
    #利用队列进行进程之间的通信:简单,方便,不用自己手动加锁,队列自带阻塞,可持续化取数据
    
    

生产者消费者模型(并发示例):

  • 利用队列进行通信,生产者生产数据,消费者获取数据使用,平衡了生产力和消费力,生产者和消费者是一种解耦合性(通过容器解决),可持续化取数据

  • 模型,设计模式,归一化设计,理论等等,教给你一个编程的思路,以后遇到类似的情况,以后直接调用就即可

  • 生产者:生产数据的进程

  • 消费者:生产出来的数据进行处理

  • #吃包子:厨师生产包子,不可能直接给你喂到嘴里,放在一个盆里,消费者从盆中取出包子食用
    #三个主体:生产者(厨师),容器队列(盘 缓冲区),消费者(人)
    
    #如果没有容器,生产者与消费者强解耦性,不合理,所以我们要有一个容器,缓冲区平衡了生产力与消费力
    
    # 生产者消费者多应用于并发:
    from multiprocessing import Queue
    from multiprocessing import Process
    import time
    import random
    
    def producer(name,q):
        for i in range(1,6):
            time.sleep(random.randint(1,3))
            res = f'{i}号包子'
            q.put(res)
            print(f'生产者{name}:生产了{res}')
    
    def consumer(name,q):
        while 1:
            try:
                time.sleep(random.randint(1, 3))
                ret = q.get(timeout = 5)           #五秒还吃不到退出
                print(f'消费者{name}:吃了{ret}')
            except Exception:
                return
    
    if __name__ == '__main__':
        q = Queue()    #盆
    
        p1 = Process(target=producer,args=("俊丽",q,))  #生产
        p2 = Process(target=consumer,args=("海洋",q,))  #消费
    
        p1.start()
        p2.start()
    
    

线程:

  • 进程:进程是分配资源的基本单位,内存中开辟空间,为线程提供资源,一个程序可以开启多个进程

  • 线程:CPU调度的最小单位,执行单位,线程也被称作为轻量级的进程,动态的

    • 主线程是进程空间存活在内存中的一个必要条件

  • 开启QQ:开启一个进程,在内存中开辟空间加载数据,启动一个线程执行代码

  • 线程依赖进程的一个进程可以包含多个线程,但是一定有一个主线程,线程才是CPU执行的最小单元

  • 进程线程对比:

    • 1,开启多进程开销非常大,10-100倍,而开启线程开销非常小

    • 2.开启多进程速度慢,开启多线程速度快

    • 3.进程之间数据不共享,线程共享数据

  • 多线程应用场景:

    • 并发:一个CPU可以来回切换(线程之间切换),多进程并发,多线程的并发

    • 多进程并发:开启多个进程,并发的执行

    • 多线程并发:开启线程,并发的执行

    • 如果遇到并发:多线程居多

开启线程的两种方式:

  • 线程绝对要比进程开启速度快

  • 函数开启:
    • #先打印海洋,线程要比进程速度快,如果是进程先打印主线程
      from threading import Thread
      
      def task(name):
          print(f'{name} is running')
      
      if __name__ == '__main__':
          t = Thread(target=task,args=("海洋",))
          t.start()
          print("主线程")
          
      #子进程睡眠3秒,先运行主进程
      from threading import Thread
      import time
      x = 1000
      
      def task():
          time.sleep(3)
          print('子线程....')
      
      def main():
          print('111')
          print('222')
          print('333')
      
      if __name__ == '__main__':
          t = Thread(target=task)
          t.start()
          main()
      
      
  • 类开启:
    • from threading import Thread
      
      class MyThread(Thread):
          def __init__(self,name):
              super().__init__()
              self.name = name
      
          def run(self):
              print(f'{self.name} is running')
      
      if __name__ == '__main__':
          t = MyThread("海洋")
          t.start()
          print("主线程")
      
      
  • 线程pid:
    • #主线程和子线程pid一样
      from threading import Thread
      import os
      
      def task():
          print(f'子线程:{os.getpid()}')
      
      if __name__ == '__main__':
          t = Thread(target=task,)
          t.start()
          print(f"主线程:{os.getpid()}")
      
      
  • 线程之间数据共享:
    • from threading import Thread
      x = 1000
      def task():
          global x
          x = 0
      
      if __name__ == '__main__':
          t = Thread(target=task, )
          t.start()
          t.join()  # 告知主线程,等待子线程运行完毕在执行
          print(f'主线程:{x}')
      
      

线程的方法:

  • from threading import Thread
    import threading
    import time
    
    def task(name):
        time.sleep(1)
        print(f'{name} is running')
    
    if __name__ == '__main__':
        for i in range(5):
            t = Thread(target=task,args=("海洋",))
            t.start()              #线程对象的方法
        # print(t.is_alive())     #判断线程是否存活
     			
        #threading模块的方法
        print(threading.current_thread().name)  #返回线程对象.name
        print(threading.enumerate())            #返回列表,返回的是所有线程对象
        print(threading.active_count())         #获取活跃的线程数量(包括主线程)
        print("主线程")
    
    

守护线程:

  • 守护线程必须等待主线程结束才结束,主线程必须等待所有的非守护线程结束才能结束,因为主线程的结束意味着进程的结束,这就是一个守护机制

  • 多线程是同一个空间,同一个进程,进程代表,空间,资源,静态的:

  • from threading import Thread
    import time
    def sayhi(name):
        time.sleep(2)
        print('%s say hello' %name)
    
    if __name__ == '__main__':
        t=Thread(target=sayhi,args=('egon',))
        t.setDaemon(True) #必须在t.start()之前设置
        t.start()
    
        print('主线程')
        print(t.is_alive())   #判断进程是否存在也是主线程
    
    from threading import Thread
    import time
    
    def foo():
        print(123)
        time.sleep(3)
        print("end123")
    
    def bar():
        print(456)
        time.sleep(1)
        print("end456")
    
    if __name__ == '__main__':
    
        t1=Thread(target=foo)
        t2=Thread(target=bar)
    
        t1.daemon = True
        t1.start()
        t2.start()		      #t2非守护线程,主线程等待子线程结束
        print("main-------")
    
    

线程互斥锁:

  • join:
  • from threading import Thread
    import time
    x = 100
    
    def task(name):
        global x
        temp = x
        time.sleep(3)
        temp -= 1
        x = temp
    
    if __name__ == '__main__':
        t = Thread(target=task,args=("海洋",))
        t.start()
        t.join()
        print(f"主线程{x}")
        
    #多个线程抢占一个资源
    from threading import Thread
    import time
    x = 100
    
    def task(name):
        global x
        temp = x
        time.sleep(3)
        temp -= 1
        x = temp
    
    if __name__ == '__main__':
        tl = []
        for i in range(100):
            t = Thread(target=task,args=("海洋",))
            tl.append(t)
            t.start()
    
        for i in tl:
            i.join()
    
        print(f"主进程{x}")   #多个线程抢占一个资源
    
    

互斥锁:

  • 所有线程串行执行,多个 线程共同抢占一个数据,保证了数据安全:

  • from threading import Thread
    from threading import Lock
    import time
    x = 100
    
    def task(lock):
        lock.acquire()
        global x
        temp = x
        time.sleep(0.1)
        temp -= 1
        x = temp
        lock.release()
    
    if __name__ == '__main__':
        lock = Lock()
        tl = []
        for i in range(100):
            t = Thread(target=task,args=(lock,))
            tl.append(t)
            t.start()
    
        for i in tl:
            i.join()
    
        print(f"主线程{x}")   #多个线程抢占一个资源,join让主线程等待子线程执行完成在执行,结果0
    
    

线程死锁现象:

  • 多个线程或者进程竞争资源,如果开启的互斥锁过多,遇到互相抢锁造成互相等待情况,程序夯住,

  • 还有一种是给同时给一个线程或者进程连续加锁多次,利用递归锁解决Rlock

  • from threading import Thread
    from threading import Lock
    import time
    
    lock_A = Lock()
    lock_B = Lock()
    
    class Mtthread(Thread):
        def run(self):
            self.f1()
            self.f2()
    
        def f1(self):
            lock_A.acquire()
            print(f"{self.name}谁拿到A锁")
    
            lock_B.acquire()
            print(f"{self.name}谁拿到B锁")
            lock_B.release()
    
            lock_A.release()
    
        def f2(self):
            lock_B.acquire()
            print(f"{self.name}谁拿到B锁")
    
            time.sleep(1)
            lock_A.acquire()
            print(f"{self.name}谁拿到A锁")
            lock_A.release()
    
            lock_B.release()
    
    if __name__ == '__main__':
        t1 = Mtthread()
        t1.start()
    
        t2 = Mtthread()
        t2.start()
    
        t3 = Mtthread()
        t3.start()
        print(f"主进程")  
    
    

递归锁:

  • 递归锁上有引用次数,每次引用计数+1,解锁计数-1,只有计数为0.在运行下个进程

  • #递归锁:
    #递归锁是一把锁,锁上有记录,只要acquire一次,锁上就计数一次,acquire2次就计数两次
    #release 1次减一,只要递归锁计数不为0,其他线程不能抢
    
    from threading import Thread
    from threading import RLock
    import time
    
    lock_A = lock_B = RLock()
    
    class Mtthread(Thread):
        def run(self):
            # lock_A.acquire()
            # lock_B.acquire()
            # print(111)
            # lock_A.release()
            # lock_B.release()
    
            self.f1()
            self.f2()
    
        def f1(self):
            lock_A.acquire()
            print(f"{self.name}谁拿到A锁")
    
            lock_B.acquire()
            print(f"{self.name}谁拿到B锁")
            lock_B.release()
    
            lock_A.release()
    
        def f2(self):
            lock_B.acquire()
            print(f"{self.name}谁拿到B锁")
    
            time.sleep(1)
            lock_A.acquire()
            print(f"{self.name}谁拿到A锁")
            lock_A.release()
    
            lock_B.release()
    
    if __name__ == '__main__':
        t1 = Mtthread()
        t1.start()
        t2 = Mtthread()
        t2.start()
        t3 = Mtthread()
        t3.start()
        print(f"主进程")  
    
    

信号量:

  • 信号量准许多个线程或者进程同时进入

  • from  threading import Thread
    from  threading import current_thread
    from  threading import Semaphore
    import time
    import random
    
    sm = Semaphore(4)
    
    def chi():
        sm.acquire()
        print(f"{current_thread().name}正在吃饭")
        time.sleep(random.randint(1,3))
        sm.release()
    
    if __name__ == '__main__':
        for i in range(20):
            t = Thread(target=chi)
            t.start()
    
    

GIL锁:

  • 全局解释器锁,就是一把互斥锁,将并发变成串行,同一时刻只能有一个线程进入解释器,自动加锁和释放锁,牺牲效率保护python解释器内部数据安全

  • 优点:
    • 强行加锁,保证解释器里面的数据安全

  • 缺点:
    • 多进程可以利用多核,多进程的每个进程里面都有python解释器程序

    • 单进程的多线程不能利用多核,python解释器内部程序,不支持多线程同时解释

  • 讨论:
    • python-单核处理IO阻塞的多线程,java多核处理IO阻塞问题,效率差不多

    • 单核处理三个IO线程,多核处理三个IO线程,多核快些

  • 代码的执行:
    • CPython独有GIL锁:
      • 将你的py文件当做实参传送给解释器传换成c语言字节码,在交给虚拟机转换成010101机器码,这些代码都是线程执行,进程进行调度资源

    • lpython:交互式解释器,可以补全代码

    • Jpython:java语言字节码,剩下的一样

    • pypy:动态编译,JAT技术,执行效率要比Cpython块,但是技术还有缺陷bug

验证Python开发效率:

  • 单核CPU:
    • 一核,都是单进程多线程并发快,因为单核开启多进程也是串行。

  • 多核CPU:
    • 计算密集型:

      • 多进程的并行比多线程的并发执行效率高很多(因为不同进程运行在不同核心上,并行执行)

    • IO密集型:

      • 多线程要比多进程处理速度快,因为进程开销大,而线程处理其实也是串行,只不过处理速度比进程更快些,线程一次只能处理一个事情(空间复用)

      • 开启150个进程(开销大,速度慢),执行IO任务耗时长

      • 开启150个线程(开销小,速度快),执行IO任务耗时短

  • 如果你的任务是io密集型并且任务数量大,用单进程下的多线程处理阻塞效率高

  • 计算密集型:
    • from multiprocessing import Process
      from threading import Thread
      import time
      import os
      # print(os.cpu_count())
      
      def task1():
          res = 1
          for i in range(1, 100000000):
              res += i
      def task2():
          res = 1
          for i in range(1, 100000000):
              res += i
      def task3():
          res = 1
          for i in range(1, 100000000):
              res += i
      def task4():
          res = 1
          for i in range(1, 100000000):
              res += i
      
      if __name__ == '__main__':
          # 四个进程 四个cpu 并行 效率
          start_time = time.time()
          p1 = Process(target=task1)
          p2 = Process(target=task2)
          p3 = Process(target=task3)
          p4 = Process(target=task4)
      
          p1.start()
          p2.start()
          p3.start()
          p4.start()
      
          p1.join()
          p2.join()
          p3.join()
          p4.join()
          print(f'主: {time.time() - start_time}')   # 10.125909328460693
      
      # 一个进程 四个线程
      #     start_time = time.time()
      #     p1 = Thread(target=task1)
      #     p2 = Thread(target=task2)
      #     p3 = Thread(target=task3)
      #     p4 = Thread(target=task4)
      #
      #     p1.start()
      #     p2.start()
      #     p3.start()
      #     p4.start()
      #
      #     p1.join()
      #     p2.join()
      #     p3.join()
      #     p4.join()
      #     print(f'主: {time.time() - start_time}')  # 22.927688121795654
      
      
  • 计算IO密集型:

    • from multiprocessing import Process
      from threading import Thread
      import time
      import os
      # print(os.cpu_count())
      
      def task1():
          res = 1
          time.sleep(3)
      
      if __name__ == '__main__':
      # 开启150个进程(开销大,速度慢),执行IO任务, 耗时 8.382229089736938
      #     start_time = time.time()
      #     l1 = []
      #     for i in range(150):
      #         p = Process(target=task1)
      #         l1.append(p)
      #         p.start()
      #     for i in l1:
      #         i.join()
      #     print(f'主: {time.time() - start_time}')
      
      # 开启150个线程(开销小,速度快),执行IO任务, 耗时 3.0261728763580322
      #     start_time = time.time()
      #     l1 = []
      #     for i in range(150):
      #         p = Thread(target=task1)
      #         l1.append(p)
      #         p.start()
      #     for i in l1:
      #         i.join()
      #     print(f'主: {time.time() - start_time}') 
      
      

GIL锁和互斥锁关系:

  • 线程计算密集型:
    • 当程序执行,开启100个线程时,第一个线程先要拿到GIL锁,然后拿到lock锁,执行代码,释放lock锁,最后释放GIL锁
  • 线程IO密集型:
    • 当程序执行,开启100个线程时,第一个线程先要拿到GIL锁,然后拿到lock锁,遇到阻塞,CPU切走,GIL释放,第一个线程挂起

    • 第二个线程执行,抢到GIL锁,进入要抢lock,但是lock锁还没释放,阻塞挂起

  • 自己加互斥锁,一定要加在处理共享数据的地方,加的范围不要扩大,范围过大,影响并发

  • GIL锁单进程的多线程不能利用多核,不能并行,但是可以并发

  • 互斥锁:
    • GIL自动上锁解锁,文件中的互斥锁Lock,手动上锁解锁

    • GIL锁,保护解释器的数据安全,互斥锁是保护的文件的数据安全

线程池:

  • 线程池在系统启动时创建了大量的空闲线程,线程执行直接调用线程池中已经开启好的空闲线程,当线程执行结束,该线程不会死亡,而是将线程变成空闲状态,放回进程池。

  • 线程池提高效率,资源复用

  • 进程池:放置进程的一个容器

  • 线程池:放置线程的一个容器

  • 完成一个简单的socket通信,服务端必须与一个客户端交流完毕,并且这个客户端断开连接之后,服务端才能接待下一个客户:

  • #开启进程池或者线程池:
    #线程池好还是进程池好:io阻塞或者计算密集型
    from  concurrent.futures import ProcessPoolExecutor
    from  concurrent.futures import ThreadPoolExecutor
    import time
    import os
    import random
    
    def task(name):
        # print(name)
        print(f"{os.getpid()}准备接客")
        time.sleep(random.randint(1,3))
    
    if __name__ == '__main__':
        # p = ProcessPoolExecutor(max_workers=5)  #限制进程数量,默认为cpu个数
        p = ThreadPoolExecutor()  				  #线程默认是CPU个数的五倍
    
        for i in range(23):
            p.submit(task,1)                      #给进程池放置任务启动,1为传参
    
    

阻塞,非阻塞:

  • 程序运行中的状态,阻塞,运行,就绪

  • 阻塞:当你程序遇到IO阻塞挂起,CPU切换,等到IO结束之后再执行

  • 非阻塞:程序没有IO,或者遇到IO通过某种手段让cpu去执行其他任务,尽可能的占用CPU

同步:

  • 任务发出去之后等待,直到这个任务最终结束之后,给我一个返回值,发布下一个任务

  • 同步示例:
  • from concurrent.futures import ProcessPoolExecutor
    import os
    import time
    import random
    
    def task():
        print(f"{os.getpid()}is running")
        time.sleep(1)
        return f'{os.getpid()} is finish'
    
    if __name__ == '__main__':
        p = ProcessPoolExecutor(4)
    
        for i in range(10):
            obj = p.submit(task,)
            print(obj.result())      #同步等待一个进程内容全部执行完成在执行下一个
    
    

异步:

  • 将任务发给进程,不管任务如何,直接运行下一个

  • 异步示例:
  • from concurrent.futures import ProcessPoolExecutor
    import os
    import time
    import random
    
    def task():
        print(f'{os.getpid()} is running')
        time.sleep(random.randint(0,2))
        return f'{os.getpid()} is finish'
    
    if __name__ == '__main__':
        p = ProcessPoolExecutor(4)
        obj_l1 = []
        for i in range(10):
            obj = p.submit(task,)   # 异步发出.
            obj_l1.append(obj)
    
        # time.sleep(3)
        p.shutdown(wait=True)
        # 1. 阻止在向进程池投放新任务,
        # 2. wait = True 十个任务是10,一个任务完成了-1,直至为零.进行下一行.
        for i in obj_l1:
            print(i.result())
    
    

异步+回调机制:

  • 异步发布任务,就不管任务结果
  • 回调:
    • 回调是你异步发布任务执行完成后,将结果丢给回调函数add_done_callback,回调函数帮你分析结果,进程继续完成下一个任务
    • 回调就是对特定的事件或者条件进行响应

  • 爬虫:游览器做的事情很简单:
    • 浏览器 封装头部,发送一个请求--->www.taobao.com ----> 服务器获取到请求信息,分析正确--->给你返回一个文件,--->游览器将这个文件的代码渲染,就成了你看的样子

    • 爬虫:利用reauests模块功能模拟游览器封装头,给服务器发送一个请求,骗过服务器之后,服务器也会给你返回一个文件,爬虫拿到文件,进行数据清洗获取到你想要的信息

  • 爬虫分两步:
    • 第一步:爬取服务器端的文件(IO阻塞)

    • 第二部:拿到文件,进行数据分析(非IO,IO极少)

  • 错误版本示例:
    • import requests
      from concurrent.futures import ProcessPoolExecutor
      from multiprocessing import Process
      import time
      import random
      import os
      
      def get(url):
          response = requests.get(url)
          print(f'{os.getpid()} 正在爬取:{url}')
          time.sleep(random.randint(1,3))
          if response.status_code == 200:
              return response.text
      
      def parse(text):
          print(f'{os.getpid()} 分析结果:{len(text)}')
      
      if __name__ == '__main__':
          url_list = [
              'http://www.taobao.com',
              'http://www.JD.com',
              'http://www.JD.com',
              'http://www.JD.com',
              'http://www.baidu.com',
              'https://www.cnblogs.com/jin-xin/articles/11232151.html',
              'https://www.cnblogs.com/jin-xin/articles/10078845.html',
              'http://www.sina.com.cn',
              'https://www.sohu.com',
              'https://www.youku.com',
          ]
          pool = ProcessPoolExecutor(4)
          obj_list = []
          for url in url_list:
              obj = pool.submit(get, url)
              obj_list.append(obj)
      
          pool.shutdown(wait=True)
      
          for obj in obj_list:          #抓取网页是串行,输出的结果
              parse(obj.result())
      
      #爬取一个网页需要2s,并发爬取10个网页:2.多s.
      #分析任务: 1s.    10s. 总共12.多秒.
       
      # 现在这个版本的过程:
      # 异步发出10个爬取网页的任务,然后4个进程并发(并行)的先去完成4个爬取网页的任务,然后谁先结束,谁进行下一个
      # 爬取任务,直至10个任务全部爬取成功.
      # 将10个爬取结果放在一个列表中,串行的分析.
      
      
      import requests
      from concurrent.futures import ProcessPoolExecutor
      from multiprocessing import Process
      import time
      import random
      import os
      
      def get(url):
          response = requests.get(url)
          print(f'{os.getpid()} 正在爬取:{url}')
          time.sleep(random.randint(1,3))
          if response.status_code == 200:
              parse(response.text)
      
      def parse(text):
          print(f'{os.getpid()} 分析结果:{len(text)}')
      
      if __name__ == '__main__':
          url_list = [
              'http://www.taobao.com',
              'http://www.JD.com',
              'http://www.JD.com',
              'http://www.JD.com',
              'http://www.baidu.com',
              'https://www.cnblogs.com/jin-xin/articles/11232151.html',
              'https://www.cnblogs.com/jin-xin/articles/10078845.html',
              'http://www.sina.com.cn',
              'https://www.sohu.com',
              'https://www.youku.com',
          ]
          pool = ProcessPoolExecutor(4)
          for url in url_list:
              obj = pool.submit(get, url)
      
          # pool.shutdown(wait=True)
          print('主')
      #异步发出10个 爬取网页+分析 的任务,然后4个进程并发(并行)的先去完成4个爬取网页+分析 的任务,
      #然后谁先结束,谁进行下一个 爬取+分析 任务,直至10个爬取+分析 任务全部完成成功.
      
      
      
      
  • 正确版本示例:
    • import requests
      from concurrent.futures import ProcessPoolExecutor
      from multiprocessing import Process
      import time
      import random
      import os
      
      def get(url):
          response = requests.get(url)
          print(f'{os.getpid()} 正在爬取:{url}')
          if response.status_code == 200:
              return response.text
      
      def parse(obj):
          time.sleep(1)
          print(f'{os.getpid()} 分析结果:{len(obj.result())}')
      
      if __name__ == '__main__':
      
          url_list = [
              'http://www.taobao.com',
              'http://www.JD.com',
              'http://www.JD.com',
              'http://www.JD.com',
              'http://www.baidu.com',
              'https://www.cnblogs.com/jin-xin/articles/11232151.html',
              'https://www.cnblogs.com/jin-xin/articles/10078845.html',
              'http://www.sina.com.cn',
              'https://www.sohu.com',
              'https://www.youku.com',
          ]
          start_time = time.time()
          pool = ProcessPoolExecutor(4)
          for url in url_list:
              obj = pool.submit(get, url)
              obj.add_done_callback(parse)
              # 增加一个回调函数
              # 现在的进程完成的还是网络爬取的任务,拿到了返回值之后,结果丢给回调函数add_done_callback,
              # 回调函数帮助你分析结果
              # 进程继续完成下一个任务.
          pool.shutdown(wait=True)   #阻止发布新的任务,代替join
      
          print(f'主: {time.time() - start_time}')
          
      # 回调函数是主进程帮助你实现的, 回调函数帮你进行分析任务. 明确了进程的任务: 只有一个网络爬取.
      # 分析任务: 回调函数执行了.对函数之间解耦.
      
      # 极值情况: 如果回调函数是IO任务,那么由于你的回调函数是主进程做的,所以有可能影响效率.
      
      # 回调不是万能的,如果回调的任务是IO,
      # 那么异步 + 回调机制 不好.此时如果你要效率只能牺牲开销,再开一个线程进程池.
      
      

队列模式:

  • FIFO 先进先出原则:
    • import queue
      q = queue.Queue(3)
      q.put(1)
      q.put(2)
      q.put('海洋')
      
      print(q.get())
      print(q.get())
      print(q.get())
      
      
  • LIFO 栈.-先进后出:
    • import queue
      q = queue.LifoQueue()
      q.put(1)
      q.put(3)
      q.put('海洋')
      
      print(q.get())
      print(q.get())
      print(q.get())
      
      
  • 优先级队列:
    • # 需要元组的形式,(int,数据) int 代表优先级,数字越低,优先级越高.
      import queue
      q = queue.PriorityQueue(3)
      
      q.put((10, '垃圾消息'))
      q.put((-9, '紧急消息'))
      q.put((3, '一般消息'))
      
      print(q.get())
      print(q.get())
      print(q.get())
      
      

事件Event:

  • 并发的执行某个任务,多进程多线程,几乎同时执行,一个线程执行到中间时,通知另一个线程开始执行

  • import time
    from threading import Thread
    from threading import current_thread
    from threading import Event
    			
    event = Event()  # 默认是False
    def task():
        print(f'{current_thread().name} 检测服务器是否正常开启....')
        time.sleep(3)   # 先运行task阻塞三秒,在将event修改为True
        event.set()     # 改成了True
    
    def task1():
        print(f'{current_thread().name} 正在尝试连接服务器')
        # event.wait()  # 轮询检测event是否为True,当其为True,继续下一行代码. 阻塞
        event.wait(1)
        # 设置超时时间,如果1s中以内,event改成True,代码继续执行.
        # 设置超时时间,如果超过1s中,event没做改变,代码继续执行.
        print(f'{current_thread().name} 连接成功')
        
    if __name__ == '__main__':
        t1 = Thread(target=task1,)
        t2 = Thread(target=task1,)
        t3 = Thread(target=task1,)
    
        t = Thread(target=task)
        t.start()
    
        t1.start()
        t2.start()
        t3.start()
    
    

协程:

  • 协程的本质也是一个线程,而使用协程目的是为了减少系统开销,协程是我们通过程序来控制任务切换,协程速度比系统更快,最大限度的利用CPU,更加轻量级

  • 线程协程的区别:

    • 协程没有锁,协程又称微线程
    • 线程和协程不同的是,线程是抢占式调度切换,而协程是需要自己调度
    • 线程和进程,调度是CPU决定的,而协程就是上帝,在一个线程中规定某个代码块的执行顺序

  • 1,协程切换开销更小,属于程序级别的切换,操作系统完全感知不到,更加轻量级

  • 2.单线程内就可以实现并发的效果,最大限度的利用CPU

  • 3.修改共享的数据不需要加锁

  • 协程就像线程一样也是在多任务间来回切换

  • 在其他语言中,协程的意义不大,多线程即可以解决I/O问题,在python中有GIL锁,在同一时间只有一个线程在工作,所以一个线程里面IO操作特别多,协程比较适用

  • 串行:多个任务执行时,一个任务从开始执行,遇到IO等待,等待IO阻塞结束之后再执行下一个

  • 并行:多核多个线程或者进程同时执行,四个CPU同时执行四个任务

  • 并发:多个任务看起来是同时执行,CPU在多个任务之间来回切换,遇到IO阻塞,计算密集型执行时间过长

    • 并发本质:遇到IO阻塞,计算密集型执行时间过长,保持原来的状态

  • 一个线程实现开发:

    • 多进程:操作系统控制,多个进程的多个任务切换 + 保持状态

    • 多线程程:操作系统控制,多个线程的多个任务切换 + 保持状态

    • 协程:程序控制一个线程的多个任务的切换以及保持状态

      • 微并发,处理任务不宜过多

      • 协程他会调度CPU,如果协程管控的任务中,遇到阻塞,他会快速的(比操作系统快),切换到另一个任务,并且能将上一个任务挂起(保持状态),让操作系统以为CPU一直在工作

  • 串行和协程对比:
    • 密集型数据串行和协程对比,肯定串行速度快,因为协程运行还要来回切换
    • import time
      def task1():
          res = 1
          for i in range(1,100000):
              res += i
      
      def task2():
          res = 1
          for i in range(1,100000):
              res -= i
      
      start_time = time.time()
      task1()
      task2()
      print(f'串行消耗时间:{time.time()-start_time}')  # 串行消耗时间:0.012489557266235352
      
      
      def task1():
          res = 1
          for i in range(1, 100000):
              res += i
              yield res
      
      def task2():
          g = task1()
          res = 1
          for i in range(1, 100000):
              res -= i
              next(g)
      
      start_time = time.time()
      task2()
      print(f'协程消耗时间:{time.time() - start_time}')  # 协程消耗时间:0.02991938591003418
      
      
  • 开启协程:
    • 遇到gevent阻塞切换:
    • import gevent
      import time
      def eat(name):
          print('%s eat 1' %name)     # 1
          gevent.sleep(2)              #协程识别gevent,可以进行IO切换
          # time.sleep(300)            #协程不识别切换不了,不可切换
          print('%s eat 2' %name)     # 4
      
      def play(name):
          print('%s play 1' %name)    # 2
          gevent.sleep(1)
          # time.sleep(3)
          print('%s play 2' %name)    # 3
      
      g1 = gevent.spawn(eat, '海洋')
      g2 = gevent.spawn(play, name='俊丽')   #协程异步发布任务
      # g1.join()
      # g2.join()
      #或者gevent.joinall([g1,g2])
      gevent.joinall([g1,g2])                #主线程等待协程执行完毕
      print('主')                            #5
      
      
    • 所有IO阻塞都可以切换:
    • import threading
      from gevent import monkey
      monkey.patch_all()         # 将你代码中的所有的IO都标识.
      
      import gevent              # 直接导入即可
      import time
      
      def eat():
          print(f'线程1:{threading.current_thread().getName()}')    # 1
          print('eat food 1')                                      # 2
          time.sleep(3)          # 加上mokey就能够识别到time模块的sleep了
          print('eat food 2')                                      # 6
      
      def play():
          print(f'线程2:{threading.current_thread().getName()}')    # 3
          print('play 1')                                          # 4
          time.sleep(1)  
          # 来回切换,直到一个I/O的时间结束,这里都是我们个gevent做得,不再是控制不了的操作系统了。
          print('play 2')                                          # 5
      
      g1=gevent.spawn(eat)
      g2=gevent.spawn(play)
      gevent.joinall([g1,g2])
      print(f'主:{threading.current_thread().getName()}')          # 7
      
原文地址:https://www.cnblogs.com/haiyang11/p/11214538.html