Python 基于线程的并行 ⭐️

Python 线程模块

Python 通过标准库的 threading 模块来管理线程。这个模块提供了很多不错的特性，让线程变得无比简单。实际上，线程模块提供了几种同时运行的机制，实现起来非常简单。

线程模块的主要组件如下：

线程对象
Lock 对象
RLock 对象
信号对象
条件对象
事件对象

创建一个线程

Python 多线程 ⭐️

Python 线程属性

threading.currentThread()：获得当前线程对象
threading.currentThread().getName()：获取当前线程对象的名字

自定义 Python 线程

使用 threading 模块实现一个新的线程，需要下面 3 步：

定义一个 Thread 类的子类
重写 __init__(self [,args]) 方法，可以添加额外的参数
最后，需要重写 run(self, [,args]) 方法来实现线程要做的事情

当你创建了新的 Thread 子类的时候，你可以实例化这个类，调用 start() 方法来启动它。线程启动之后将会执行 run() 方法。

使用 Lock 进行线程同步

在 Python 中，锁是一种同步机制，用于防止多个线程同时访问或修改同一块内存区域，以此避免数据混乱或不一致，这种现象通常被称为竞态条件。

import threading
lock = threading.Lock()
lock.acquire()      # 阻塞中，直到获得锁
# ...执行一些操作...
lock.release()      # 释放锁

在 Python 中，我们可以使用 with 语句来简化锁的使用，并确保锁总是能被正确地释放，例如：

import threading
lock = threading.Lock()
with lock:
    # ...执行一些操作...

使用 RLock 进行线程同步

Python 的 Lock 和 RLock 都是线程同步的基本工具，它们都可以防止多线程环境中的数据竞争。但是，它们在工作方式上有一些区别。

Lock 对象只能被获取一次。它不能再次被获取，直到它被释放。释放后，它可以被任何线程重新获取。另一方面，RLock（可重入锁）可以被同一个线程多次获取。它需要被释放相同次数才能被“解锁”。

例如，如果一个线程已经获取了一个 Lock，然后它再次尝试获取这个 Lock，那么这个线程将会被阻塞，直到这个 Lock 被释放。如果这个线程尝试获取一个已经被它自己获取的 RLock，那么这个操作将会成功，并且锁的持有计数器将会增加。

另一个区别是，Lock 对象可以被任何线程释放，而 RLock 对象只能被获取它的线程释放。此外，RLock 对象可以被多个线程拥有，而 Lock 对象一次只能被一个线程拥有。

使用信号量进行线程同步

在 Python 中，信号量（Semaphore）是一种基于计数的锁机制，它用于管理对共享资源的访问。信号量内部有一个计数器，初始值通常设置为共享资源的数量。当一个线程请求使用共享资源时，如果信号量的计数器大于 0，则计数器减 1，线程可以使用资源。如果计数器为 0，线程将被阻塞，直到其他线程释放资源，信号量计数器增加。

Python 的 threading 模块提供了一个信号量对象，我们可以通过以下方式来使用它：

import threading
 
# 创建一个信号量对象，初始值设为1（表示只有一个共享资源）
sem = threading.Semaphore(1)
# 在需要同步的代码前获取信号量（尝试减小信号量的计数器）
sem.acquire()
# ...执行一些操作...
# 在完成后释放信号量（增加信号量的计数器）
sem.release()

在这个例子中，acquire 方法会尝试减小信号量的计数器，如果计数器大于 0，那么操作会成功，否则线程将被阻塞，直到计数器变为正数。release 方法会增加信号量的计数器。

和锁一样，我们也可以使用 with 语句来简化信号量的使用，并确保信号量能被正确地释放，例如：

import threading
sem = threading.Semaphore(1)
with sem:
    # ...执行一些操作...

使用条件进行线程同步

条件指的是应用程序状态的改变。这是另一种同步机制，其中某些线程在等待某一条件发生，其他的线程会在该条件发生的时候进行通知。一旦条件发生，线程会拿到共享资源的唯一权限。

from threading import Thread, Condition
import time
 
items = []
condition = Condition()
 
class consumer(Thread):
 
    def __init__(self):
        Thread.__init__(self)
 
    def consume(self):
        global condition
        global items
        condition.acquire()
        if len(items) == 0:
            condition.wait()
            print("Consumer notify : no item to consume")
        items.pop()
        print("Consumer notify : consumed 1 item")
        print("Consumer notify : items to consume are " + str(len(items)))
 
        condition.notify()
        condition.release()
 
    def run(self):
        for i in range(0, 20):
            time.sleep(2)
            self.consume()
 
class producer(Thread):
 
    def __init__(self):
        Thread.__init__(self)
 
    def produce(self):
        global condition
        global items
        condition.acquire()
        if len(items) == 10:
            condition.wait()
            print("Producer notify : items producted are " + str(len(items)))
            print("Producer notify : stop the production!!")
        items.append(1)
        print("Producer notify : total items producted " + str(len(items)))
        condition.notify()
        condition.release()
 
    def run(self):
        for i in range(0, 20):
            time.sleep(1)
            self.produce()

使用事件进行线程同步

在 Python 中，事件是 threading 模块提供的一种简单的线程同步机制。事件对象管理着一个内部标志，可以由线程设置或清除。其他线程可以监听这个标志，从而实现线程间的同步。

事件对象主要提供了以下几个方法：

set()：将内部标志设置为 True。
clear()：将内部标志设置为 False。
wait()：阻塞线程，直到内部标志为 True。
is_set()：检查内部标志是否为 True。

以下是一个简单的使用示例：

import threading
event = threading.Event()
 
def worker():
    # 等待事件被设置
    event.wait()
    # 执行一些操作...
 
# 在其他线程中
event.set()  # 设置事件，使得worker线程可以继续执行

在这个例子中，worker 线程会在 event.wait() 处阻塞，直到其他线程调用 event.set() 设置事件。这样，我们就可以控制 worker 线程何时开始执行。

使用 queue 进行线程通信

在 Python 中，queue 模块提供了一个线程安全的队列类 queue.Queue()
，可以用来实现线程间的通信。队列中的元素可以由一个线程放入，再由另一个线程取出。这种方式可以避免多个线程同时访问或修改同一数据造成的竞态条件。
Queue 常用的方法有以下四个：

put(): 往 queue 中放一个 item
get(): 从 queue 删除一个 item，并返回删除的这个 item
task_done(): 每次 item 被处理的时候需要调用这个方法
join(): 所有 item 都被处理之前一直阻塞

LIST FROM [[]]

🪴 Obsidian Publish

探索