多线程概念+python

python 线程threading 官方文档：https://docs.python.org/zh-cn/3/library/threading.html#lock-objects

多线程理解

前言：

​ 本人是一个刚工作的小白，在python开发中使用多线程的时候，发现python的多线程，知识点比较散乱，故做了一个整理，这些思路是基于操作系统和Java做的，然后配合python的代码实现，内容中如果有不对或不合适的地方希望大家多多提意见，互相学习。

一、什么是并发与并行。

并行： A时间和B时间在同一个时间点(时刻)上运行。

并发: A和B在同一时间段同时运行

二、什么是进程、线程

进程： 你可以认为是一列火车。

1. 进程是正在运行的程序的实例。
2. 进程是线程的容器，即一个进程中可以开启多个线程。

线程： 火车上的一节车厢， 在进程中运行。

1.    线程是进程内部的一个独立执行单元
2.    一个进程可以同时并发运行多个线程

多线程：多个线程并发执行。

三、线程生命周期

1. 新建

   首先为线程分配内存， 初始化成员变量。

2. 就绪

   当线程调用start()方法后，该线程处于就绪状态

   为线程创建方法栈和程序计数器，等待线程调度器调度

3. 运行

   就绪状态的线程获得CPU资源，开始运行run()方法，改线程进入运行状态

4. 阻塞

   当发生以下情况时，线程将会进入阻塞状态

   • 线程调用sleep()方式主动放弃所占用的处理器资源
   • 线程调用了一个阻塞式IO方法，在该方法返回之前，该线程被阻塞
   • 线程视图获得一个同步锁(同步监视器)，但该同步锁正被其他线程所持有。
   • 线程在等待某个通知
   • 程序调用了线程的suspend() 方式将该线程挂起。但这个方法容易导致死锁，所以应该尽量避免使用该方法。

5. 死亡

   • 线程执行完，线程正常结束。
   • 线程抛出一个未捕获的Exception或Error。
   • 调用该线程stop()方法来结束该线程，该方法容易导致死锁，不推荐使用。

四、线程安全问题

​ 当你运行线程的时候，单线程多线程运行结果一样，且变量的值也和预期是一样的，线程就是安全；反之就是不安全的。

原因：

• 多个线程在操作共享的数据
• 操作共享数据的线程代码有多条
• 多个线程对共享数据有写操作

解决方法：

​ 线程同步：

​ 在某个线程修改共享资源的时候，其他线程不能修改该资源，等待修改完毕同 步之后，才能去抢夺CPU资源，完成对应的操作，保证数据的同步性。

• 同步代码块
• 同步方法
• 同步锁
• 特殊域变量
• 局部变量
• 阻塞队列
• 原子变量

python 多线程的线程安全问题展示。(问题原因在上述有描述) 窗口卖票

​ 我们用多个线程操作共享数据。我们

import threading
import time
count = 50
def draw():
    global count
    # 账户余额大于取钱数目
    while True:
        if count > 0:
            print(threading.current_thread().name, count)
            count -= 1
        else:
            break
t1 = threading.Thread(name="窗口一", target=draw)
t1.start()
t2 = threading.Thread(name="窗口二", target=draw)
t2.start()

面对这种情况， 我的打印结果是51条，正常来说应该是50条，其原因如下：

​ 基于多线程的情况，可能 t1 取到第49张票的时候还来的及卖出去，恰巧，这个时候 t2 也开始取值也取到了 49， 那这样的话第49张票同时被两个窗口卖掉了所以出现了线程安全问题。

另一个演示的经典问题是银行取钱问题(基于类实现的)：http://c.biancheng.net/view/2617.html

python 锁机制。

锁对象简介

• ​ python 有两个锁分别为 threading.Lock() （原始锁）和 threading.RLock() （重入锁）

• ​ 一旦一个线程获得一个锁， 会阻塞随后尝试获取锁的线程，直到他被释放

Lock()

• ​ 任何线程都可以释放这个锁

