2-线程操作

在Python程序中，可以通过“_thread”和“threading（推荐使用）”这两个模块来处理线程。在Python 3程序中，thread模块已废弃。可以使用 threading 模块代替。

2-1-threading模块

2-1-1-介绍

Python的threading模块提供了一个高级接口来处理线程。它基于底层的_thread模块，但提供了更强大和灵活的功能。使用threading模块可以让你在同一程序中同时运行多个任务，从而实现并发操作。

2-1-2-常用方法和属性

2-1-2-1-Thread类

Thread是创建线程的核心类

2-1-2-2-创建线程

Thread(target=callable, args=(arg1,arg2,...)):

创建一个新线程，其中`target`是线程启动时要调用的函数，`args`是传递给该函数的参数元组。

2-1-2-3-线程的属性和方法

2-1-3-示例代码

下面是一个简单的例子，演示如何使用`threading.Thread`来创建并启动两个线程，每个线程都运行一个打印数字的函数：

import threading
import time

def show_num():
    '打印0-4的数字'
    for i in range(0,5):
        print(f'打印数字:{i}')
        time.sleep(1)

def show_upper_word():
    for i in ('A','B','C','D','E'):
        print(f'打印字母:{i}')
        time.sleep(1)


thread_show_num = threading.Thread(show_num())
thread_show_upper_word = threading.Thread(show_upper_word())

thread_show_num.start()
thread_show_upper_word.start()


thread_show_num.join()
thread_show_upper_word.join()

print("所有线程已完成")

2-1-4-输出结果

1. 在这个例子中，我们定义了两个函数`print_numbers`和`print_letters`，它们分别打印数字和字母，并且在每次打印后都会暂停一秒（模拟耗时操作）。
2. 然后，我们创建了两个线程实例`number_thread`和`letter_thread`，分别对应于这两个函数，并通过调用`start()`方法启动它们。
3. 最后，我们使用`join()`方法确保主线程会等待这两个线程完成后再继续执行后续代码。

2-2-创建线程

2-2-1-创建线程的方式

1. 通过threading.Thread直接在线程中运行函数；
2. 通过继承类threading.Thread来创建线程。

2-2-2-语法01

在Python程序中，使用threading.Thread的基本语法格式如下所示。

Thread (group=None, target=None, name=None, args=(),kwargs={},*,daemon=None)

其中，参数“target”表示要运行的函数，参数“args”表示传入函数的参数元组。

2-2-2-1-例子

import threading #导入库threading 

def zhiyun(x,y): #定义函数zhiyun() 
 for i in range(x,y): #遍历操作 
     print(str(i*i)+';') #输出一个数的平方 

ta = threading.Thread(target=zhiyun,args=(1,6)) 
tb = threading.Thread(target=zhiyun, args=(16,21)) 

ta.start() #启动第1个线程活动 
tb.start() #启动第2个线程活动

2-2-2-2-输出结果

2-2-2-3-代码分析

在上述实例代码中，首先定义函数zhiyun()，然后以线程方式来运行这个函数，并且在每次运行

时传递不同的参数。运行后两个子线程会并行执行，可以分别计算出一个数的平方并输出，这两个子线程是交替运行的。

2-2-3-语法02

继承类threading.Thread创建通过继承类threading.Thread的方式来创建一个线程。这种方法只要重载类threading.Thread中的方法run()，然后再调用方法start()就能够创建线程，并运行方法run()中的代码

2-2-3-1-例子

import threading

class myThread(threading.Thread):
  #定义继承于类threading.Thread的子类myThread

  def __init__ (self,mynum): #构造函数
     super().__init__ ()
     #使用super()处理子类和父类关系
     self.mynum = mynum

  def run(self): #定义函数run()
     for i in range(self.mynum,self.mynum+5):
         print(str(i*i)+';')

ma = myThread(1)  #创建类myThread的对象实例ma
mb = myThread(16) #创建类myThread的对象实例mb

ma.start() #启动线程
mb.start() #启动线程

2-2-3-2-输出结果

2-2-3-3-代码分析

