使用 Python 多线程提升你的编码技能

原文：towardsdatascience.com/level-up-your-coding-skills-with-python-threading-8f1bd06b9476

https://github.com/OpenDocCN/towardsdatascience-blog-zh-2024/raw/master/docs/img/9cbfec975450d8357e227d828448ea09.png

由Sonika Agarwal在Unsplash上的照片

简介

在大多数机器学习工作中，你不会研究改进某些模型架构或设计新的损失函数。大多数时候你必须利用现有的东西并将其适应到你的用例中。因此，在架构设计和实现方面优化你的项目非常重要。一切从这里开始：你想要最优的代码，它应该是干净、可重用且尽可能快地运行。线程是 Python 内置的本地库，人们没有像应该的那样经常使用它。

关于线程

线程是程序程序将自身分割成两个或更多同时（或伪同时）运行的任务的方式……一般来说，线程包含在进程内，同一进程中的不同线程共享相同的资源。

在这篇文章中，我们不讨论多进程，但 Python 的多进程库与多线程库非常相似。一般来说：

多线程非常适合 I/O 密集型任务，比如在 for 循环中调用 API
多进程用于 CPU 密集型任务，比如需要一次性转换多个表格数据的任务

所以如果我们想同时运行多个任务，我们可以通过使用线程来实现。利用线程的 Python 库被称为 threading。

让我们从简单开始。我想让两个 Python 线程同时打印一些东西。让我们编写两个包含 for 循环以打印一些单词的函数。

defprint_hello():forxinrange(1_000):print("hello")defprint_world():forxinrange(1_000):print("world")

现在如果我是依次运行，我会在我的终端中看到 1.000 次“hello”后面跟着 1.000 次“world”。

让我们使用线程。让我们定义两个线程，并将每个线程分配给上面定义的函数。然后我们将启动线程。你应该会在你的终端上看到“hello”和“world”交替打印。

如果在继续执行你的代码之前你想等待线程完成，你可以通过使用：join()来实现。

importthreding thread_1=threding.Thread(target=print_hello)thread_2=threding.Thread(target=print_world)thread_1.start()thread_2.start()# wait for the threads to finish before continuing running the codethread_1.join()thread_2.join()print("do other stuff")

使用线程锁定资源

有时候可能会发生两个或更多线程编辑相同的资源，比如说一个包含数字的变量。

一个线程有一个总是将 1 加到该变量的 for 循环，另一个线程减 1。如果我们同时运行这些线程，它将“始终”具有零（或多或少）的值。但我们要实现不同的行为。第一个获得这个变量的线程需要添加或减去 1，直到达到某个限制。然后它将释放变量，另一个线程就可以自由地获得变量并执行其操作。

importthreadingimporttime x=0lock=threading.Lock()defadd_one():globalx,lock# use global to work with globa varslock.acquire()whilex<10:x=x+1print(x)time.sleep(1)print("reached maximum")lock.release()defsubtract_one():globalx,lock lock.acquire()while>-10:x=x-1print(x)time.sleep(1)print("reached minimum")lock.release()

在上面的代码中，我们有两个函数。每个函数将由一个线程运行。一旦函数开始，它将锁定变量 lock，这样第二个线程在第一个线程完成之前无法访问它。

thread_1=threading.Thread(target=add_one)thread_2=threading.Thread(target=subtract_one)thread_1.start()thread_2.start()

使用信号量加锁

我们可以通过使用信号量来实现与上面类似的结果。假设我们想要一个函数同时被一定数量的线程访问。这意味着不是所有线程都能访问这个函数，例如只有 5 个线程。其他线程需要等待这 5 个线程中的某些完成它们的计算才能访问该函数并运行脚本。我们可以通过使用信号量和将其值设置为 5 来实现这一点。要使用某些参数启动一个线程，我们可以在 Thread 对象中使用args。

importtimeimportthreading semaphore=threading.BoudnedSemaphore(value=5)deffunc(thread_number):print(f"{thread_number}is trying to access the resource")semaphore.acquire()print(f"{thread_number}granted access to the resource")time.sleep(12)#fake some computationprint(f"{thread_number}is releasing resource")semaphore.release()if__name__=="__main__":forthread_numberinrange(10):t=threading.Thread(target=func,args=(thread_number,)t.start()time.sleep(1)

