python并发编程：多线程，多进程

Posted 2021-11-28 lancelotxly

‘‘‘
多进程(线程)编程:
         1. 进程和线程
         2. 并发与并行
         3. 同步和异步
         4. 阻塞和非阻塞
         5. 进程/线程的调度模型
‘‘‘ # 综述

‘‘‘
进程和线程:
    1. 进程: 
         1) 是一个程序在数据集上的一次动态执行过程
         2) 由程序, 数据集, 进程控制模块组成
    
    2. 线程: 
         1) 是CPU的一个最小执行单元, 
         2) 线程的出现是为了降低进程间切换的消耗
         3) 实现在一个进程内的并发
         4) 由线程ID, 程序计数器, 寄存器集合, 堆栈组成
    
    3. 进程和线程的关系:
         1) 进程是线程的容器, 程序至少有一个进程, 一个进程至少有一个线程
         2) 一个进程内的线程们共享进程资源, 线程几乎不拥有资源
         3) 进程之间的资源不共享
    
    4. 多进程/多线程的切换方式:
         1) IO阻塞
         2) 时间轮询               
‘‘‘ # 进程和线程

‘‘‘
并发和并行:
    1. 并发:  系统具有处理多个任务的能力,(可能不是并行, 如快速切换)
    
    2. 并行:  系统具有‘同时‘处理多个任务的能力
    
    3. 并发和并行的关系:  并行是并发的子集
‘‘‘ # 并发和并行

‘‘‘
同步和异步:
    1. 同步: 当进程(线程)执行IO操作时, 没得到结果就等待
    
    2. 异步: 当进程(线程)执行IO操作时, 不等待, 数据接收成功后, 再回来处理
    
    3. 同步和异步时针对任务的调度方式而言(等待/不等待)
‘‘‘ # 同步和异步

‘‘‘
阻塞和非阻塞:
    1. 阻塞: 当进程(线程)执行IO操作时, 该进程(线程)被挂起, 直到有结果时才被激活
    
    2. 非阻塞: 当进程(线程)执行IO操作时, 该进程(线程)不会被挂起, 没有结果也返回
    
    3. 阻塞和非阻塞时针对进程(线程)的调度(挂起/不挂起)
‘‘‘ # 阻塞和非阻塞

‘‘‘
进程(线程)的调度模型: 
     1. 进程(线程)的三种状态: 就绪, 执行, 阻塞
     2. 调度模型:

                ------执 行--------------------
                |        ^_______            |
               等待IO             |       执行其他线程
                |             执行该线程       |
                v                |            |
               阻 塞 ---有效IO--> 就 绪 <-------
‘‘‘ # 进程(线程)调度模型

‘‘‘
Python的GIL全局解释器锁
    1. Python没有多线程: 同一个时刻, 在一个进程内(无论有多少线程), Python解释器只允许执行一个线程
    
    2. 一般任务类型:
                 1) IO密集型 --> IO切换(阻塞即切换)     无影响
                 2) 计算密集型 --> 轮询切换             Python不适合用于计算密集型任务
    
    3. 解决GIL锁的方法:
          多进程 + 协程 (但主要还是为IO密集型服务的)                     
‘‘‘ # GIL全局锁

‘‘‘
多线程使用:
     1. 导入模块 import threading
     2. 开启线程的方法:
        1) 直接开启:  
                  1> 创建Thread对象
                     t = threading.Thread(target=func,args=(a,),kwargs={‘k1‘:v1})
                  2> 开启线程
                     t.start
        2) 继承开启:
                  1> 继承Thread类, 重写run()方法[即要执行的函数]
                     class MyThread(threading.Thread):
                          def __init__(self):
                             super(MyThread,self).__init__()
                             # 其他初始化代码,如要传给函数值
                          def run(self): pass
                  2. 创建实例,开启显线程
                     t = MyThread()
                     t.start()           # 会调用其run()方法
     
     3. 子线程和主线程的关系:
        1> 主线程和子线程分别执行, 主线程会等待最后一个子线程执行完毕后, 关闭
        
        2> 子线程t.join()到主线程, 会阻塞主线程, 执行完该子线程后主线程继续执行, 最后主线程等待最后一个子线程执行完毕后, 关闭
        
        3> 在子线程开启之前, 申明该子线程为t.setDaemon(True)守护线程, 最后主线程不会等待该子线程, 其他线程执行完就关闭主线程
     
     4. 线程的属性和方法:
        1) 对象属性和方法: t.attr t.func()    
             t.func(): 
                     t.start()          # 启动线程
                     t.join()           # 阻塞当前线程
                     t.setDaemon(True)  # 设置为守护线程
                     t.run()            # 在当前线程下调用该线程的run方法
                     
                     t.getName()        # 获取线程名
                     t.setName()        # 设置线程名
                     
                     t.isAlive()        # 判断线程是否被激活
                     t.isDaemon()       # 判断线程是否为守护线程       
             t.attr:
                     t.name                # 线程名
                     t.daemon = True/False # 是否为守护线程
        
        2) threading方法:
             threading.currentThread()     # 当前线程对象   
             threading.enumerate()         # 返回一个List包含当前所有活跃的线程对象
             thread.activeCount()          # 相等于len(threading.enumerate())                                                     