上例代码中，首先定义了一个继承于类threading.Thread的子类myThread，

然后创建了两个类myThread的实例，

并使用方法myThread()和start()分别创建线程和启动线程功能。

2-3-线程等待

2-3-1-概念

让主线程暂停，一直等到调用该方法的线程执行完毕。

Python提供了`time.sleep()`函数和`Thread.join()`方法。

2-3-2-使用time.sleep()函数

2-3-2-1-作用

能够让当前线程暂停执行指定的秒数。

适合需要线程暂停一段时间的场景，像定时任务

2-3-2-2-例子

import time
import threading

def worker():
    print("Worker thread started")
    time.sleep(2)  # 线程暂停 2 秒
    print("Worker thread finished")

# 创建并启动线程
thread = threading.Thread(target=worker)
thread.start()

print("Main thread continues while worker thread is sleeping")

2-3-2-3-输出结果

2-3-3-使用Thread.join()方法

2-3-3-1-作用

Thread.join()方法可使主线程等待子线程执行完毕。

适用于需要确保某个线程执行完毕后再继续执行后续代码的场景。

2-3-3-2-例子1

2-3-3-2-1-代码

import threading

def worker():
    print("Worker thread started")
    # 模拟一些工作
    for i in range(5):
        print(f"Worker thread is working: {i}")
    print("Worker thread finished")

# 创建并启动线程
thread = threading.Thread(target=worker)
thread.start()

# 主线程等待子线程完成
thread.join()

print("Main thread continues after worker thread has finished")

2-3-3-2-2-输出结果

2-3-3-3-例子2

2-3-3-3-1-代码

import threading
import time

def worker(x,y,thr=None):
     #当在函数worker()的参数中包括一个线程实例时,也就是thread02启动的时候会进入这个代码块
     if thr:
         thr.join() #调用方法join()
     else:
         time.sleep(2) #睡眠2s
         for i in range(x,y): #遍历参数x和y
             print(str(i*i)+';') #输出i的平方

thread01 = threading.Thread(target=worker,args=(1,6))

# thread02传入了线程实例a，所以thread02线程应等待thread01结束后才运行。
thread02 = threading.Thread(target=worker,args=(16,21,thread01))

# 运行后会发现，线程thread02等到线程thread01输出结束后才输出结果。
thread01.start()  #启动线程
thread02.start()  #启动线程

2-3-3-3-3-输出结果

2-3-3-3-4-代码分析

当在函数worker()的参数中包括一个线程实例时,也就是thread02启动的时候会进入 thr.join()这个代码块

这样thread02要等到thread01执行完毕才开始执行

2-4-线程同步

2-4-1-概念

1. 在多线程编程里，线程同步是指对多个线程对共享资源的访问进行协调的机制。
2. 由于多个线程可能会同时访问和修改共享资源，若不加以协调，就会出现数据不一致、竞态条件等问题。
3. 线程同步的目的在于保证在同一时刻，仅有一个或特定数量的线程能够访问共享资源，从而保证数据的一致性和完整性。
4. 同步原语:是用来控制多个线程或者进程访问共享资源的工具，目的是防止数据竞争和不一致问题,以下Lock、RLock、Semaphore、Event和Condition都成为同步原语

2-4-2-实现方法

1. 锁（Lock）：最基本的同步原语，同一时刻只允许一个线程访问共享资源。
2. 递归锁（RLock）：允许同一个线程多次获取同一把锁，避免死锁。
3. 信号量（Semaphore）：允许指定数量的线程同时访问共享资源。
4. 事件（Event）：用于线程间的简单通信，一个线程可以等待另一个线程设置的事件。
5. 条件变量（Condition）：允许线程在满足特定条件时才访问共享资源。

2-4-3-Lock锁

threading.Lock 是用于线程同步的基础工具，它可以保证同一时刻只有一个线程能访问共享资源，从而避免数据竞争问题。

1-Lock()

功能

创建一个新的锁对象。

示例代码

import threading lock = threading.Lock()

2-acquire(blocking=True, timeout=-1)

