Python多线程操作

什么是线程：

线程（Thread）也称为轻量级进程。它是操作系统可执行操作调度的最小单位。它包含在过程中，并且是过程中的实际操作单元。线程不拥有系统资源，而仅具有运行中必不可少的一些资源，但是它可以与属于同一进程的其他线程共享该进程拥有的所有资源。一个线程可以创建和取消另一个线程，并且同一进程中的多个线程可以并发执行。

举一个简单的例子来理解：
假设有一个7 * 24小时不间断的工厂。由于功率有限，一次只能使用一个车间。当一个生产车间投入生产时，其他生产车间将关闭。在这里我们可以了解到，这个工厂相当于一个操作系统，电源设备相当于一个CPU，而一个车间相当于一个进程。

一个车间里可能有很多工人。他们共同努力完成一项任务。讲习班中的空间由工作人员共享，其中一个工作人员等效于一个线程，并且一个进程可以包含多个线程。例如，许多房间可供每个工人使用。这表示进程的内存空间是共享的，并且每个线程都可以使用这些共享的内存。

有时资源是有限的。例如，某些房间最多只能容纳一个人。当一个人占领时，其他人无法进入，只能等待。这意味着当一个线程使用某些共享内存时，其他线程必须等待其结束才可以使用此内存。

防止其他人进入的一种简单方法是在门上加锁。先到达的人将门锁上，然后到达的人看到门锁并在门口排队。等待锁打开，然后再输入。这称为“互斥”（互斥，缩写为Mutex），它防止多个线程同时读取和写入某个内存区域。
也有可以同时容纳n个人的房间，例如厨房。换句话说，如果人数大于n，则多余的人只能在外面等。这就像某些内存区域只能由固定数量的线程使用。目前的解决方案是在门上悬挂n个钥匙。进来的人拿出一把钥匙，出来时把钥匙挂回去。后来到达的人发现钥匙是空的，并且知道他必须在门口排队等候。这种方法称为“信号量”（Semaphore），用于确保多个线程不会相互冲突。

不难看出互斥锁是信号量的一种特殊情况（当n = 1时）。换句话说，完全有可能用后者代替前者。但是，由于互斥锁相对简单且高效，因此在必须独占资源时仍会使用此设计。

线程具有就绪，已阻止，正在运行的三种基本状态。
1.就绪状态意味着线程具有所有要运行的条件，它可以逻辑地运行，等待处理器；
2.运行状态表示线程拥有处理器正在运行；
3.阻塞状态意味着线程正在等待逻辑上不可执行的事件（例如某个信号量）。

这三种状态的相互转换如下所示：

多线程的优点

因此，问题是，与单线程相比，多线程有哪些优势？
优点是显而易见的，它可以提高资源利用率，并使程序响应更快。单线程按顺序执行，例如，单线程程序执行以下操作：

5Read file A in seconds
3Process file A in seconds
5Read file B in seconds
3Process file B in seconds

需要16秒钟才能完成。如果开始执行两个线程，它将如下所示：

5Read file A in seconds
5Read file B+ in seconds 3Process file A in seconds
3Process file B in seconds

需要13秒才能完成。

Python中的多线程GIL

当涉及到Python中的多线程时，不能回避的一个主题是全局锁GIL（全局解释器锁）。GIL限制一次只能运行一个线程，并且不能利用多核CPU。首先需要明确的一点是，GIL不是Python的功能，它是在实现Python解析器（CPython）时引入的一个概念。就像C ++是一组语言（语法）标准一样，但是可以使用不同的编译器将其编译为可执行代码。著名的编译器，例如GCC，INTEL C ++，Visual C ++等。Python也是如此，并且相同的代码段可以由不同的Python执行环境（例如CPython，PyPy和Psyco）执行。像JPython一样，没有GIL。但是，由于CPython是大多数环境中的默认Python执行环境。所以，在很多人的概念中，CPython是Python，并且假定GIL归因于Python语言的缺陷。因此，在这里让我们清楚：GIL不是Python的功能，Python可以完全独立于GIL。