‘‘‘ # 多线程: 使用, 见Mulit_Threading.py

‘‘‘
线程之间通信:
一、同步锁(Lock):
       1. 多线程操作公共变量的问题:
一个线程还未来得及对公共变量进行操作, 就又有其他线程来访问该公共变量并修改了值

       2. 同步锁的作用:  部分串行     
          将一部分代码打包成一个整体(加锁), 同一个时刻只有获得锁的线程才能运行锁内的代码
          Lock = threading.Lock()
          Lock.acquire()
          # code
          Lock.release()

       3. GIL下的Lock:  线程A，线程B，线程C,...
                       #1. 多线程争夺GIL, 若A抢到GIL的获得运行权限
                       #2. 若A有Lock则执行锁内代码
                       ...
                       #3. 多线程继续争夺GIL, 若B抢到GIL
                       #4. 此时A尚未执行完锁内代码, 锁未释放; B没有Lock, 阻塞
                       ...
                       #5. 直到A再次抢到GIL,执行完锁内代码,释放Lock
       4. join和Lock:
           t.join()是将该线程整体阻塞到当前线程, 即整体串行
           Lock是部分串行

       5. 应用: 账户模型    


二、死锁:  同步锁不能解决锁中锁的问题
      一把锁不能重用导致的, 后面的代码需要前面的锁, 但前面的锁已经被其他线程抢到了

      考虑: 线程A, 线程B, Lock_A(Lock_B), Lock_B(Lock_A)
          1. 线程A先竞争到了Lock_A, 又竞争到了Lock_B, 执行锁中代码
          2. 线程A执行完了, 顺序释放Lock_B, Lock_A
          3. 线程A竞争到了Lock_B, 线程B竞争到了Lock_A
          4. 死锁, 线程A在等Lock_A, 线程B在等Lock_B

三、递归锁(RLock): 解决死锁问题 
              可重入锁, 可递归使用, 没用完就不会释放
      1. 原理:
         1) RLock内部维护着一个Lock和一个counter变量，counter记录了acquire的次数，从而使得资源可以被多次require。
         2) 直到一个线程所有的acquire都被release，其他的线程才能获得资源
      2.使用:
          RLock = threading.RLock()
          with RLock:
              #code 1
              with RLock:
                  #code 2
‘‘‘ # 多线程通信: 同步锁(threading.Lock), 死锁, 递归锁(threading.RLock)

‘‘‘
同步对象(Event): 
       1.作用: 多线程运行中, 每个线程都是独立运行, Event保证其他线程需要等待某个线程的指令才能执行

       2. 使用: 通过event.wait()来阻塞线程，event.wait()后的代码必须等待某一个线程event.set()执行了才能执行
              #1. Event = threading.Event()
              #2. 指定某个线程为大哥, Event.set()发出指令
              #3. 将其他线程设为小弟, Event.wait(), 要等待大哥发出指令后，Event.wait()之后代码才会执行
                  执行后将Event.clear() 等待下一个指令
       3. 方法:
          Event.isSet()
          Event.set()

          Event.wait()            
          Event.clear()
‘‘‘ # 多线程通信: 同步对象(threading.Event)

‘‘‘
信号量(Semaphore): 
       1. 信号量与进(线)程池的区别:
信号量是, 可以开启无限多的进程(线程), 但某一时刻只允许num个进程(线程)修改数据(涉及加锁)
进程池是, 始终只开启num个进程(没有加锁)

       2. 使用:
              #1. Semaphore = threading.Semaphore(num)    // 创建一个num线程的Lock
              #2. Semaphore.acquire()                     // 从池子中取一个线程, 没有则阻塞
                  # code
                  Semaphore.release()                     // 释放一个空间
‘‘‘ # 多线程通信: 信号量(threading.Semaphore)