功能

尝试获取锁。

若锁处于未锁定状态，当前线程将获取锁并将其锁定；

若锁已被其他线程锁定，根据 `blocking` 和 `timeout` 参数决定线程的行为。

参数

1. blocking：布尔值，默认为 `True`。若为 `True`，线程会阻塞直到获取到锁；若为 `False`，线程不会阻塞，会立即返回结果。
2. timeout：浮点数，默认为 `-1`。表示等待获取锁的最长时间（秒）。若为 `-1`，则无限期等待。
3. acquire(blocking=True, timeout=-1) 等价于 acquire()

返回值

若成功获取锁，返回 `True`；若因 `blocking=False` 或超时未能获取锁，返回 `False`。

示例代码

import threading
import time

lock = threading.Lock()

def operator():
    # 尝试获取锁，阻塞直到获取到锁
    lock.acquire()
    try:
        print("operator has acquired the lock.")
        time.sleep(2)
    finally:
        # 释放锁
        lock.release()

thread = threading.Thread(target=operator)
thread.start()

输出结果

3-release()

功能

释放锁。此操作会将锁的状态从锁定变为未锁定，从而允许其他线程获取该锁。

必须在持有锁的线程中调用该方法，若在未持有锁的线程中调用会引发 `RuntimeError`。

示例代码

import threading

lock = threading.Lock()

def operator():
    lock.acquire()
    try:
        print("operator is working...")
    except Exception as e:
        print(f'异常:{e}')
    finally:
        lock.release()
        print("operator has released the lock.")

thread = threading.Thread(target=operator)
thread.start()

输出结果

4-locked()

功能

检查锁是否处于锁定状态。

若锁已被锁定，返回 `True`；若未被锁定，返回 `False`。

示例代码

import threading

lock = threading.Lock()

def operator():
    lock.acquire()
    try:
        print("operator is working...")
        print(f"now lock status :{lock.locked()}")
    except Exception as e:
        print(f'异常:{e}')
    finally:
        lock.release()
        print("operator has released the lock.")
        print(f"now lock status :{lock.locked()}")

thread = threading.Thread(target=operator)
thread.start()

输出结果

5-Lock锁的实现原理

在 Python 中，Lock 是基于操作系统提供的底层同步原语实现的，通常使用 mutex（互斥量）。mutex 是一种操作系统级别的同步机制，能保证同一时间只有一个线程可以访问共享资源。

以下是 Lock 的基本操作流程：

初始化：

创建 Lock 对象时，会在底层初始化一个 mutex。

获取锁（acquire 方法）：

当线程调用 Lock 的 acquire 方法时，会尝试获取底层的 mutex。若 mutex 处于未锁定状态，线程会将其锁定并继续执行后续代码；若 mutex 已经被其他线程锁定，当前线程会被阻塞，进入等待状态。

释放锁（release 方法）：

当线程调用 Lock 的 release 方法时，会将底层的 mutex 解锁，唤醒等待该锁的其他线程。

2-4-4-RLock递归锁

1-什么是RLock

RLock 是可重入的，同一线程可以多次获取同一把 RLock 而不会被阻塞。

每次获取锁时，锁的内部计数器会加 1；每次释放锁时，计数器会减 1。

只有当计数器为 0 时，锁才会真正被释放。

2-语法

2-1-基本使用

import threading

# 创建一个 RLock 对象
rlock = threading.RLock()

# 共享资源
shared_resource = 0

def recursive_increment(level):
    global shared_resource
    # 获取锁
    rlock.acquire()
    try:
        # 增加共享资源的值
        shared_resource += 1
        print(f"Level {level}: Shared resource is {shared_resource}")
        if level > 0:
            # 递归调用
            recursive_increment(level - 1)
    finally:
        # 释放锁
        rlock.release()

# 创建线程
thread = threading.Thread(target=recursive_increment, args=(3,))
# 启动线程
thread.start()
# 等待线程执行完毕
thread.join()

2-2-使用with语句

使用with语句代替了lock.acquire()和lock.release()