GIL的本质是互斥锁。由于它是互斥锁，因此所有互斥锁本质上是相同的。它们都将并行操作更改为串行操作，以便控制共享数据一次只能被一个任务修改。确保数据安全。在Python进程中，不仅有主线程或由主线程启动的其他线程，而且还有解释器级别的线程，例如由解释器启用的垃圾回收。简而言之，所有线程都在此过程中运行，所有数据都被共享。其中，代码作为一种数据也被所有线程共享。多个线程首先访问解释器的代码，即获得执行许可，然后将目标的代码提供给要执行的解释器的代码，
解释器的代码由所有线程共享，因此垃圾回收线程也可能访问解释器的代码并执行它，这导致一个问题：对于相同的数据100，它可能是线程1在x处执行x = 100同时，垃圾回收执行回收100的操作。没有解决此问题的聪明方法，即锁定处理，即GIL。
因此，随着GIL的存在，在同一进程中只能同时执行一个线程，然后有人可能会问：该进程可以使用多核，但是Python的多线程不能使用多核优点，就是Python的多线程没用吗？
当然答案是否定的。
首先，很清楚我们的线程执行什么任务，无论是执行计算（计算密集型）还是输入和输出（I / O密集型），并且在不同的场景中使用不同的方法。多核CPU意味着多核可以并行完成计算，因此多核可提高计算性能，但是一旦每个CPU遇到I / O阻塞，它仍然需要等待，因此多核对于I / O来说并不太高。 O密集型任务促进。

这里有两个示例说明：

示例1：计算密集型任务

计算密集型任务-多进程

from multiprocessing import Process
import os, time#Computation-intensive tasks
def work():res = 0for i in range(100000000):res *= i if __name__ == "__main__":l = []print("This machine is",os.cpu_count(),"Core CPU")  # This machine is 4 coresstart = time.time()for i in range(4):p = Process(target=work)  # multi-Progressl.append(p)p.start()for p in l:p.join()stop = time.time()print("Computation-intensive tasks, multi-process takes %s" % (stop - start))

结果如下：

This machine is 4  Core CPUComputationally intensive tasks, multi-process time-consuming 14.901630640029907

计算密集型任务-多线程

from threading import Thread
import os, time#Computation-intensive tasks
def work():res = 0for i in range(100000000):res *= iif __name__ == "__main__":l = []print("This machine is",os.cpu_count(),"Core CPU")  # This machine is 4 coresstart = time.time()for i in range(4):p = Thread(target=work)  # multi-Progressl.append(p)p.start()for p in l:p.join()stop = time.time()print("Computation-intensive tasks, multi-threading takes %s" % (stop - start))

结果如下：

This machine is 4  Core CPUComputationally intensive tasks, multi-threaded 23.559885025024414

示例2：I / O密集型任务

I / O密集型任务-多进程

from multiprocessing import Process
import os, time#I/0intensive tasks
def work():time.sleep(2)print("===>", file=open("tmp.txt", "w"))if __name__ == "__main__":l = []print("This machine is", os.cpu_count(), "Core CPU")  # This machine is 4 coresstart = time.time()for i in range(400):p = Process(target=work)  # multi-Progressl.append(p)p.start()for p in l:p.join()stop = time.time()print("I/0 intensive task, multi-process takes %s" % (stop - start))

结果如下：

This machine is 4  Core CPU
I/0Intensive tasks, multi-process time-consuming 21.380212783813477

I / O密集型任务-多线程

from threading import Thread
import os, time#I/0intensive tasks
def work():time.sleep(2)print("===>", file=open("tmp.txt", "w"))if __name__ == "__main__":l = []print("This machine is", os.cpu_count(), "Core CPU")  # This machine is 4 coresstart = time.time()for i in range(400):p = Thread(target=work)  # Multithreadingl.append(p)p.start()for p in l:p.join()stop = time.time()print("I/0 intensive tasks, multi-threading takes %s" % (stop - start))

结果如下：

This machine is 4  Core CPU
I/0Intensive tasks, multi-threaded 2.1127078533172607