事件

事件****是简单的信号机制，用于协调线程。你可以把事件想象成一个你可以设置或清除的标志，其他线程可以在它被设置之前等待，然后继续它们的工作。

例如，在下面的例子中，想要执行“func”函数的 thread_1 需要等待用户输入“yes”并触发事件，才能完成整个函数。

importthreading event=threading.Event()deffunc():print("This event function is waiting to be triggered")event.wait()print("event is triggered, performing action now")thread_1=threading.Thread(target=func)thread_1.start()x=input("Do you want to trigger the event? n")ifx=="yes":event.set()elseprint("you chose not to trigger the event")

守护线程

这些只是运行在后台的线程。即使这个后台线程仍在运行，主脚本也会终止。例如，你可以使用守护线程来持续读取一个随时间更新的文件。

让我们编写一个脚本，其中有一个守护线程持续从文件中读取并更新一个字符串变量，另一个线程将变量的内容打印到控制台。

importthreadingimporttime path="myfile.txt"text=""defread_from_file():globalpath,textwhileTrue:withopen(path,"r")asf:text=f.read()time.sleep(4)defprint_loop():forxinrange(30):print(text)time.sleep(1)thread_1=threading.Thread(target=read_from_file,daemon=True)thread_2=threading.Thread(target=print_loop)thread_1.start()thread_2.start()

队列

队列是一组遵循先进先出（FIFO）原则的项目集合。它是一种处理数据结构的方法，其中首先处理第一个元素，最后处理最新元素。

我们还可以更改我们处理集合中项目顺序的优先级。例如，LIFO 代表后进先出。或者，我们可以有一个优先队列，我们可以手动选择顺序。

https://github.com/OpenDocCN/towardsdatascience-blog-zh-2024/raw/master/docs/img/44c3dfac64561c0dd50f0a645cfb8372.png

src:miro.medium.com/v2/resize:fit:1280/0*HUWegihFk4x2x5vS.gif

如果多个线程想要处理一个项目列表，比如数字列表，可能会出现两个线程对同一项进行计算的问题。我们想要避免这种情况。因此，我们可以在线程之间共享一个队列，当一个线程对其项目进行计算时，该项目将从队列中移除。让我们看看一个例子。

importqueue q=queue.Queue()# it can also be a LifoQueue or PriorityQueuenumber_list=[10,20,30,40,50,60,70,80]fornumberinnumber_list:q.put(number)print(q.get())# -> 10print(1.het())# -> 20

机器学习项目中线程的例子

假设你正在做一个项目，需要数据流和预处理管道。这在许多项目中都会发生，比如物联网设备或任何类型的传感器。后台守护线程可以持续获取和预处理数据，而主线程则专注于推理。

例如，在一个简单的案例中，我需要使用我的相机流开发一个实时图像分类系统。我会设置我的线程，拥有 2 个线程：

实时从相机流中获取图像。
将图像传递给机器学习模型进行推理。

importthreadingimporttimeimportqueueimportrandom# Sfake image classifierdefclassify_image(image):time.sleep(0.5)# fake the model inference timereturnf"Classified{image}"defcamera_feed(image_queue,stop_event):whilenotstop_event.is_set():# Simulate capturing an imageimage=f"Image_{random.randint(1,100)}"print(f"[Camera] Captured{image}")image_queue.put(image)time.sleep(1)# Simulate time between capturesdefmain_inference_loop(image_queue,stop_event):whilenotstop_event.is_set()ornotimage_queue.empty():try:image=image_queue.get(timeout=1)# Fetch image from the queueresult=classify_image(image)print(f"[Model]{result}")exceptqueue.Empty:continueif__name__=="__main__":image_queue=queue.Queue()stop_event=threading.Event()camera_thread=threading.Thread(target=camera_feed,args=(image_queue,stop_event),daemon=True)camera_thread.start()try:main_inference_loop(image_queue,stop_event)exceptKeyboardInterrupt:print("Shutting down...")stop_event.set()# Signal the camera thread to stopfinally:camera_thread.join()# Ensure the camera thread terminates properlyprint("All threads terminated.")