RLock()

• ​ 只有当拥有锁的线程才能释放。

共有方法：

acquire(blocking=True, timeout=-1)

​ 说简单一点： True 就是阻塞式加锁， False 就是非阻塞式加锁， 默认未True

​ 加锁（官方介绍如下）

可以阻塞或非阻塞地获得锁。

当无参数调用时： 如果这个线程已经拥有锁，递归级别增加一，并立即返回。否则，如果其他线程拥有该锁，则阻塞至该锁解锁。一旦锁被解锁(不属于任何线程)，则抢夺所有权，设置递归等级为一，并返回。如果多个线程被阻塞，等待锁被解锁，一次只有一个线程能抢到锁的所有权。在这种情况下，没有返回值。

当发起调用时将 blocking 参数设为真值，则执行与无参数调用时一样的操作，然后返回 True。

当发起调用时将 blocking 参数设为假值，则不进行阻塞。 如果一个无参数调用将要阻塞，则立即返回 False；在其他情况下，执行与无参数调用时一样的操作，然后返回 True。

当发起调用时将浮点数的 timeout 参数设为正值时，只要无法获得锁，将最多阻塞 timeout 所指定的秒数。 如果已经获得锁则返回 True，如果超时则返回假值。

release()

​ 释放锁（官方介绍如下）

释放锁，自减递归等级。如果减到零，则将锁重置为非锁定状态(不被任何线程拥有)，并且，如果其他线程正被阻塞着等待锁被解锁，则仅允许其中一个线程继续。如果自减后，递归等级仍然不是零，则锁保持锁定，仍由调用线程拥有。

只有当前线程拥有锁才能调用这个方法。如果锁被释放后调用这个方法，会引起 RuntimeError 异常。

代码实现：

import threading
import time

count = 100
lock = threading.RLock() 
def draw():
    # 加锁
    global count
    while True:
        lock.acquire(True)
        try:
            if count > 0:
                print(threading.current_thread().name, count)
                count -= 1
            else:
                break
        finally:
            lock.release()

t1 = threading.Thread(name="窗口一", target=draw)
t1.start()
t2 = threading.Thread(name="窗口二", target=draw)
t2.start()

这样的话没个线程开始执行的时候,会进行加锁阻塞，直到锁结束。

注意：

• 这个锁应该是全局的，处理同一个逻辑的多线程应该公用一个锁
• 用try finally 来执行解锁， 中间放的应该是你的逻辑代码
• 你可以调整你的加锁和解锁位置实现独占锁

五、线程死锁

1. 什么是死锁

   多个线程因为争夺资源而造成的一种僵局(互相等待)，若无外力作用。这些进程都将无法推进。

2. 死锁条件

   • 互斥条件：在一段时间内某资源仅为一个进程所占有。此时若有其他进程请求资源，请求进程只能等待。

   • 不可剥夺条件：进程所获得的资源在未使用完毕之前，不能被其他资源强行夺走，即只能由获得该资源的进程自己来释放(主动释放)

   • 请求与保持条件：该进程已经保持了至少一个资源，但又提出了新的资源请求，而该资源已被其他进程占有，此时请求进程被阻塞，但对自己获取的资源保持不放。

   • 循环等待条件：一个等一个刚好构成一个循环， 循环等待不一定死锁，死锁一定循环等待。

简单演示：

实例1：

• a 为资源
• 每一个线程在调用的时候都会将其起来，

import time
import threading

a = 100

lock = threading.Lock()

def fun1():
    global a
    while True:
        
        if not lock.locked():
            lock.acquire(True)
            try:
                print(threading.current_thread().name, a)
                time.sleep(10)
            finally:
                lock.release()
                break
        else:
            print("被加锁")
            time.sleep(3)
            

if __name__ == "__main__":
    t1 = threading.Thread(name="t1", target=fun1)
    t1.start()
    t2 = threading.Thread(name="t2", target=fun1)
    t2.start()