结论：在Python中，多进程在计算密集型任务中占主导地位，而多线程在I / O密集型任务中占主导地位。
当然，对于运行程序，执行效率肯定会随着CPU的增加而提高，这是因为程序基本上不是纯粹的计算或纯粹的I / O，所以我们只能看看程序是计算密集型还是I / O密集型。

如何使用Python多线程

Python提供了如下的多线程编程模块：

_thread
threading
Queue
multiprocessing

_thread模块提供了低级基本功能来支持多线程功能，并提供了简单的锁以确保同步。建议使用穿线模块。
2.线程模块封装了_thread，它提供了更高级别，更强大且更易于使用的线程管理功能。对线程的支持更加完整和广泛。在大多数情况下，仅高级线程模块就足够了。

使用线程进行多线程操作：
方法1：创建一个threading.Thread实例并调用其start（）方法

import time
import threadingdef task_thread(counter):print(f'Thread name: {threading.current_thread().name} Parameter: {counter} Start time: {time.strftime("%Y-%m-%d %H:%M:%S")}')num = counterwhile num:time.sleep(3)num -= 1print(f'Thread name: {threading.current_thread().name} Parameter: {counter} End time: {time.strftime("%Y-%m-%d %H:%M:%S")}')if __name__ == '__main__':print(f'Main thread start time: {time.strftime("%Y-%m-%d %H:%M:%S")}')#Initialize 3 threads, passing different parameterst1 = threading.Thread(target=task_thread, args=(3,))t2 = threading.Thread(target=task_thread, args=(2,))t3 = threading.Thread(target=task_thread, args=(1,))#Open three threadst1.start()t2.start()t3.start()#Wait for the end of the runt1.join()t2.join()t3.join()print(f'Main thread end time: {time.strftime("%Y-%m-%d %H:%M:%S")}')

结果如下

Main thread start time:2018-07-06 23:03:46Thread name: Thread-1  parameter:3  Starting time:2018-07-06 23:03:46Thread name: Thread-2  parameter:2  Starting time:2018-07-06 23:03:46Thread name: Thread-3  parameter:1  Starting time:2018-07-06 23:03:46Thread name: Thread-3  parameter:1  End Time:2018-07-06 23:03:49Thread name: Thread-2  parameter:2  End Time:2018-07-06 23:03:52Thread name: Thread-1  parameter:3  End Time:2018-07-06 23:03:55Main thread end time:2018-07-06 23:03:55

方法2：继承Thread类并重写子类中的run（）和init（）方法

import time
import threadingclass MyThread(threading.Thread):def __init__(self, counter):super().__init__()self.counter = counterdef run(self):print(f'Thread name: {threading.current_thread().name} Parameter: {self.counter} Start time: {time.strftime("%Y-%m-%d %H:%M:%S")}')counter = self.counterwhile counter:time.sleep(3)counter -= 1print(f'Thread name: {threading.current_thread().name} Parameter: {self.counter} End time: {time.strftime("%Y-%m-%d %H:%M:%S")}')if __name__ == "__main__":print(f'Main thread start time: {time.strftime("%Y-%m-%d %H:%M:%S")}')# Initialize 3 threads and pass different parameterst1 = MyThread(3)t2 = MyThread(2)t3 = MyThread(1)# Start three threadst1.start()t2.start()t3.start()# Wait for the run to endt1.join()t2.join()t3.join()print(f'Main thread end time: {time.strftime("%Y-%m-%d %H:%M:%S")}')

运算结果如下，与方法一的运算结果一致

Main thread start time:2018-07-06 23:34:16Thread name: Thread-1  parameter:3  Starting time:2018-07-06 23:34:16Thread name: Thread-2  parameter:2  Starting time:2018-07-06 23:34:16Thread name: Thread-3  parameter:1  Starting time:2018-07-06 23:34:16Thread name: Thread-3  parameter:1  End Time:2018-07-06 23:34:19Thread name: Thread-2  parameter:2  End Time:2018-07-06 23:34:22Thread name: Thread-1  parameter:3  End Time:2018-07-06 23:34:25Main thread end time:2018-07-06 23:34:25