import threading

rlock = threading.RLock()
shared_variable = 0

def worker():
    global shared_variable
    with rlock:
        for _ in range(1000):
            shared_variable += 1
    print(f"Worker finished. Shared variable: {shared_variable}")

threads = []
for _ in range(3):
    t = threading.Thread(target=worker)
    threads.append(t)
    t.start()

for t in threads:
    t.join()

3-RLock的底层实现原理

RLock 的底层实现依赖于 Lock 和一个计数器。具体操作流程如下：

初始化

创建 RLock 对象时，会初始化一个 Lock 和一个计数器，计数器初始值为 0。

获取锁（acquire 方法）

当线程调用 RLock 的 acquire 方法时，会进行以下检查：

若当前线程已经持有该 RLock，则将计数器加 1，并立即返回。

若当前线程未持有该 RLock，则尝试获取底层的 Lock。若获取成功，将计数器加 1，并记录当前持有锁的线程 ID；若获取失败，线程会被阻塞。

释放锁（release 方法）

当线程调用 RLock 的 release 方法时，会进行以下操作：

检查当前线程是否持有该 RLock，若不持有则抛出异常。

若持有，则将计数器减 1。若计数器为 0，则释放底层的 Lock，并清除当前持有锁的线程 ID。

4-RLock和Lock的区别

在 Python 的多线程编程中，`Lock` 和 `RLock` 都是用于线程同步的工具，它们都能防止多个线程同时访问共享资源，避免数据竞争和不一致问题，但两者存在明显区别，适用场景也有所不同。

4-1-可重入性

1. Lock：不可重入,当一个线程已经获取了 `Lock` 后，若该线程再次尝试获取这个锁，会被阻塞，从而可能导致死锁。
2. RLock：可重入,同一线程可以多次获取同一把 `RLock` 而不会被阻塞。每次获取锁时，锁的内部计数器会加 1；每次释放锁时，计数器会减 1。只有当计数器为 0 时，锁才会真正被释放。

4-2-释放规则

1. Lock 只需释放一次，锁就会被释放，其他线程便能获取该锁。也就是说，`acquire()` 和 `release()` 操作需一一对应。
2. RLock 每次获取锁后都要进行一次释放操作，只有当获取和释放的次数相等（即内部计数器为 0）时，锁才会真正被释放。

4-3-性能开销

1. Lock 由于实现相对简单，不涉及内部计数器的维护，所以性能开销相对较小。
2. RLock 需要维护一个内部计数器，以记录锁的获取和释放次数，因此相比 `Lock`，它的性能开销会略高一些。

5-RLock和Lock的使用场景

5-1-Lock 的使用场景

简单互斥访问

当多个线程需要对同一个共享资源进行互斥访问，且不存在递归调用或嵌套方法调用需要多次获取锁的情况时，使用 Lock即可。

例如，多个线程对一个全局变量进行修改操作，使用 `Lock` 可以确保同一时间只有一个线程能修改该变量。

import threading

shared_variable = 0
lock = threading.Lock()

def increment():
    global shared_variable
    lock.acquire()
    try:
        for _ in range(100000):
            shared_variable += 1
    finally:
        lock.release()

threads = []
for _ in range(5):
    t = threading.Thread(target=increment)
    threads.append(t)
    t.start()

for t in threads:
    t.join()

print(shared_variable)

对性能要求较高

在一些对性能要求较高的场景中，由于 `Lock` 性能开销小，优先选择 `Lock`。

5-2-RLock的使用场景

递归调用或嵌套方法调用

当代码存在递归调用或者嵌套方法调用，并且这些调用中需要多次获取同一把锁时，必须使用 `RLock` 以避免死锁。

import threading

rlock = threading.RLock()

def recursive_function(level):
    rlock.acquire()
    try:
        print(f"Acquired RLock at level {level}")
        if level > 0:
            recursive_function(level - 1)
        print(f"Releasing RLock at level {level}")
    finally:
        rlock.release()

thread = threading.Thread(target=recursive_function, args=(3,))
thread.start()
thread.join()