结果：

t1 100
被加锁
被加锁
被加锁
被加锁
t2 100

我们可以看到 t2 一直等到 t1 释放资源后才会执行，那假如不释放呢，是不是就一直等待，就进入了死锁。

示例2：

import time
import threading

a = 100
b = 50

lock_a = threading.Lock()
lock_b = threading.Lock()

def fun1():
    global a
    while True:
        if not lock_a.locked():
            lock_a.acquire(True)
            try:
                print(threading.current_thread().name, a)
                time.sleep(10)
                while True:
                    if not lock_b.locked():
                        lock_b.acquire(True)
                        print(threading.current_thread().name, b)
            finally:
                lock_a.release()
                lock_b.release()
                break


def fun2():
    global a
    while True:
        if not lock_b.locked():
            lock_b.acquire(True)
            try:
                print(threading.current_thread().name, b)
                time.sleep(10)
                while True:
                    if not lock_a.locked():
                        lock_a.acquire(True)
                        print(threading.current_thread().name, a)
            finally:
                lock_a.release()
                lock_b.release()
                break


if __name__ == "__main__":
    t1 = threading.Thread(name="t1", target=fun1)
    t1.start()
    t2 = threading.Thread(name="t2", target=fun2)
    t2.start()

结果：

t1 100
t2 50

这个时候 fun1 就说我有a，我还需要b， fun2 说我有 b 我还需要 a， 这两就杠到这了谁也不想松手，就造成了死锁。堵死了

3. 死锁处理

   • 预防死锁： 通过设置某些限制条件，去破坏死锁的四个必要条件中的一个或几个条件，来防止死锁的发生
   • 避免死锁： 在资源的动态分配过程中，用某种方法去防止系统进入不安全状态，从而避免死锁的发生。
   • 检测死锁： 允许系统在运行过程中发生死锁，但可以设置检测机构即时检测死锁的发生，并采取适当措施加以清除。
   • 解除死锁：当检测出死锁后，便采取适当措施将进程从死锁状态中解脱出来。

3.1. 死锁预防

3.1.1 破坏"互斥"条件

​ "互斥"条件是无法破坏的（例如：打印机的"互斥" 是无法改变的）。因为死锁预 防中一般不去涉及"互斥"条件。

3.1.2 破坏"占有并等待" 条件

​ 不允许在以获得某种资源的情况下，申请其他资源。

​ 方法一： 一次性分配资源，即创建线程时，一次申请所需的全部资源，系统或满足其所有要求，或什么也不给他。（要提前知道，线程需要的所有 资源）

​ 方法二: 要求每个进程提出新的资源申请前，释放它所占有的资源。（每隔资源之前必须互不依赖）

3.1.3 破坏"不可抢占"条件

​ 允许对资源实行抢夺。

​ 方法一: 如果占有某些资源的一个进程进行一步资源请求被拒绝，则该进程必须释放它最初占有的资源，如果有必要，可再次请求这些资源和另外的资 源。

​ 方法二: 如果一个进程请求当前被另一个进程占有的一个资源，则操作系统可以抢占另一个进程，要求它释放资源。只有在任意两个进程的优先级都 不相同的条件下，方法二才能预防死锁

3.1.4 破坏"循环等待"条件

​ 将系统中的所有资源同一编号，进程可在任何时刻提出资源申请，但所有申请必须按照资源的编号顺序(升序)提出。

3.2 死锁避免

​ 避免死锁不严格限制产生死锁的必要条件的存在，因为即使死锁的必要条件存在，也不一定发生死锁。

3.2.1 有序资源分配法

• 为所有资源统一编号，例如：打印机为1、传真机为2、磁盘为3

• 同类资源必须一次性申请完， 例如： 打印机和传真机一般为同一个机器，必须同时申请。

• 不同类资源必须按顺序(编号顺序)申请。

  例如：有两个进程p1 和 p2, 有两个资源 r1 和 r2.

  p1请求资源： r1、r2

  p2请求资源: r1、r2

  这样就破坏了环路条件,避免了死锁的发生。