如果您继承Thread类并想调用外部传入函数，则代码如下

import time
import threadingdef task_thread(counter):print(f'Thread name: {threading.current_thread().name} Parameter: {counter} Start time: {time.strftime("%Y-%m-%d %H:%M:%S")}')num = counterwhile num:time.sleep(3)num -= 1print(f'Thread name: {threading.current_thread().name} Parameter: {counter} End time: {time.strftime("%Y-%m-%d %H:%M:%S")}')class MyThread(threading.Thread):def __init__(self, target, args):super().__init__()self.target = targetself.args = argsdef run(self):self.target(*self.args)if __name__ == "__main__":print(f'Main thread start time: {time.strftime("%Y-%m-%d %H:%M:%S")}')# Initialize 3 threads and pass different parameterst1 = MyThread(target=task_thread,args=(3,))t2 = MyThread(target=task_thread,args=(2,))t3 = MyThread(target=task_thread,args=(1,))# Start three threadst1.start()t2.start()t3.start()# Wait for the run to endt1.join()t2.join()t3.join()print(f'Main thread end time: {time.strftime("%Y-%m-%d %H:%M:%S")}')

这样，它与第一种方法相同。实例化自定义线程类，并且运行结果保持不变。

线程同步锁（互斥锁）：

如果多个线程一起修改某个数据，则可能会发生不可预测的结果。在这种情况下，您需要使用互斥锁来改善同步。在下面显示的代码中，三个线程对公用变量num执行一百万次加减运算后，num的结果不为0。

import time, threadingnum = 0def task_thread(n):global numfor i in range(1000000):num = num + nnum = num - nt1 = threading.Thread(target=task_thread, args=(6,))
t2 = threading.Thread(target=task_thread, args=(17,))
t3 = threading.Thread(target=task_thread, args=(11,))
t1.start()
t2.start()
t3.start()
t1.join()
t2.join()
t3.join()
print(num)

结果：

-19

之所以存在非0的情况，是因为修改num需要多个语句，所以当一个线程执行num + n时，另一个线程执行num-m，导致前一个线程执行num-n时num的值是不再是先前的值，导致最终结果不为0。
为了确保数据的准确性，您需要使用互斥锁来同步多个线程，限制一个线程访问数据的时间，其他线程只能等待直到前一个线程释放锁。使用threading.Thread对象的Lock和Rlock可以实现简单的线程同步。这两个对象都有获取和释放方法。对于一次只需要一个线程运行的数据，您可以将其操作放在获取和释放Between方法之间。如下：

import time, threadingnum = 0
lock = threading.Lock()
def task_thread(n):global num# Acquire lock for thread synchronizationlock.acquire()for i in range(1000000):num = num + nnum = num - n#Release the lock and start the next threadlock.release()t1 = threading.Thread(target=task_thread, args=(6,))
t2 = threading.Thread(target=task_thread, args=(17,))
t3 = threading.Thread(target=task_thread, args=(11,))
t1.start(); t2.start(); t3.start()
t1.join(); t2.join(); t3.join()
print(num)

运算结果：

线程同步信号量（信号量）

互斥锁仅允许一个线程同时访问共享数据，而信号量则允许一定数量的线程同时访问共享数据。例如，如果在银行柜台有5个窗口，则允许5个人同时处理业务，而后面的人只能在前面等候。完成业务后，您只能去柜台。
示例代码如下：

import threading
import time# Only 5 people handle business at the same time
semaphore = threading.BoundedSemaphore(5)
# Simulated banking business
def yewubanli(name):semaphore.acquire()time.sleep(3)print(f"{time.strftime('%Y-%m-%d %H:%M:%S')} {name} is processing business")semaphore.release()thread_list = []
for i in range(12):t = threading.Thread(target=yewubanli, args=(i,))thread_list.append(t)for thread in thread_list:thread.start()for thread in thread_list:thread.join()# while threading.active_count() != 1:
#    time.sleep(1)

结果如下

