• Python 多线程、线程池、进程池

    线程间的通讯机制

    消息队列

    event 事件对象

    当线程创建完成之后，并不会马上执行线程，而是等待某一事件发生，线程才会启动

    1. import threading
    2.  
    3. # # 创建 event 对象
    4. # event = threading.Event()
    5. # # 重置代码中的 event 对象，使得所有该event事件都处于待命状态
    6. # event.clear()
    7. # # 阻塞线程，等待 event 指令
    8. # event.wait()
    9. # # 发送 event 指令，使得所有设置该 event 事件的线程执行
    10. # event.set()
    11.  
    12. class MyThreading(threading.Thread):
    13.     def __init__(self, event):
    14.         super().__init__()
    15.         self.event = event
    16.  
    17.     def run(self):
    18.         print('线程{}已经初始化完成，随时准备启动...'.format(self.name))
    19.         # 阻塞线程，让线程等待指令后再启动
    20.         self.event.wait()
    21.         print('{}开始执行...'.format(self.name))
    22.  
    23. if __name__ == '__main__':
    24.     event = threading.Event()
    25.     # 创建 10 个自定义线程对象并放入列表
    26.     threads = [MyThreading(event) for i in range(10)]
    27.  
    28.     # 重置代码中的 event 对象，使得所有该event事件都处于待命状态
    29.     event.clear()
    30.  
    31.     # 执行线程
    32.     # 执行到 run 方法中 self.event.wait() 位置，即打印了：线程{}已经初始化完成...
    33.     [t.start() for t in threads]
    34.  
    35.     # 发送 event 指令，使得所有设置该 event 事件的线程执行
    36.     # 即启动 threads 列表中的所有线程
    37.     # 接着执行 run 方法中 self.event.wait() 后面的代码，即打印了：{}开始执行...
    38.     event.set()
    39.  
    40.     [t.join() for t in threads]

    condition 条件对象

    1. import threading
    2.  
    3. # condition 对象适用于线程轮流执行，或一个线程等待另一个线程的情况，如两个人的对话等
    4.  
    5. # 创建 condition 对象
    6. cond = threading.Condition()
    7.  
    8. class ThreadA(threading.Thread):
    9.     def __init__(self, cond, name):
    10.         super().__init__(name=name)
    11.         self.cond = cond
    12.  
    13.     def run(self):
    14.         # 获取锁
    15.         self.cond.acquire()
    16.         # 线程A说了第一句话
    17.         print(self.getName(), '：一二三四五')
    18.         # 唤醒其他处于 wait 状态的线程（通知线程B可以说话了）
    19.         self.cond.notify()
    20.         # 线程A进入 wait 状态，等待线程B通知（唤醒）
    21.         self.cond.wait()
    22.  
    23.         # 被线程A唤醒后说了第二句话
    24.         print(self.name, '：山无棱，天地合，乃敢与君绝')
    25.         self.cond.notify()  # 通知线程B
    26.         self.cond.wait()  # 等待线程B通知
    27.  
    28.         # 被线程A唤醒后说了第三句话，最后一句话
    29.         print(self.name, '：有钱吗？借点')
    30.         self.cond.notify()  # 通知线程B
    31.         self.cond.release()  # 释放锁
    32.  
    33.  
    34. class ThreadB(threading.Thread):
    35.     def __init__(self, cond, name):
    36.         super().__init__(name=name)
    37.         self.cond = cond
    38.  
    39.     def run(self):
    40.         # 获取锁
    41.         self.cond.acquire()
    42.         self.cond.wait()  # 由于它不是第一个说话的人，所以一开始等待通知
    43.  
    44.         # 线程B说了第一句话
    45.         print(self.getName(), '：上山打老虎')
    46.         # 唤醒其他处于 wait 状态的线程（通知线程A可以说话了）
    47.         self.cond.notify()
    48.         # 线程B进入 wait 状态，等待线程A通知（唤醒）
    49.         self.cond.wait()
    50.  
    51.         # 被线程B唤醒后说了第二句话
    52.         print(self.name, '：海可枯，石可烂，激情永不散')
    53.         self.cond.notify()  # 通知线程A
    54.         self.cond.wait()  # 等待线程A通知
    55.  
    56.         # 被线程B唤醒后说了第三句话，最后一句话
    57.         print(self.name, '：没有，滚')
    58.         # self.cond.notify()  # 已经是最后一句话，不需要通知线程A
    59.         self.cond.release()  # 释放锁
    60.  
    61. if __name__ == '__main__':
    62.     a = ThreadA(cond, 'AAA')
    63.     b = ThreadB(cond, 'BBB')
    64.  
    65.     # 线程A先说话，但是不能先启动线程A
    66.     # 因为如果启动了线程A，然后线程A说完第一句话后，通知线程B
    67.     # 但是此时线程B没有启动，就接收不了A的通知，B就会一直处于 wait 状态，即说不了话，也通知不了A
    68.     # A等不到B的通知，也会一直处于 wait 状态
    69.     # a.start()
    70.     # b.start()
    71.     
    72.     b.start()
    73.     a.start()

    线程间的消息隔离机制

    使用场景

    在使用多线程的过程中，会有一种变量的使用场景： 一个变量会被所有的线程使用，但是每个线程都会对该变量设置不同的值， threading.local() 提供了这种变量

    使用方法

    1. """
    2. 在使用多线程的过程中，会有一种变量的使用场景：
    3.     一个变量会被所有的线程使用，但是每个线程都会对该变量设置不同的值
    4.     threading.local() 提供了这种变量
    6. 假设有一个场景：
    7.     设置一个 threading.local 变量，然后新建两个线程
    8.     分别设置这两个 threading.local 的值
    9.     再分别打印这两个 threading.local 的值
    10.     看每个线程打印出来的 threading.local 值是否不一样
    11. """
    12. import threading
    13.  
    14. # local_data 实际上是一个对象
    15. local_data = threading.local()
    16. # 设置 local_data 的名字
    17. local_data.name = 'local_data'
    18.  
    19.  
    20. class MyThread(threading.Thread):
    21.     def run(self):
    22.         print('赋值前-子线程：', threading.currentThread(), local_data.__dict__)
    23.         # 在子线程中修改 local_data.name 的值
    24.         local_data.name = self.getName()
    25.         print('赋值后-子线程：', threading.currentThread(), local_data.__dict__)
    26.  
    27.  
    28. if __name__ == '__main__':
    29.     print('开始前-主线程：', local_data.__dict__)
    30.  
    31.     t1 = MyThread()
    32.     t1.start()
    33.     t1.join()
    34.  
    35.     t2 = MyThread()
    36.     t2.start()
    37.     t2.join()
    38.     
    39.     print('结束后-主线程：', local_data.__dict__)
    40.     
    41. """
    42. 输出结果：
    43. 开始前-主线程： {'name': 'local_data'}
    44. 赋值前-子线程：  {}
    45. 赋值后-子线程：  {'name': 'Thread-1'}
    46. 赋值前-子线程：  {}
    47. 赋值后-子线程：  {'name': 'Thread-2'}
    48. 结束后-主线程： {'name': 'local_data'}
    49. """

    线程池

    线程池中存放多个线程，当有业务需要线程来执行时，可以直接从线程池中获取一个线程来执行该业务， 业务执行完毕之后，线程不会释放，而是被放回线程池中，从而节省了线程的创建以及销毁的时间。 Python concurrent.futures 模块中的 ThreadPoolExecutor 就提供了线程池，该线程池有以下特点：

    • 主线程可以获取某一个线程或任务的状态，以及返回值
    • 当一个线程完成的时候，主线程能够立即知道
    • 让多线程和多进程的编码接口一致

    线程池的简单应用

    1. from concurrent.futures import ThreadPoolExecutor
    2. import time
    3.  
    4. # 创建线程池对象，并指定线程池中最大的线程数为 3
    5. # 当业务数不超过 3 的时候，ThreadPoolExecutor 就会创建一个新的线程来执行业务
    6. # 当超过 3 时，ThreadPoolExecutor 不会创建新的线程，而是等待执行其他业务的线程执行完毕后返回
    7. # 再将返回的线程分配给需要的业务
    8. executor = ThreadPoolExecutor(max_workers=3)
    9.  
    10. # 定义一个业务
    11. # 假设这里模拟一个爬虫，爬取一个网页页面
    12. def get_html(timers):
    13.     time.sleep(timers)  # 模拟耗时操作
    14.     print('获取网页信息{}完毕'.format(timers))
    15.     return timers
    16.  
    17. # 提交要执行的函数，即要完成的业务到线程池中，然后线程池就会自动分配线程去完成对应的业务
    18. # submit 方法会立即返回，不会阻塞主线程
    19. # get_html 的参数放在后面，即 1 会作为参数传递给 get_html() 中的 timers
    20. # 以下创建了四个任务
    21. task1 = executor.submit(get_html, 1)
    22. task2 = executor.submit(get_html, 2)
    23. task3 = executor.submit(get_html, 3)
    24. task4 = executor.submit(get_html, 4)
    25.  
    26. bool1 = task1.done()  # 检查任务是否完成，完成返回 True
    27. bool2 = task2.cancel()  # 取消任务执行，只有该任务没有被放入线程池中才能取消成功，成功返回 True
    28.  
    29. # 拿到任务执行的结果，如 get_html 的返回值
    30. # timeout 参数用于设置等待结果的最长等待时间，单位为秒
    31. # result 方法是一个阻塞方法
    32. timers = task3.result(timeout=10)
    33. print(timers)
    34. print(111)

    线程池中常用的方法

    • as_complete
    1. # 线程池的简单应用
    2. from concurrent.futures import ThreadPoolExecutor, as_completed
    3. import time
    4.  
    5. # 创建线程池对象，并指定线程池中最大的线程数为 3
    6. # 当业务数不超过 3 的时候，ThreadPoolExecutor 就会创建一个新的线程来执行业务
    7. # 当超过 3 时，ThreadPoolExecutor 不会创建新的线程，而是等待执行其他业务的线程执行完毕后返回
    8. # 再将返回的线程分配给需要的业务
    9. executor = ThreadPoolExecutor(max_workers=3)
    10.  
    11. # 定义一个业务
    12. # 假设这里模拟一个爬虫，爬取一个网页页面
    13. def get_html(timers):
    14.     time.sleep(timers)  # 模拟耗时操作
    15.     print('获取网页信息{}完毕'.format(timers))
    16.     return timers
    17.  
    18. # 模拟要爬取的 url
    19. urls = [1, 2, 3]
    20. # 通过列表推导式构造多线程任务
    21. all_tasks = [executor.submit(get_html, url) for url in urls]
    22. # as_completed 接收一个可迭代对象
    23. # as_completed 是一个生成器，当任务没有完成时，它会阻塞，只有当任务结束返回结果时才会继续往下执行
    24. # as_completed 函数的作用：拿到所有任务执行完毕之后的结果
    25. # 不需要我们手动调用 done 方法不停地判断任务是否完成
    26. for item in as_completed(all_tasks):
    27.     data = item.result()
    28.     print('主线程中获取任务的返回值是{}'.format(data))
    29. """
    30. 执行结果：
    31. 获取网页信息1完毕
    32. 主线程中获取任务的返回值是1
    33. 获取网页信息2完毕
    34. 主线程中获取任务的返回值是2
    35. 获取网页信息3完毕
    36. 主线程中获取任务的返回值是3
    37. """
    • map
    1. from concurrent.futures import ThreadPoolExecutor
    2. import time
    3.  
    4. # 创建线程池对象，并指定线程池中最大的线程数为 3
    5. # 当业务数不超过 3 的时候，ThreadPoolExecutor 就会创建一个新的线程来执行业务
    6. # 当超过 3 时，ThreadPoolExecutor 不会创建新的线程，而是等待执行其他业务的线程执行完毕后返回
    7. # 再将返回的线程分配给需要的业务
    8. executor = ThreadPoolExecutor(max_workers=3)
    9.  
    10. # 定义一个业务
    11. # 假设这里模拟一个爬虫，爬取一个网页页面
    12. def get_html(timers):
    13.     time.sleep(timers)  # 模拟耗时操作
    14.     print('获取网页信息{}完毕'.format(timers))
    15.     return timers
    16.  
    17. # 模拟要爬取的 url
    18. urls = [4, 2, 3]
    19.  
    20.  
    21. # map 方法和 as_complete 类似
    22. # map 也是一个生成器，当任务没有完成时，它会阻塞，只有当任务结束返回结果时才会继续往下执行
    23. # map 会自动映射 urls 中的每一个元素传递给 get_html 函数，并自动提交 ，不需要通过 submit 方法提交任务
    24. # map 方法直接拿到任务执行的结果
    25. # as_complete 和 map 都可以拿到线程池中各个线程执行的结果，但有以下区别：
    26. # as_complete 会根据任务完成的快慢得到结果，即哪个任务先完成就会先得到该任务的结果
    27. # 而 map 会严格按照任务的顺序得到结果，比如按照 urls 列表中的映射顺序得到对应的结果
    28. # 所以两种适用于不同的场景
    29. for data in executor.map(get_html, urls):
    30.     print('主线程中获取任务的返回值是{}'.format(data))
    31. """
    32. 获取网页信息2完毕
    33. 获取网页信息3完毕
    34. 获取网页信息4完毕
    35. 主线程中获取任务的返回值是4
    36. 主线程中获取任务的返回值是2
    37. 主线程中获取任务的返回值是3
    38. """
    • wait
    1. from concurrent.futures import ThreadPoolExecutor, wait, ALL_COMPLETED, FIRST_COMPLETED
    2. import time
    3.  
    4. # 创建线程池对象，并指定线程池中最大的线程数为 3
    5. # 当业务数不超过 3 的时候，ThreadPoolExecutor 就会创建一个新的线程来执行业务
    6. # 当超过 3 时，ThreadPoolExecutor 不会创建新的线程，而是等待执行其他业务的线程执行完毕后返回
    7. # 再将返回的线程分配给需要的业务
    8. executor = ThreadPoolExecutor(max_workers=3)
    9.  
    10. # 定义一个业务
    11. # 假设这里模拟一个爬虫，爬取一个网页页面
    12. def get_html(timers):
    13.     time.sleep(timers)  # 模拟耗时操作
    14.     print('获取网页信息{}完毕'.format(timers))
    15.     return timers
    16.  
    17. # 模拟要爬取的 url
    18. urls = [4, 2, 3]
    19.  
    20. all_tasks = [executor.submit(get_html, url) for url in urls]
    21.  
    22. # 让主线程阻塞，直到参数里的条件成立
    23. # 根据 wait 函数的参数，条件成立的情况是：所有任务执行完毕
    24. # ALL_COMPLETED 表示所有任务都执行完成
    25. # 还有其他的参数，如 FIRST_COMPLETED 表示只要有一个任务完成就条件成立
    26. wait(all_tasks, return_when=FIRST_COMPLETED)
    27.  
    28. # 如果想等代码执行完毕之后再打印下列语句，可以使用 wait 语句
    29. print('代码执行完毕')

    进程池

    使用 concurrent.future 模块提供的 ProcessPoolExecutor 来实现进程池，用法和线程池完全一致，参考上述线程池的使用（建议使用该种方式使用进程池）

    下面是基于 Pool 类实现的进程池的使用

    1. import multiprocessing
    2. import time
    3.  
    4. # 定义一个业务
    5. # 假设这里模拟一个爬虫，爬取一个网页页面
    6. def get_html(n):
    7.     time.sleep(n)  # 模拟耗时操作
    8.     print('子进程{}获取内容成功'.format(n))
    9.     return n
    10.  
    11. if __name__ == '__main__':
    12.     # 设置进程数，一般设置为和 CPU 数量一致的比较合理
    13.     # multiprocessing.cpu_count() 获取当前主机的 CPU 核心数
    14.     pool = multiprocessing.Pool(multiprocessing.cpu_count())
    15.  
    16.     # apply_async 是一个异步方法
    17.     # apply 是一个同步方法
    18.     # 作用类似于 submit 方法
    19.     result = pool.apply_async(get_html, args=(2,))
    20.  
    21.     pool.close()  # 必须在 join 方法前调用，否则会抛出异常
    22.     # join 方法会等待所有的子进程执行完毕之后，才会继续往下执行主进程的代码
    23.     # 即 join 会阻塞主进程代码
    24.     pool.join()
    25.  
    26.     # result.get() 拿到子进程执行结果， get 方法是一个阻塞方法
    27.     print(result.get())
    28.     print('end...')
    29.     print()
    30.     
    31.     # map 方法的使用
    32.     pool = multiprocessing.Pool(multiprocessing.cpu_count())
    33.     # imap 方法会按照列表顺序输出
    34.     # imap_unordered 方法则不会按照列表顺序执行，而是按照任务执行的快慢输出
    35.     for result in pool.imap(get_html, [1, 2, 3]):
    36.         print('{}休眠执行成功'.format(result))

    线程同步信号量（semaphore）的使用

    同步信号量的作用是用于控制同时工作的线程数量，如读文件时只能同时允许两个线程读，在爬虫时控制同时爬虫的线程，防止触发网站反扒机制

    1. import threading
    2. import time
    3.  
    4. # 还是模拟一个爬虫
    5. # HtmlSpider 类负责根据给定的 URL 去爬取网页内容
    6. class HtmlSpider(threading.Thread):
    7.     def __init__(self, url, sem):
    8.         super().__init__()
    9.         self.url = url
    10.         self.sem = sem
    11.  
    12.     def run(self):
    13.         time.sleep(2)
    14.         print('网页内容获取完成')
    15.         self.sem.release()  # 线程完成任务，释放锁
    16.  
    17. # UrlProducer类负责给 HtmlSpider 类提供网页的 URL
    18. class UrlProducer(threading.Thread):
    19.     def __init__(self, sem):
    20.         super().__init__()
    21.         self.sem = sem
    22.  
    23.     def run(self):
    24.         for i in range(10):
    25.             self.sem.acquire()  # 获取锁，获取成功才能执行线程
    26.             html_thread = HtmlSpider('url{}'.format(i), self.sem)  # 创建HtmlSpider线程
    27.             html_thread.start()  # 启动线程
    28.  
    29. if __name__ == '__main__':
    30.     # 创建线程同步信号量
    31.     # 参数 value 指定允许同时工作的线程数
    32.     sem = threading.Semaphore(value=3)
    33.     url_producer = UrlProducer(sem)
    34.     url_producer.start()

  • 相关阅读:
    MySQL中的大表优化方案
    搭建自己的SSR
    朋友圈里都是中顶会的大神，是一种什么样的体验？
    Android12源码编译报错ninja: build stopped: subcommand failed.解决
    ubuntu utopic unicorn static ip
    【英雄哥六月集训】第 21天: 堆(优先队列)
    3、云原生安全之falco的部署
    【学习笔记】[AGC058D] Yet Another ABC String
    Volcano 原理、源码分析（二）
    javaweb实现的在线鲜花商城源码(电商购物系统)
  • 原文地址：https://blog.csdn.net/2301_77659011/article/details/132830166