multiprocessing

multiprocessing 是一个用与 threading 模块相似API的支持产生进程的包。 multiprocessing 包同时提供本地和远程并发，使用子进程代替线程，有效避免 Global Interpreter Lock 带来的影响。因此， multiprocessing 模块允许程序员充分利用机器上的多个核心。Unix 和 Windows 上都可以运行。
multiprocessing 模块还引入了在 threading 模块中没有类似物的API。这方面的一个主要例子是 Pool 对象，它提供了一种方便的方法，可以跨多个输入值并行化函数的执行，跨进程分配输入数据（数据并行）。以下示例演示了在模块中定义此类函数的常见做法，以便子进程可以成功导入该模块。这个数据并行的基本例子使用 Pool 

  1. from multiprocessing import Pool
  3. def f(x):
  4.     return x*x
  6. if __name__ == '__main__':
  7.     with Pool(5) as p:
  8.         print(p.map(f, [1, 2, 3]))
  9. >>

threading

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

  • threading.currentThread(): 返回当前的线程变量。
  • threading.enumerate(): 返回一个包含正在运行的线程的list。正在运行指线程启动后、结束前，不包括启动前和终止后的线程。
  • threading.activeCount(): 返回正在运行的线程数量，与len(threading.enumerate())有相同的结果。

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

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

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

    1. import threading
    2. def action(*add):
    3.   for arc in add:
    4.       print(threading.currentThread().getName()+' '+arc)
    5. my_tuple = ('tuple_1','tuple_2','tuple_3')
    6. # 创建线程
    7. thread=threading.Thread(target=action,args=my_tuple)
    8. thread.start()
    9. thread.join()
    10. for i in range(5):
    11.   print(threading.currentThread().getName())
    12. >>>>
    13. #action执行完成退出后才执行主线程
    14. Thread-9 tuple_1
    15. Thread-9 tuple_2
    16. Thread-9 tuple_3
    17. MainThread
    18. MainThread
    19. MainThread
    20. MainThread
    21. MainThread

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

  • getName(): 返回线程名。
  • setName(): 设置线程名。
  • daemon(): 一个表示这个线程是（True）否（False）守护线程的布尔值。一定要在调用 start() 前设置好，不然会抛出 RuntimeError 。初始值继承于创建线程；主线程不是守护线程，因此主线程创建的所有线程默认都是 daemon = False。

    创建线程

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

向线程中传递函数

创建类的实例，使用args函数

  1. import threading
  2. def func1(s,func):
  3.     fun(s)
  4. def ff(s):
  5.     print(f'输出了{s}')
  6. t1 = threading.Thread(target=func1,args=('hello world',ff))
  7. t1.start()

线程取得返回值

  1. import threading
  2. class MyThread(threading.Thread):
  3.     def __init__(self,func,args=()):
  4.         super(MyThread,self).__init__()
  5.         self.func=func
  6.         self.args=args
  7.     def run(self):
  8.         self.result = self.func(*self.args)
  9.     def get_return(self):
  10.         threading.Thread.join(self)
  11.         try:
  12.             return self.result
  13.         except Exception as e:
  14.             return None

线程中创建和使用全局对象

使用threading.local()来定义

  1. import threading
  2. import time
  3. a = threading.local()
  4. def worker():
  5.     a.x = 0
  6.     for i in range(20):
  7.         a.x +=1
  8.     print(threading.current_thread(),a.x)
  9. for i in range(10):
  10.     threading.Thread(target=worker).start()

image.png
生成了10个线程，每个线程a.x最后都是20,相互不干扰

线程同步

如果多个线程共同对某个数据修改，则可能出现不可预料的结果，为了保证数据的正确性，需要对多个线程进行同步。
使用 Thread 对象的 Lock 和 Rlock 可以实现简单的线程同步，这两个对象都有 acquire 方法和 release 方法，对于那些需要每次只允许一个线程操作的数据，可以将其操作放到 acquire 和 release 方法之间。如下：
多线程的优势在于可以同时运行多个任务（至少感觉起来是这样）。但是当线程需要共享数据时，可能存在数据不同步的问题。
考虑这样一种情况：一个列表里所有元素都是0，线程”set”从后向前把所有元素改成1，而线程”print”负责从前往后读取列表并打印。
那么，可能线程”set”开始改的时候，线程”print”便来打印列表了，输出就成了一半0一半1，这就是数据的不同步。为了避免这种情况，引入了锁的概念。
锁有两种状态——锁定和未锁定。每当一个线程比如”set”要访问共享数据时，必须先获得锁定；如果已经有别的线程比如”print”获得锁定了，那么就让线程”set”暂停，也就是同步阻塞；等到线程”print”访问完毕，释放锁以后，再让线程”set”继续。
经过这样的处理，打印列表时要么全部输出0，要么全部输出1，不会再出现一半0一半1的尴尬场面。

  1. import threading
  2. import time
  3. class myThread (threading.Thread):
  4.     def __init__(self, threadID, name, delay):
  5.         threading.Thread.__init__(self)
  6.         self.threadID = threadID
  7.         self.name = name
  8.         self.delay = delay
  9.     def run(self):
  10.         print ("开启线程： " + self.name)
  11.         # 获取锁，用于线程同步
  12.         threadLock.acquire()
  13.         print_time(self.name, self.delay, 3)
  14.         # 释放锁，开启下一个线程
  15.         threadLock.release()
  16. def print_time(threadName, delay, counter):
  17.     while counter:
  18.         time.sleep(delay)
  19.         print ("%s: %s" % (threadName, time.ctime(time.time())))
  20.         counter -= 1
  21. #创建threading的Lock()方法对象
  22. threadLock = threading.Lock()
  23. threads = []
  24. # 创建新线程
  25. thread1 = myThread(1, "Thread-1", 1)
  26. thread2 = myThread(2, "Thread-2", 2)
  28. # 开启新线程
  29. thread1.start()
  30. thread2.start()
  32. # 添加线程到线程列表
  33. threads.append(thread1)
  34. threads.append(thread2)
  36. # 等待所有线程完成,Thread-1完成后才会开始Thread-2
  37. for t in threads:
  38.     t.join()
  39. print ("退出主线程")
  40. >>>>
  42. 开启线程： Thread-1
  43. 开启线程： Thread-2
  44. Thread-1: Thu Jan 20 16:40:03 2022
  45. Thread-1: Thu Jan 20 16:40:04 2022
  46. Thread-1: Thu Jan 20 16:40:05 2022
  47. Thread-2: Thu Jan 20 16:40:07 2022
  48. Thread-2: Thu Jan 20 16:40:09 2022
  49. Thread-2: Thu Jan 20 16:40:11 2022
  50. 退出主线程