2018-07-08 12:33:57 4  Business in progress
2018-07-08 12:33:57 1  Business in progress
2018-07-08 12:33:57 3  Business in progress
2018-07-08 12:33:57 0  Business in progress
2018-07-08 12:33:57 2  Business in progress
2018-07-08 12:34:00 7  Business in progress
2018-07-08 12:34:00 5  Business in progress
2018-07-08 12:34:00 6  Business in progress
2018-07-08 12:34:00 9  Business in progress
2018-07-08 12:34:00 8  Business in progress
2018-07-08 12:34:03 11  Business in progress
2018-07-08 12:34:03 10  Business in progress

线程同步条件

条件对象可以使线程A停止并等待其他线程B。线程B满足特定条件后，通知线程A继续运行。线程首先获取条件变量锁。如果条件不足，线程将等待并释放条件变量锁。如果满足该线程，则执行该线程，并且还可以通知其他具有等待状态的线程。处于等待状态的其他线程在收到通知后将重新判断条件。
以下是一个有趣的示例

import threadingclass Boy(threading.Thread):def __init__(self, cond, name):super(Boy, self).__init__()self.cond = condself.name = namedef run(self):self.cond.acquire()print(self.name + ": Marry me!?")self.cond.notify()  # Wake up a suspended thread, let hanmeimei standself.cond.wait()  # Release the internal occupancy, and the thread is suspended until the notification is received to wake up or time out, waiting for hanmeimei to answerprint(self.name + ": I knelt down and gave the ring!")self.cond.notify()self.cond.wait()print(self.name + ": Mrs. Li, your choice is too Meiji.")self.cond.release()class Girl(threading.Thread):def __init__(self, cond, name):super(Girl, self).__init__()self.cond = condself.name = namedef run(self):self.cond.acquire()self.cond.wait()  # Waiting for Lilei to proposeprint(self.name + ": No sentiment, not romantic enough, don't agree")self.cond.notify()self.cond.wait()print(self.name + ": Okay, promise you")self.cond.notify()self.cond.release()cond = threading.Condition()
boy = Boy(cond, "LiLei")
girl = Girl(cond, "HanMeiMei")
girl.start()
boy.start()

结果如下：

LiLei: Marry me! ?HanMeiMei: No sentiment, not romantic enough, don’t agree
LiLei: I knelt down and gave the ring!HanMeiMei: Okay, promise youLiLei: Mrs. Li, your choice is too Meiji.

线程同步事件

事件用于线程间通信。一个线程发送信号，其他一个或多个线程等待，调用事件对象的wait方法，该线程将阻止等待，直到设置了另一个线程才被唤醒。上面的求婚示例使用事件代码，如下所示：

import threading, timeclass Boy(threading.Thread):def __init__(self, cond, name):super(Boy, self).__init__()self.cond = condself.name = namedef run(self):print(self.name + ": Marry me!?")self.cond.set()  # Wake up a suspended thread, let hanmeimei standtime.sleep(0.5)self.cond.wait()print(self.name + ": I knelt down and gave the ring!")self.cond.set()time.sleep(0.5)self.cond.wait()self.cond.clear()print(self.name + ": Mrs. Li, your choice is too Meiji.")class Girl(threading.Thread):def __init__(self, cond, name):super(Girl, self).__init__()self.cond = condself.name = namedef run(self):self.cond.wait()  # Waiting for Lilei to proposeself.cond.clear()print(self.name + ": No sentiment, not romantic enough, don't agree")self.cond.set()time.sleep(0.5)self.cond.wait()print(self.name + ": Okay, promise you")self.cond.set()cond = threading.Event()
boy = Boy(cond, "LiLei")
girl = Girl(cond, "HanMeiMei")
boy.start()
girl.start()

结果如下：

LiLei: Marry me! ?HanMeiMei: No sentiment, not romantic enough, don’t agreeHanMeiMei: Okay, promise youLiLei: I knelt down and gave the ring!LiLei: Mrs. Li, your choice is too Meiji

线程优先级队列（队列）