代码逻辑复杂

在复杂的多线程代码中，可能会出现同一线程多次获取锁的情况，使用 `RLock` 可以保证代码的正确性。

综上所述，在满足需求的前提下，优先选择简单高效的 Lock；当需要可重入性时再使用 RLock。

2-4-5-Semaphore信号量

1. Semaphore是 threading模块中的一个同步原语，其作用是限制同时访问特定资源的线程数量。
2. 可以把它看作是一个计数器，该计数器会记录当前可用的资源数量。
3. 当一个线程想要访问资源时，它需要先获取信号量，此时信号量的计数器会减 1。
4. 当线程释放资源时，信号量的计数器会加 1。
5. 要是计数器为 0，那么线程就得等待，直到有其他线程释放资源。

2-4-5-1-Semaphore的基本使用

1-导入模块

你得先导入threading模块。

import threading

2-创建信号量对象

创建 Semaphore对象时，要指定初始的计数器值，这个值代表了同时可以访问资源的线程数量。

# 创建一个信号量对象，允许同时有 2 个线程访问资源
semaphore = threading.Semaphore(2)

3-获取和释放信号量

线程在访问资源之前，要调用 acquire()方法来获取信号量；访问结束后，调用 `release()` 方法来释放信号量。

# 获取信号量
semaphore.acquire()
try:
    # 访问共享资源
    print("Accessing shared resource")
finally:
    # 释放信号量
    semaphore.release()

2-4-5-2-完整示例

代码

import threading
import time

# 创建一个信号量对象，允许同时有 2 个线程访问资源
semaphore = threading.Semaphore(2)

def worker(id):
    print(f"Thread {id} is waiting to access the shared resource.")
    # 获取信号量
    semaphore.acquire()
    try:
        print(f"Thread {id} is accessing the shared resource.")
        # 模拟访问资源的耗时操作
        time.sleep(2)
    finally:
        # 释放信号量
        semaphore.release()
        print(f"Thread {id} has released the shared resource.")


# 创建 5 个线程
threads = []
for i in range(5):
    t = threading.Thread(target=worker, args=(i,))
    threads.append(t)
    t.start()

# 等待所有线程完成
for t in threads:
    t.join()

print("All threads have finished.")

输出结果

代码解释

1. 首先创建了一个初始值为 2 的Semaphore对象，这意味着同时最多只能有 2 个线程访问共享资源。
2. 每个线程在访问共享资源之前都会调用 acquire()方法来获取信号量。
3. 若信号量的计数器大于 0，线程就能获取到信号量，计数器减 1；若计数器为 0，线程就得等待。
4. 线程访问完共享资源后，会调用 release()方法来释放信号量，计数器加 1。
5. 借助 Semaphore可以有效控制同时访问共享资源的线程数量，防止资源过度使用。

2-4-6-Event事件

2-4-6-1-概念

1. Event 是 threading模块提供的一个同步原语，它能让一个线程等待另一个线程，直到另一个线程通知它。
2. Event对象有一个内部标志，可通过 set()和clear()方法来设置或清除。
3. 当标志被设置时，调用wait()方法的线程会被唤醒；
4. 当标志被清除时，wait()方法会阻塞线程。

2-4-6-2-主要方法

1. wait(timeout=None)：阻塞线程，直至内部标志为 True，或者达到指定的超时时间。
2. clear()：将内部标志重置为 False。
3. set()：把内部标志设置为 True，唤醒所有等待的线程。
4. is_set()：检查内部标志是否为 True。

2-4-6-3-示例

代码

import threading
import time

# 创建一个 Event 对象
event = threading.Event()

def waiter():
    print("等待事件被设置...")
    # 等待事件被设置
    event.wait()
    print("事件已被设置，继续执行。")

def setter():
    print("等待 3 秒后设置事件...")
    time.sleep(3)
    # 设置事件
    event.set()
    print("事件已设置。")

# 创建并启动线程
t1 = threading.Thread(target=waiter)
t2 = threading.Thread(target=setter)

t1.start()
t2.start()