‘‘‘
线程队列(queue): 
       1. 线程安全: 
          1) list不是线程安全的数据类型
          2) 保证线程安全的方法: 
                            1> 加锁
                            2> 线程队列
       2. 队列类型:
          q = queue.Queue(num)              # 正常队列, FIFO
          q = queue.LifoQueue(num)          # 栈, FILO
          q = queue.PriorityQueue(num)      # 优先队列
             : q.put([num,data])            # num从小到大, 优先级从高到低
               q.get()                      # 返回一个list

       3. 线程队列保证线程安全

          q.put() # 向队列中添加数据, 队列为满则会阻塞       //           q.put_nowait() # 添加, 队列为满报异常
          q.get() # 从队列中取数据, 队列为空则会阻塞         //           q.get_nowait() # 取值, 队列为空报异常

          另一种阻塞方式:
          q.task_done() <--> q.join()

       4. 其他方法
          q.qsize()   # 返回当前队列长度
          q.empty()   # 队列是否为空
          q.full()    # 队列是否为满
‘‘‘ # 多线程通信: 线程队列(queue.Queue)

‘‘‘
线程局部变量(ThreadLocal):
    1. 作用: 每个线程只能访问公共变量, ThreadLocal为每个线程创建了自己的变量
    2. 用法:
           #1. thread_local = threading.local()    // 创建全局的thread_local对象
           #2. thread_local.attr = value           // 在每个线程内, 线程可通过thread_local的属性设定变量
                                                   // thread_local的相同属性在各个线程中不一样, 互不影响
    3. 实质:
       #1. thread_local对象内部维护一个dict, 其key为不同线程的ID, value为这些线程存储值
       #2. thread_local.attr调用属性值时, 实际上是以当前线程的ID为key, 去找值
‘‘‘ # 多线程通信: 线程局部变量(threading.local)

‘‘‘
多进程使用:
     1. 导入模块 import threading
     2. 开启进程的方法:
        1) 直接开启:  
                  1> 创建Process对象
                     p = mulitprocessing.Process(target=func,args=(a,),kwargs={‘k1‘:v1})
                  2> 开启线程
                     p.start
        2) 继承开启:
                  1> 继承Process类, 重写run()方法[即要执行的函数]
                     class MyProcess(multiprocessing.Process):
                          def __init__(self):
                             super(MyThread,self).__init__()
                             # 其他初始化代码,如要传给函数值
                          def run(self): pass
                  2. 创建实例,开启显线程
                     p = MyProcess()
                     p.start()           # 会调用其run()方法

     3. 子进程和主进程的关系:
        1> 主进程和子进程分别执行, 主线程会等待最后一个子进程执行完毕后, 关闭

        2> 子进程p.join()到主进程, 会阻塞主线程, 执行完该子进程后主进程继续执行, 最后主进程等待最后一个子进程执行完毕后, 关闭

        3> 在子进程开启之前, 申明该子进程为p.daemon = True守护进程, 最后主线程不会等待该子线程, 其他进程执行完就关闭主线程

     4. 进程的属性和方法:
        1) 对象属性和方法: p.attr p.func()    
             p.func(): 
                     p.start()          # 启动进程
                     p.join()           # 阻塞当前进程
                     p.run()            # 在当前进程下调用该进程的run方法

                     p.isAlive()        # 判断进程是否被激活
                     p.terminate()      # 结束进程
             p.attr:
                     p.name                # 线程名
                     p.daemon = True/False # 是否为守护线程
                     p.num/p.pid           # 进程号

        2) os方法:
             os.getpid()                   # 获取当前进程号
             os.getppid()                  # 获取当前进程父进程号
‘‘‘  # 多进程: 使用, 见Mulit_Process.py

‘‘‘
多进程通信:
    1. 锁机制: Lock/ RLock/ Event/ Semaphore 同多线程
    2. 进程队列: multiprocessing.Queue       同多线程
    进程间资源不能共享, 因此在主线程中创建的Lock, RLock, Event, Semaphore, Queue, 要作为参数传入到子线程中

    3. 管道(Pipe): 通过管道两端的conn双向通信
           用法:  
               #1. conn1, conn2 = multiprocessing.Pipe()    //在主进程中创建
               #2. 将conn1, conn2作为参数传入进程中
               #3. 每一个conn有conn.send(), conn.recv()方法
               #4. conn.recv()为空阻塞, conn.send()为空不阻塞

    4. 共享数据(Manager): 所有进程都可访问的数据块
          用法:
              #1. manager = multiprocessing.Manager()   // 主进程中创建
              #2. 通过manager对象可创建公共的
                     list, dict                            // 也需要传入
                     Lock, RLock, Event, Semaphore, Queue  // 也需要传入
              #3. 各个进程可访问这些数据集, 注意进程不安全的(除了Queue)要加锁

    5. 进程池(Pool):  与信号量类似，但没有加锁。
               信号量是，可以开无限多的进程(线程)，但某一时刻只允许num个进程(线程)修改数据
               进程池是，从始至终只开num个进程
‘‘‘  # 多进程通信: 锁机制, 进程队列, Pipe, Manager, Pool