2022年 11月 4日

Python多线程

Python的多线程

目录

基本概念

并行与并发

进程与线程

进程与线程的特点

Python多线程编程

Threading模块

创建线程

方式一：直接继承Thread，改写对应的run方法

方式二：直接调用Thread

多线程问题

多线程加锁

多线程编程死锁

解决死锁的办法

多线程多进程对于cpu bound任务与I/O bound任务的性能对比

总结

一直想对Python的多线程和多进程做一个简单的总结，毕竟开发到后面始终都是会用到并发和并行的设计知识的，现简单总结如下。

基本概念

并行与并发

  • 并行：同时处理多个任务的能力，指的是任务数小于等于cpu核数，任务真的是一起执行的。
  • 并发：交替处理多个任务的能力。指的是任务数多于cpu核数，通过操作系统的各种任务调度算法，实现用多个任务“一起”执行（实际上总有一些任务不在执行，因为切换任务的速度相当快，看上去一起执行而已）

 

              并行                                                                        并发

进程与线程

进程与线程的特点

 

多线程

多进程

优点
  • 多线程能够有效提升I/O阻塞型程序的效率
  • 与进程相比，占用更少的系统资源
  • 线程间能够共享所述进程的资源，方便进行通信
  • 更好地利用多核处理器
  • 处理cpu密集型任务时比多线程要好
  • 单个进程的崩溃不会导致整个程序的崩溃

缺点
  • CPython中有全局解释器锁（GIL）的限制。抢占式多线程容易造成线程颠簸
  • 线程共享地址空间，一个线程非法操作共享数据崩溃后，会导致整个进程崩溃，继而多个线程崩溃
  • 进程之间没有共享资源，如果需要共享资源需要自己编码设计。实现进程间通信，继而完成数据传递或者数据共享。
  • 进程需要消耗更多的内存

Python多线程编程

Threading模块

Python3 通过两个标准库 _thread 和 threading 提供对线程的支持。

_thread 提供了低级别的、原始的线程以及一个简单的锁，它相比于 threading 模块的功能还是比较有限的。

threading 模块除了包含 _thread 模块中的所有方法外，还提供的其他方法：

threading.currentThread(): 返回当前的线程变量。

threading.enumerate(): 返回一个包含正在运行的线程的list。正在运行指线程启动后、结束前，不包括启动前和终止后的线程。

threading.activeCount(): 返回正在运行的线程数量，与len(threading.enumerate())有相同的结果。

除了使用方法外，线程模块同样提供了Thread类来处理线程，Thread类提供了以下方法:

run(): 用以表示线程活动的方法。

start():启动线程活动。

join([time]): 等待至线程中止。这阻塞调用线程直至线程的join() 方法被调用中止-正常退出或者抛出未处理的异常-或者是可选的超时发生。

isAlive(): 返回线程是否活动的。

getName(): 返回线程名。

setName(): 设置线程名。

创建线程

方式一：直接继承Thread，改写对应的run方法

  1. #!/usr/bin/python3
  2.  
  3.  
  4. import threading
  5. import time
  6.  
  7. exitFlag = 0
  8.  
  9. class myThread (threading.Thread):
  10.     def __init__(self, threadID, name, counter):
  11.         threading.Thread.__init__(self)
  12.         self.threadID = threadID
  13.         self.name = name
  14.         self.counter = counter
  15.     def run(self):
  16.         print ("开始线程：" + self.name)
  17.         print_time(self.name, self.counter, 5)
  18.         print ("退出线程：" + self.name)
  19.  
  20.  
  21.  
  22. def print_time(threadName, delay, counter):
  23.     while counter:
  24.         if exitFlag:
  25.             threadName.exit()
  26.         time.sleep(delay)
  27.         print ("%s: %s" % (threadName, time.ctime(time.time())))
  28.         counter -= 1
  29.  
  30.  
  31. # 创建新线程
  32. thread1 = myThread(1, "Thread-1", 1)
  33. thread2 = myThread(2, "Thread-2", 2)
  34.  
  35. # 开启新线程
  36. thread1.start()
  37. thread2.start()
  38. thread1.join()
  39. thread2.join()
  40.  
  41. print ("退出主线程")

 

 