3.2.2 银行家算法

3.2.3 顺序加锁

​ 当多个线程需要相同的一些锁，但是按照不同的顺序加锁，死锁就很容易发生。

​

3.2.4 限时加锁

​ 在线程尝试获取资源的时候超过了时间，那么放弃对该锁请求，并且释放掉自己所拥有的所有资源，在等待一段随机的时间后在重新申请。

缺点：

​ 1）当线程数量少的时候，这种方式可避免，但是当线程数量特别多的时候，加锁时限相同的概率会提高，依旧有可能进入死锁。

3.3 死锁检测

​ 预防死锁和死锁避免的系统开销大且不能充分利用资源，更好的方法时不采取任何限制性措施，而是提供检测和解脱死锁的手段。

3.4 死锁恢复

​ 1) 利用抢占恢复：

​ 临时将某个资源从当前所属进程转移到另一个进程。(需要人工干预)

​ 2) 利用回滚恢复

​ 周期性的将进程的状态进行备份，当发现进程死锁后，根据备份将该进程复位到一个更早的状态，还没有取得所需资源的状态，接着就将这些资源分配给其他死锁进程。

​ 3)通过杀死进程

​ 就是杀死一个或若干个进程。（尽可能保证杀死的进程/线程，可以重新启动，不带来副作用。）

六、线程通讯

6.1 为什么要通讯

​ 多个线程并发执行时，在默认情况下CPU时随机切换线程的，有时我们希望CPU按我们的规律执行线程，此时就需要线程之间协调通讯。

6.2 线程通讯方式

​##### 采用Event:

• is_set()

  当且仅当内部标识为 true 时返回 True 。

• set()

  将内部标识设置为 true 。所有正在等待这个事件的线程将被唤醒。当标识为 true 时，调用 wait() 方法的线程不会被被阻塞。

• clear()

  将内部标识设置为 false 。之后调用 wait() 方法的线程将会被阻塞，直到调用 set() 方法将内部标识再次设置为 true 。

• wait(timeout=None)

  阻塞线程直到内部变量为 true 。如果调用时内部标识为 true，将立即返回。否则将阻塞线程，直到调用 set() 方法将标识设置为 true 或者发生可选的超时。当提供了timeout参数且不是 None 时，它应该是一个浮点数，代表操作的超时时间，以秒为单位（可以为小数）。当且仅当内部旗标在等待调用之前或者等待开始之后被设为真值时此方法将返回 True，也就是说，它将总是返回 True 除非设定了超时且操作发生了超时。

  示例：当线程1跑完，再跑线程2， 当然这个例子也可以通过join() 来实现

  import threading
  
  event = threading.Event()
  
  def fun1():
      count = 10
      event.clear()
      try:
          while count > 0:
              print(threading.current_thread().name, "t11",  count)
              count -= 1
      finally:
          event.set()
  
  def fun2():
      count = 10
      while count < 20:
          event.wait()
          print(threading.current_thread().name,  count)
          count += 1
  
  t1 = threading.Thread(target=fun1, name="t1")
  t1.start()
  t2 = threading.Thread(target=fun2, name="t2")
  t2.start()
  

采用条件对象 Condition

• threading.Condition(lock=None)

​ 如果给出了非 None 的 lock 参数，则它必须为 Lock 或者 RLock 对象，并且它将被用作底层锁。否则，将会创建新的 RLock 对象，并将其用作底层锁

• acquire(**args*)

  请求底层锁。此方法调用底层锁的相应方法，返回值是底层锁相应方法的返回值。

• release()

  释放底层锁。此方法调用底层锁的相应方法。没有返回值。

• wait(timeout=None)

  进入阻塞，直到被notify() 唤醒或者 notify_all() 唤醒， 或者设置了timeout等待超时。

  wait之前必须获得锁