# 等待线程完成
t1.join()
t2.join()

print("程序结束。")

输出结果

代码解释

1. waiter函数：调用 event.wait() 方法，使线程进入阻塞状态，等待事件被设置。
2. setter 函数：等待 3 秒后，调用 event.set() 方法来设置事件，唤醒所有等待的线程。
3. 主程序：创建两个线程，分别执行 waiter 和 setter函数，然后等待这两个线程执行完毕。

2-4-6-4-实现原理

threading.Event对象是基于底层的操作系统线程同步机制实现的，其具体实现细节会因不同的操作系统和 Python 解释器而有所差异。下面以 CPython（Python 的标准解释器）为例，为你大致介绍其底层实现原理。

1-基本数据结构

Event内部维护了一个布尔类型的标志（flag），用于表示事件的状态。当该标志为 True 时，表示事件已被设置；当标志为 False时，表示事件未被设置。

2-锁机制

为了保证对标志的并发访问是线程安全的，hreading.Event使用了一个锁（通常是 threading.Lock或类似的同步原语）。在修改标志状态（如调用 set()和 clear()方法）以及检查标志状态（如调用 is_set()方法）时，会先获取这个锁，操作完成后再释放锁。

3-条件变量（Condition Variable）

Event还使用了条件变量（threading.Condition）来实现线程的等待和唤醒机制。条件变量是一种高级的同步原语，它允许线程在某个条件不满足时等待，当条件满足时被唤醒。

4-主要方法的实现逻辑

2-4-7-条件变量

threading.Condition是用于线程同步的条件变量，它结合了锁和条件等待机制，能让线程在某个条件满足时才继续执行。

2-4-7-1-原理概述

条件变量内部有一个锁（默认为 threading.Lock），并且维护着一个等待线程的列表。

线程可以在条件不满足时调用wait() 方法进入等待状态，当其他线程改变了条件并且调用notify()或者notify_all() 方法时，等待的线程会被唤醒。

2-4-7-2-常用方法

2-4-7-3-示例代码

下面是一个使用 threading.Condition实现生产者 - 消费者模型的例子：

2-4-7-3-1-代码

import threading
import time
import random

# 创建一个条件变量
condition = threading.Condition()
# 模拟共享资源
queue = []
# 队列最大容量
MAX_ITEMS = 5

# 生产者线程函数
def producer():
    global queue
    while True:
        with condition:
            # 当队列满时，生产者线程等待
            while len(queue) == MAX_ITEMS:
                print("队列已满，生产者等待...")
                condition.wait()
            # 生产一个随机数
            item = random.randint(1, 100)
            queue.append(item)
            print(f"生产者生产了 {item}，当前队列: {queue}")
            # 通知消费者线程有新的物品
            condition.notify()
        # 模拟生产时间
        time.sleep(random.random())

# 消费者线程函数
def consumer():
    global queue
    while True:
        with condition:
            # 当队列为空时，消费者线程等待
            while len(queue) == 0:
                print("队列为空，消费者等待...")
                condition.wait()
            # 消费队列中的一个物品
            item = queue.pop(0)
            print(f"消费者消费了 {item}，当前队列: {queue}")
            # 通知生产者线程队列有空间了
            condition.notify()
        # 模拟消费时间
        time.sleep(random.random())

# 创建生产者和消费者线程
producer_thread = threading.Thread(target=producer)
consumer_thread = threading.Thread(target=consumer)

# 启动线程
producer_thread.start()
consumer_thread.start()

# 等待线程结束（这里不会结束，因为是无限循环）
producer_thread.join()
consumer_thread.join()

2-4-7-3-2-输出结果

2-4-7-3-3-代码解释

1-producer 函数

当队列已满时，生产者线程会调用 condition.wait()进入等待状态。当生产了一个新物品后，调用 condition.notify()唤醒可能正在等待的消费者线程。

2-consumer 函数

当队列为空时，消费者线程会调用condition.wait()进入等待状态。当消费了一个物品后，调用condition.notify()唤醒可能正在等待的生产者线程。