方式二：直接调用Thread

  1. # 多任务可以由多进程完成，也可以由一个进程内的多线程完成。
  2. # 我们前面提到了进程是由若干线程组成的，一个进程至少有一个线程。
  3. # Python使用threading模块对线程进行操作
  4. import time, threading
  5.  
  6. # 新线程执行的代码
  7. def loop():
  8.     print('thread %s is running...' % threading.current_thread().name)
  9.     n = 0
  10.     while n < 5:
  11.         n += 1
  12.         print('thread %s >>> %s ' % (threading.current_thread().name, n))
  13.         time.sleep(1)
  14.     print('thread %s ended.' % threading.current_thread().name)
  15.  
  16.  
  17. if __name__ == "__main__":
  18.     print('thread %s is running...' % threading.current_thread().name)
  19.     t = threading.Thread(target=loop, name='LoopThread')
  20.     t.start()
  21.     # 如果不写join，似乎对子线程的运行没有什么影响，只是主线程会提前结束。
  22.     # 主线程的结束不会导致子线程的结束
  23.     t.join()
  24.  
  25. print('thread %s ended.' % threading.current_thread().name)

 

多线程问题

多线程加锁

当多线程中多个线程修改共享资源时，需要加锁，否则会出现线程不安全问题。

# 多线程和多进程最大的不同在于，多进程中，同一个变量，各自有一份拷贝存在于每个进程中，互不影响，

# 而多线程中，所有变量都由所有线程共享，所以，任何一个变量都可以被任何一个线程修改，因此，线程之间共享数据最大的危险在于多个线程同时改一个变量，把内容给改乱了。

# 上述所说的即为线程不安全问题

  1. import time, threading
  2.  
  3. balance = 0
  4. lock = threading.Lock()
  5.  
  6. def change_it(n):
  7.     # 先存后取，结果应该为0
  8.     global balance
  9.     balance += n
  10.     balance -= n
  11.  
  12.  
  13. def run_thread(n):
  14.     # 在跑线程的时候不加锁，就和之前写的Beringei操作数据库代码一样，由于不是原子性操作
  15.     # 两个线程并行交叉运行多条语句，所以会出现线程不安全，需要枷锁
  16.     for i in range(100000):
  17.         # lock.acquire()
  18.         # try:
  19.         change_it(n)
  20.         # finally:
  21.         #     lock.release()
  22.  
  23. if __name__ == "__main__":
  24.     t1 = threading.Thread(target=run_thread, args=(5,))
  25.     t2 = threading.Thread(target=run_thread, args=(8,))
  26.     t1.start()
  27.     t2.start()
  28.     t1.join()
  29.     t2.join()
  30.     print(balance)
  31.

# 锁的好处就是确保了某段关键代码只能由一个线程从头到尾完整地执行，坏处当然也很多，首先是阻止了多线程并发执行，包含锁的某段代码

# 实际上只能以单线程模式执行，效率就大大地下降了。其次，由于可以存在多个锁，不同的线程持有不同的锁，并试图获取对方持有的锁时，

# 可能会造成死锁，导致多个线程全部挂起，既不能执行，也无法结束，只能靠操作系统强制终止

以上代码段的结果由于没有加锁，不一定是0， 如下

所以需要对run_thread这个函数进行小小的改动

  1. def run_thread(n):
  2.     # 在跑线程的时候不加锁，就和之前写的Beringei操作数据库代码一样，由于不是原子性操作
  3.     # 两个线程并行交叉运行多条语句，所以会出现线程不安全，需要枷锁
  4.     for i in range(100000):
  5.         lock.acquire()
  6.         try:
  7.             change_it(n)
  8.         finally:
  9.             lock.release()

 

 

多线程编程死锁

如果两个线程分别占有一部分资源并且同时等待对方的资源，就会造成死锁，因为系统判断这部分资源正在使用，所以这两个线程如果不人为终止，将一直等待下去。死锁示意图如下：

 