线程锁仅仅用来确保线程不会被打断，但是无法限制进程访问的个数，所有就需要引入一个信号量的概念，用来控制线程访问和释放的次数
信号量是一个计数器，当资源释放时，计数器就递增，当资源申请时，计数器就递减

  1. from threading import BoundedSemaphore
  2. MAX = 10
  3. semaphore = BoundedSemaphore(MAX)
  4. print(semaphore._value)
  5. semaphore.acquire()
  6. print(semaphore._value)
  7. semaphore.release()
  8. print(semaphore._value)
  9. >>>>
  10. 10
  11. 9
  12. 10

当semaphore._value为0 时，表示资源消耗殆尽，再次调用acquire()方法会阻塞

多线程下载图片

  1. import urllib3
  2. import threading
  3. http = urllib3.PoolManager()
  4. disable_warnings()
  5. f = open('url.txt','r')
  6. urllist = []
  7. for line in f.readlines():
  8.     urllist.append(line.rstrip())
  9. class Download(threading.Thread):
  10.     def __init__(self,func,args):
  11.         super().__init__(target=func,args=args)
  12. def download(filename,url):
  13.     response = http.request('GET',url)
  14.     f = open(filename,'wb')
  15.     f.write(response.data)
  16.     f.close()
  17.     print('<%s>下载完成'%url)
  18. for i in range(len(urllist)):
  19.     thread = Download(download,(str(i)+'.jpg',urllist[i]))
  20.     thread.start()

线程优先级队列

Python 的 Queue 模块中提供了同步的、线程安全的队列类，包括FIFO（先入先出)队列Queue，LIFO（后入先出）队列LifoQueue，和优先级队列 PriorityQueue。这些队列都实现了锁原语，能够在多线程中直接使用，可以使用队列来实现线程间的同步。
Queue 模块中的常用方法:

  • Queue.qsize() 返回队列的大小
  • Queue.empty() 如果队列为空，返回True,反之False
  • Queue.full() 如果队列满了，返回True,反之False
  • Queue.full 与 maxsize 大小对应
  • Queue.get([block[, timeout]])获取队列，timeout等待时间
  • Queue.get_nowait() 相当Queue.get(False)
  • Queue.put(item) 写入队列，timeout等待时间
  • Queue.put_nowait(item) 相当Queue.put(item, False)
  • Queue.task_done() 在完成一项工作之后，Queue.task_done()函数向任务已经完成的队列发送一个信号
  • Queue.join() 实际上意味着等到队列为空，再执行别的操作 ```python import queue import threading import time exitFlag = 0 class myThread (threading.Thread): def init(self, threadID, name, q):
    1.   threading.Thread.__init__(self)
    2.   self.threadID = threadID
    3.   self.name = name
    4.   self.q = q
    def run(self):
    1.   print ("开启线程：" + self.name)
    2.   process_data(self.name, self.q)
    3.   print ("退出线程：" + self.name)

def process_data(threadName, q): while not exitFlag: queueLock.acquire() if not workQueue.empty(): data = q.get() print (“%s processing %s” % (threadName, data)) queueLock.release() else: queueLock.release() time.sleep(1)

threadList = [“Thread-1”, “Thread-2”, “Thread-3”] nameList = [“One”, “Two”, “Three”, “Four”, “Five”] queueLock = threading.Lock()

先进先出队列

workQueue = queue.Queue(10) threads = [] threadID = 1

创建新线程

for tName in threadList: thread = myThread(threadID, tName, workQueue) thread.start() threads.append(thread) threadID += 1

for word in nameList: workQueue.put(word)

等待队列清空

while not workQueue.empty(): pass

通知线程是时候退出

exitFlag = 1

等待所有线程完成

for t in threads: t.join() print (“退出主线程”)

> 开启线程：Thread-1 开启线程：Thread-2 开启线程：Thread-3 Thread-3 processing One Thread-1 processing Two Thread-2 processing Three Thread-3 processing Four Thread-1 processing Five 退出线程：Thread-1 退出线程：Thread-2 退出线程：Thread-3 退出主线程 ```