• wait_for(predicate, timeout=None)

  等待，直到条件计算为真。 predicate 应该是一个可调用对象而且它的返回值可被解释为一个布尔值。可以提供 timeout 参数给出最大等待时间。这个实用方法会重复地调用 wait() 直到满足判断式或者发生超时。返回值是判断式最后一个返回值，而且如果方法发生超时会返回 False 。忽略超时功能，调用此方法大致相当于编写:while not predicate(): cv.wait() 因此，规则同样适用于 wait() ：锁必须在被调用时保持获取，并在返回时重新获取。 随着锁定执行判断式。3.2 新版功能.

  （简单说就是等待 predicate 为证，predicate 不是表达式，而是可调用的）

  def wait_for(self, predicate, timeout=None):
          """Wait until a condition evaluates to True.
  
          predicate should be a callable which result will be interpreted as a
          boolean value.  A timeout may be provided giving the maximum time to
          wait.
  
          """
          endtime = None
          waittime = timeout
          result = predicate() # 调用
          while not result:
              if waittime is not None:
                  if endtime is None:
                      endtime = _time() + waittime
                  else:
                      waittime = endtime - _time()
                      if waittime <= 0:
                          break
              self.wait(waittime)
              result = predicate() # 调用
          return result
  

• notify(n=1)

  默认唤醒一个等待这个条件的线程。如果调用线程在没有获得锁的情况下调用这个方法，会引发 RuntimeError 异常。这个方法唤醒最多 n 个正在等待这个条件变量的线程；如果没有线程在等待，这是一个空操作。当前实现中，如果至少有 n 个线程正在等待，准确唤醒 n 个线程。但是依赖这个行为并不安全。未来，优化的实现有时会唤醒超过 n 个线程。注意：被唤醒的线程实际上不会返回它调用的 wait() ，直到它可以重新获得锁。因为 notify() 不会释放锁，只有它的调用者应该这样做。

• notify_all()

  唤醒所有正在等待这个条件的线程。这个方法行为与 notify() 相似，但并不只唤醒单一线程，而是唤醒所有等待线程。如果调用线程在调用这个方法时没有获得锁，会引发 RuntimeError 异常。

信号量对象Semaphore(value)

​ value是内部计数器的值

​ 调用acquire() 计数器减一

​ 调用acquire() 计数器加一

• acquire(blocking=True, timeout=None)

​ 在不带参数的情况下：

​ value 为0 阻塞，直到调用release（）

​ value > 0 继续执行

  当发起调用时将 *blocking* 设为假值（False），则不进行阻塞。 如果一个无参数调用将要阻塞，则立即返回 False；在其他情况下，执行与无参数调用时一样的操作，然后返回 True。

  当发起调用时如果 *timeout* 不为 None，则它将阻塞最多 *timeout* 秒。 请求在此时段时未能成功完成获取则将返回 False。 在其他情况下返回 True。

• release(n=1)
  释放一个信号量，将内部计数器的值增加 n。 当进入时值为零且有其他线程正在等待它再次变为大于零时，则唤醒那 n 个线程

有几个例子：

这是一个Leetcode 的题，我用了这三个方法解了一下，大家可以看下，有问题可以指出来。

https://leetcode-cn.com/problems/print-foobar-alternately/solution/jiao-ti-da-yin-xin-hao-liang-dui-xiang-b-9noe/

打印奇数偶数

https://leetcode-cn.com/problems/print-zero-even-odd/solution/da-yin-ji-shu-ou-shu-cai-yong-tiao-jian-ot91s/

7. 多线程特性

​ 原子性、可见性、有序性

7.1原子性

​ 一个操作或者多个操作，要么全部执行，且执行过程中不会被打断，要么不执行。

7.2可见性

​ 当多个线程访问同一变量时，一个线程修改了这个变量的值，其他线程能够立即看到修改的值。显然，对于单线程来说，可见性问题是不存在的。

7.3有序性

​ 程序的执行顺序按照代码的先后顺序执行。