死锁示例代码与死锁结果：

  1. import time
  2. import threading
  3.  
  4.  
  5. class Mythread(threading.Thread):
  6.     def __init__(self):
  7.         threading.Thread.__init__(self)
  8.  
  9.     def run(self):
  10.         self.foo()
  11.         self.bar()
  12.  
  13.     def foo(self):
  14.         locka.acquire()
  15.         print('I am %s got locka---%s' % (self.name, time.ctime()))
  16.         lockb.acquire()
  17.         print('I am %s got lockb---%s' % (self.name, time.ctime()))
  18.         lockb.release()
  19.         locka.release()
  20.  
  21.  
  22.     def bar(self):
  23.         lockb.acquire()
  24.         print('I am %s got lockb---%s' % (self.name, time.ctime()))
  25.         locka.acquire()
  26.         print('I am %s got locka---%s' % (self.name, time.ctime()))
  27.         locka.release()
  28.         lockb.release()
  29.  
  30.  
  31. if __name__ == '__main__':
  32.     locka = threading.Lock()
  33.     lockb = threading.Lock()
  34.  
  35.     for i in range(2):
  36.         t = Mythread()
  37.         t.start()

 

 

解决死锁的办法

将lock换为递归锁，即Rlock()

  1. import time
  2. import threading
  3.  
  4. class Mythread(threading.Thread):
  5.     def __init__(self):
  6.         threading.Thread.__init__(self)
  7.  
  8.     def run(self):
  9.         self.foo()
  10.         self.bar()
  11.  
  12.     def foo(self):
  13.         lock.acquire()
  14.         print('I am %s got locka---%s' % (self.name, time.ctime()))
  15.         lock.acquire()
  16.         print('I am %s got lockb---%s' % (self.name, time.ctime()))
  17.         lock.release()
  18.         lock.release()
  19.  
  20.     def bar(self):
  21.         lock.acquire()
  22.         print('I am %s got lockb---%s' % (self.name, time.ctime()))
  23.         lock.acquire()
  24.         print('I am %s got locka---%s' % (self.name, time.ctime()))
  25.         lock.release()
  26.         lock.release()
  27.  
  28. if __name__ == '__main__':
  29.     lock = threading.RLock()
  30.     for i in range(2):
  31.         t = Mythread()
  32.         t.start()