Python的队列模块提供了同步的线程安全队列类，包括先进先出队列Queue，后进先出队列LifoQueue和优先级队列PriorityQueue。这些队列都实现了锁原语，可以直接用于实现线程之间的同步。
举个简单的例子，如果有一个用于存放冷饮的小冰箱，如果该小冰箱只能容纳5杯冷饮，则A不断将冷饮放入冰箱，B不断从冰箱中取出冷饮。A和B的释放速度可能不相同。如何使它们保持同步？队列在这里派上用场了。
首先看一下代码

import threading,timeimport queue#First in first out
q = queue.Queue(maxsize=5)
#q = queue.LifoQueue(maxsize=3)
#q = queue.PriorityQueue(maxsize=3)def ProducerA():count = 1while True:q.put(f"Cold Drink {count}")print(f"A Put in: [Cold Drink {count}]")count +=1time.sleep(1)def  ConsumerB():while True:print(f"B Take out [{q.get()}]")time.sleep(5)p = threading.Thread(target=ProducerA)
c = threading.Thread(target=ConsumerB)
c.start()
p.start()

结果如下：

16:29:19  A Put in: [Cold Drink 1]
16:29:19  B Take out [Cold Drink 1]
16:29:20  A Put in: [Cold Drink 2]
16:29:21  A Put in: [Cold Drink 3]
16:29:22  A Put in: [Cold Drink 4]
16:29:23  A Put in: [Cold Drink 5]
16:29:24  B Take out [Cold Drink 2]
16:29:24  A Put in: [Cold Drink 6]
16:29:25  A Put in: [Cold Drink 7]
16:29:29  B Take out [Cold Drink 3]
16:29:29  A Put in: [Cold Drink 8]
16:29:34  B Take out [Cold Drink 4]
16:29:34  A Put in: [Cold Drink 9]

上面的代码是生产者和消费者模型的最简单示例。在并发编程中使用生产者和消费者模式可以解决大多数并发问题。如果生产者的处理速度非常快，而使用者的处理速度非常慢，则生产者必须等待使用者完成处理后才能继续产生数据。同样，如果消费者比生产者拥有更多的处理能力，则消费者必须等待生产者。为了解决这个问题，引入了生产者和消费者模型。生产者－消费者模型通过容器（队列）解决了生产者与消费者之间的强耦合问题。生产者和消费者之间并不直接进行通信，而是通过阻塞队列进行通信，因此，生产者不必等到消费者处理完数据后再处理数据，就可以将它们直接放入阻塞队列中。消费者不向生产者索要数据，但是直接从阻塞队列中获取数据，阻塞队列相当于一个缓冲区，平衡了生产者和消费者的处理能力。以下是生产者使用者模式的数据流程图：

多处理

Python中的线程和进程使用的同一模块多处理。使用方法基本相同，唯一的区别是从多处理导入池中导入的池代表进程池，从多处理导入池中的导入池代表虚拟线程池。这样可以在线程中实现并发。
线程池示例：

from multiprocessing.dummy import Pool as ThreadPool
import timedef fun(n):time.sleep(2)start = time.time()
for i in range(5):fun(i)
print("Single-thread sequential execution takes time:", time.time() - start)start2 = time.time()
# Open 8 workers, the default is the number of CPU cores when there is no parameter
pool = ThreadPool(processes=2)
# Execute urllib2.urlopen(url) in the thread and return the execution result
results2 = pool.map(fun, range(5))
pool.close()
pool.join()
print("Thread pool (5) concurrent execution time-consuming:", time.time() - start2)

上面的代码模拟了一个耗时2秒的任务，并比较了按顺序执行5次和线程池（并行数为5）的执行所花费的时间。结果如下

Single thread sequential execution takes time: 10.002546310424805Thread Pool(5) Concurrent execution takes time: 2.023442268371582

显然，并发执行效率更高，接近单次执行的时间。

总结

Python多线程适用于I / O密集型任务。I / O密集型任务花在CPU计算上的时间更少，而在I / O上花费更多的时间，例如文件读写，Web请求，数据库请求等。对于计算密集型任务，应使用多个进程。