多线程多进程对于cpu bound任务与I/O bound任务的性能对比

  1. import requests
  2. import time
  3. from threading import Thread
  4. from multiprocessing import Process
  5.  
  6.  
  7. # 定义cpu密集型的计算函数
  8. def count(x, y):
  9.     c = 0
  10.     while c < 50000:
  11.         c += 1
  12.         x += x
  13.         y += y
  14.  
  15.  
  16. # 定义io密集型的文件读写函数
  17. def write():
  18.     f = open("test.txt", "w")
  19.     for x in range(5000000):
  20.         f.write("testwrite\n")
  21.     f.close()
  22.  
  23. def read():
  24.     f = open("test.txt", "r")
  25.     lines = f.readlines()
  26.     f.close()
  27.  
  28. # 定义网络请求函数
  29. _head = {
  30.     'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/48.0.2564.116 Safari/537.36'}
  31. url = "http://www.tieba.com"
  32.  
  33. def http_request():
  34.     try:
  35.         webPage = requests.get(url, headers=_head)
  36.         html = webPage.text
  37.         return {"context": html}
  38.     except Exception as e:
  39.         return {"error": e}
  40.  
  41.  
  42. # 测试单进程单线程IO密集型、cpu密集型、网络请求密集型操作所需要的时间
  43. def line():
  44.     # CPU密集操作
  45.     t = time.time()
  46.     for x in range(10):
  47.         count(1, 1)
  48.     print("Line cpu", time.time() - t)
  49.     # IO密集操作
  50.     t = time.time()
  51.     for x in range(10):
  52.         write()
  53.         read()
  54.     print("Line IO", time.time() - t)
  55.     # 网络请求密集型操作
  56.     t = time.time()
  57.     for x in range(10):
  58.         http_request()
  59.     print("Line Http Request", time.time() - t)
  60.  
  61. def multi_thread_cpu():
  62.     counts = []
  63.     t = time.time()
  64.     for x in range(10):
  65.         thread = Thread(target=count, args=(1, 1))
  66.         counts.append(thread)
  67.         thread.start()
  68.     e = counts.__len__()
  69.     while True:
  70.         for th in counts:
  71.             if not th.is_alive():
  72.                 e -= 1
  73.         if e <= 0:
  74.             break
  75.     print(time.time() - t)
  76.  
  77. def io():
  78.     write()
  79.     read()
  80.  
  81.  
  82. def multi_thread_io():
  83.     t = time.time()
  84.     ios = []
  85.     t = time.time()
  86.     for x in range(10):
  87.         thread = Thread(target=count, args=(1, 1))
  88.         ios.append(thread)
  89.         thread.start()
  90.  
  91.     e = ios.__len__()
  92.     while True:
  93.         for th in ios:
  94.             if not th.is_alive():
  95.                 e -= 1
  96.         if e <= 0:
  97.             break
  98.     print(time.time() - t)
  99.  
  100. def multi_thread_http():
  101.     t = time.time()
  102.     ios = []
  103.     t = time.time()
  104.     for x in range(10):
  105.         thread = Thread(target=http_request)
  106.         ios.append(thread)
  107.         thread.start()
  108.     e = ios.__len__()
  109.     while True:
  110.         for th in ios:
  111.             if not th.is_alive():
  112.                 e -= 1
  113.         if e <= 0:
  114.             break
  115.     print("Thread Http Request", time.time() - t)
  116.  
  117.  
  118. def mp_cpu():
  119.     counts = []
  120.     t = time.time()
  121.     for x in range(10):
  122.         process = Process(target=count, args=(1, 1))
  123.         counts.append(process)
  124.         process.start()
  125.     e = counts.__len__()
  126.     while True:
  127.         for th in counts:
  128.             if not th.is_alive():
  129.                 e -= 1
  130.         if e <= 0:
  131.             break
  132.     print("Multiprocess cpu", time.time() - t)
  133.  
  134.  
  135. def mp_io():
  136.     t = time.time()
  137.     ios = []
  138.     t = time.time()
  139.     for x in range(10):
  140.         process = Process(target=io)
  141.         ios.append(process)
  142.         process.start()
  143.  
  144.     e = ios.__len__()
  145.     while True:
  146.         for th in ios:
  147.             if not th.is_alive():
  148.                 e -= 1
  149.         if e <= 0:
  150.             break
  151.     print("Multiprocess IO", time.time() - t)
  152.  
  153.  
  154.  
  155. def mp_http():
  156.     t = time.time()
  157.     httprs = []
  158.     ios = []
  159.     t = time.time()
  160.     for x in range(10):
  161.         process = Process(target=http_request)
  162.         ios.append(process)
  163.         process.start()
  164.  
  165.     e = httprs.__len__()
  166.     while True:
  167.         for th in httprs:
  168.             if not th.is_alive():
  169.                 e -= 1
  170.         if e <= 0:
  171.             break
  172.     print("Multiprocess Http Request", time.time() - t)
  173.  
  174.  
  175. if __name__ == '__main__':
  176.     mp_cpu()
  177.     mp_io()
  178.     mp_http()

 

CPU密集型/s

IO密集型/s

HTTP密集型/s

Line（单进程单线程）

94.91825

22.462

7.3296

多线程

101.17

24.8605

0.5053333

多进程

53.89

12.784

0.5045

总结

多线程在IO密集型的操作下似乎也没有很大的优势（也许IO操作的任务再繁重一些就能体现出优势），在CPU密集型的操作下明显地比单线程线性执行性能更差，但是对于网络请求这种忙等阻塞线程的操作，多线程的优势便非常显著了

 

多进程无论是在CPU密集型还是IO密集型以及网络请求密集型（经常发生线程阻塞的操作）中，都能体现出性能的优势。不过在类似网络请求密集型的操作上，与多线程相差无几，但却更占用CPU等资源，所以对于这种情况下，我们可以选择多线程来执行

1、对于CPU密集型的程序，顺序执行比并发执行效率更高，这是因为并发时GIL带来了切换线程额外的开销。

   （循环处理、计数等）

2. 对于IO密集型的程序，Python的多线程并发执行能够提高效率的，因为，Python线程在等待IO时，会释放GIL；

  （文件处理、网络爬虫）

3、多核多线程比单核多线程更差，单核下的多线程，每次释放GIL，唤醒的那个线程都能获取到GIL锁，所以能够无缝执行，但多核下，CPU0释放GIL后，其他CPU上的线程都会进行竞争，但GIL可能会马上又被CPU0拿到，导致其他几个CPU上被唤醒后的线程会醒着等待到切换时间后又进入待调度状态，这样会造成线程颠簸(thrashing),导致效率更低。