[python]多线程快速入门

二十一木

前言

线程是操作系统能够进行运算调度的最小单位，它被包含在进程之中，是进程中的实际运作单位。由于CPython的GIL限制，多线程实际为单线程，大多只用来处理IO密集型任务。
Python一般用标准库threading来进行多线程编程。
基本使用

• 方式1，创建threading.Thread类的示例
  

1. import threading
2. import time
4. def task1(counter: int):
5.     print(f"thread: {threading.current_thread().name}, args: {counter}, start time: {time.strftime('%F %T')}")
6.     num = counter
7.     while num > 0:
8.         time.sleep(3)
9.         num -= 1
10.     print(f"thread: {threading.current_thread().name}, args: {counter}, end time: {time.strftime('%F %T')}")
12. if __name__ == "__main__":
13.     print(f"main thread: {threading.current_thread().name}, start time: {time.strftime('%F %T')}")
14.     # 创建三个线程
15.     t1 = threading.Thread(target=task1, args=(7,))
16.     t2 = threading.Thread(target=task1, args=(5,))
17.     t3 = threading.Thread(target=task1, args=(3,))
19.     # 启动线程
20.     t1.start()
21.     t2.start()
22.     t3.start()
24.     # join() 用于阻塞主线程, 等待子线程执行完毕
25.     t1.join()
26.     t2.join()
27.     t3.join()
28.     print(f"main thread: {threading.current_thread().name}, end time: {time.strftime('%F %T')}")

复制代码

执行输出示例

1. main thread: MainThread, start time: 2024-10-26 12:42:37
2. thread: Thread-1 (task1), args: 7, start time: 2024-10-26 12:42:37
3. thread: Thread-2 (task1), args: 5, start time: 2024-10-26 12:42:37
4. thread: Thread-3 (task1), args: 3, start time: 2024-10-26 12:42:37
5. thread: Thread-3 (task1), args: 3, end time: 2024-10-26 12:42:46
6. thread: Thread-2 (task1), args: 5, end time: 2024-10-26 12:42:52
7. thread: Thread-1 (task1), args: 7, end time: 2024-10-26 12:42:58
8. main thread: MainThread, end time: 2024-10-26 12:42:58

复制代码

• 方式2，继承threading.Thread类，重写run()和__init__()方法
  

1. import threading
2. import time
4. class MyThread(threading.Thread):
5.     def __init__(self, counter: int):
6.         super().__init__()
7.         self.counter = counter
9.     def run(self):
10.         print(f"thread: {threading.current_thread().name}, args: {self.counter}, start time: {time.strftime('%F %T')}")
11.         num = self.counter
12.         while num > 0:
13.             time.sleep(3)
14.             num -= 1
15.         print(f"thread: {threading.current_thread().name}, args: {self.counter}, end time: {time.strftime('%F %T')}")
17. if __name__ == "__main__":
18.     print(f"main thread: {threading.current_thread().name}, start time: {time.strftime('%F %T')}")
19.     # 创建三个线程
20.     t1 = MyThread(7)
21.     t2 = MyThread(5)
22.     t3 = MyThread(3)
24.     # 启动线程
25.     t1.start()
26.     t2.start()
27.     t3.start()
29.     # join() 用于阻塞主线程, 等待子线程执行完毕
30.     t1.join()
31.     t2.join()
32.     t3.join()
33.     print(f"main thread: {threading.current_thread().name}, end time: {time.strftime('%F %T')}")

复制代码

继承threading.Thread类也可以写成这样，调用外部函数。

1. import threading
2. import time
4. def task1(counter: int):
5.     print(f"thread: {threading.current_thread().name}, args: {counter}, start time: {time.strftime('%F %T')}")
6.     num = counter
7.     while num > 0:
8.         time.sleep(3)
9.         num -= 1
10.     print(f"thread: {threading.current_thread().name}, args: {counter}, end time: {time.strftime('%F %T')}")
12. class MyThread(threading.Thread):
13.     def __init__(self, target, args: tuple):
14.         super().__init__()
15.         self.target = target
16.         self.args = args
17.    
18.     def run(self):
19.         self.target(*self.args)
21. if __name__ == "__main__":
22.     print(f"main thread: {threading.current_thread().name}, start time: {time.strftime('%F %T')}")
23.     # 创建三个线程
24.     t1 = MyThread(target=task1, args=(7,))
25.     t2 = MyThread(target=task1, args=(5,))
26.     t3 = MyThread(target=task1, args=(3,))
28.     # 启动线程
29.     t1.start()
30.     t2.start()
31.     t3.start()
33.     # join() 用于阻塞主线程, 等待子线程执行完毕
34.     t1.join()
35.     t2.join()
36.     t3.join()
37.     print(f"main thread: {threading.current_thread().name}, end time: {time.strftime('%F %T')}")

复制代码

多线程同步

如果多个线程共同对某个数据修改，则可能出现不可预料的后果，这时候就需要某些同步机制。比如如下代码，结果是随机的（个人电脑用python3.13实测结果都是0，而低版本的python3.6运行结果的确是随机的）

1. import threading
2. import time
4. num = 0
6. def task1(counter: int):
7.     print(f"thread: {threading.current_thread().name}, args: {counter}, start time: {time.strftime('%F %T')}")
8.     global num
9.     for _ in range(100000000):
10.         num = num + counter
11.         num = num - counter
12.     print(f"thread: {threading.current_thread().name}, args: {counter}, end time: {time.strftime('%F %T')}")
14. if __name__ == "__main__":
15.     print(f"main thread: {threading.current_thread().name}, start time: {time.strftime('%F %T')}")
16.     # 创建三个线程
17.     t1 = threading.Thread(target=task1, args=(7,))
18.     t2 = threading.Thread(target=task1, args=(5,))
19.     t3 = threading.Thread(target=task1, args=(3,))
20.     t4 = threading.Thread(target=task1, args=(6,))
21.     t5 = threading.Thread(target=task1, args=(8,))
23.     # 启动线程
24.     t1.start()
25.     t2.start()
26.     t3.start()
27.     t4.start()
28.     t5.start()
30.     # join() 用于阻塞主线程, 等待子线程执行完毕
31.     t1.join()
32.     t2.join()
33.     t3.join()
34.     t4.join()
35.     t5.join()
36.    
37.     print(f"num: {num}")
38.     print(f"main thread: {threading.current_thread().name}, end time: {time.strftime('%F %T')}")

复制代码

Lock-锁

使用互斥锁可以在一个线程访问数据时，拒绝其它线程访问，直到解锁。threading.Thread中的Lock()和Rlock()可以提供锁功能。

1. import threading
2. import time
4. num = 0
6. mutex = threading.Lock()
8. def task1(counter: int):
9.     print(f"thread: {threading.current_thread().name}, args: {counter}, start time: {time.strftime('%F %T')}")
10.     global num
11.     mutex.acquire()
12.     for _ in range(100000):
13.         num = num + counter
14.         num = num - counter
15.     mutex.release()
16.     print(f"thread: {threading.current_thread().name}, args: {counter}, end time: {time.strftime('%F %T')}")
18. if __name__ == "__main__":
19.     print(f"main thread: {threading.current_thread().name}, start time: {time.strftime('%F %T')}")
20.     # 创建三个线程
21.     t1 = threading.Thread(target=task1, args=(7,))
22.     t2 = threading.Thread(target=task1, args=(5,))
23.     t3 = threading.Thread(target=task1, args=(3,))
25.     # 启动线程
26.     t1.start()
27.     t2.start()
28.     t3.start()
30.     # join() 用于阻塞主线程, 等待子线程执行完毕
31.     t1.join()
32.     t2.join()
33.     t3.join()
34.    
35.     print(f"num: {num}")
36.     print(f"main thread: {threading.current_thread().name}, end time: {time.strftime('%F %T')}")

复制代码

Semaphore-信号量

互斥锁是只允许一个线程访问共享数据，而信号量是同时允许一定数量的线程访问共享数据。比如银行有5个窗口，允许同时有5个人办理业务，后面的人只能等待，待柜台有空闲才可以进入。

1. import threading
2. import time
3. from random import randint
5. semaphore = threading.BoundedSemaphore(5)
7. def business(name: str):
8.     semaphore.acquire()
9.     print(f"{time.strftime('%F %T')} {name} is handling")
10.     time.sleep(randint(3, 10))
11.     print(f"{time.strftime('%F %T')} {name} is done")
12.     semaphore.release()
14. if __name__ == "__main__":
15.     print(f"main thread: {threading.current_thread().name}, start time: {time.strftime('%F %T')}")
16.     threads = []
17.     for i in range(10):
18.         t = threading.Thread(target=business, args=(f"thread-{i}",))
19.         threads.append(t)
21.     for t in threads:
22.         t.start()
24.     for t in threads:
25.         t.join()
26.    
27.     print(f"main thread: {threading.current_thread().name}, end time: {time.strftime('%F %T')}")

复制代码

执行输出

1. main thread: MainThread, start time: 2024-10-26 17:40:10
2. 2024-10-26 17:40:10 thread-0 is handling
3. 2024-10-26 17:40:10 thread-1 is handling
4. 2024-10-26 17:40:10 thread-2 is handling
5. 2024-10-26 17:40:10 thread-3 is handling
6. 2024-10-26 17:40:10 thread-4 is handling
7. 2024-10-26 17:40:15 thread-2 is done
8. 2024-10-26 17:40:15 thread-5 is handling
9. 2024-10-26 17:40:16 thread-0 is done
10. 2024-10-26 17:40:16 thread-6 is handling
11. 2024-10-26 17:40:19 thread-3 is done
12. 2024-10-26 17:40:19 thread-4 is done
13. 2024-10-26 17:40:19 thread-7 is handling
14. 2024-10-26 17:40:19 thread-8 is handling
15. 2024-10-26 17:40:20 thread-1 is done
16. 2024-10-26 17:40:20 thread-9 is handling
17. 2024-10-26 17:40:21 thread-6 is done
18. 2024-10-26 17:40:23 thread-7 is done
19. 2024-10-26 17:40:24 thread-5 is done
20. 2024-10-26 17:40:24 thread-8 is done
21. 2024-10-26 17:40:30 thread-9 is done
22. main thread: MainThread, end time: 2024-10-26 17:40:30

复制代码

Condition-条件对象

Condition对象能让一个线程A停下来，等待其他线程，其他线程通知后线程A继续运行。

1. import threading
2. import time
3. import random
5. class Employee(threading.Thread):
6.     def __init__(self, username: str, cond: threading.Condition):
7.         self.username = username
8.         self.cond = cond
9.         super().__init__()
11.     def run(self):
12.         with self.cond:
13.             print(f"{time.strftime('%F %T')} {self.username} 到达公司")
14.             self.cond.wait()  # 等待通知
15.             print(f"{time.strftime('%F %T')} {self.username} 开始工作")
16.             time.sleep(random.randint(1, 5))
17.             print(f"{time.strftime('%F %T')} {self.username} 工作完成")
19. class Boss(threading.Thread):
20.     def __init__(self, username: str, cond: threading.Condition):
21.         self.username = username
22.         self.cond = cond
23.         super().__init__()
25.     def run(self):
26.         with self.cond:
27.             print(f"{time.strftime('%F %T')} {self.username} 发出通知")
28.             self.cond.notify_all()  # 通知所有线程
29.         time.sleep(2)
31. if __name__ == "__main__":
32.     cond = threading.Condition()
33.     boss = Boss("老王", cond)
34.    
35.     employees = []
36.     for i in range(5):
37.         employees.append(Employee(f"员工{i}", cond))
39.     for employee in employees:
40.         employee.start()
41.     boss.start()
43.     boss.join()
44.     for employee in employees:
45.         employee.join()

复制代码

执行输出

1. 2024-10-26 21:16:20 员工0 到达公司
2. 2024-10-26 21:16:20 员工1 到达公司
3. 2024-10-26 21:16:20 员工2 到达公司
4. 2024-10-26 21:16:20 员工3 到达公司
5. 2024-10-26 21:16:20 员工4 到达公司
6. 2024-10-26 21:16:20 老王 发出通知
7. 2024-10-26 21:16:20 员工4 开始工作
8. 2024-10-26 21:16:23 员工4 工作完成
9. 2024-10-26 21:16:23 员工1 开始工作
10. 2024-10-26 21:16:28 员工1 工作完成
11. 2024-10-26 21:16:28 员工2 开始工作
12. 2024-10-26 21:16:30 员工2 工作完成
13. 2024-10-26 21:16:30 员工0 开始工作
14. 2024-10-26 21:16:31 员工0 工作完成
15. 2024-10-26 21:16:31 员工3 开始工作
16. 2024-10-26 21:16:32 员工3 工作完成

复制代码

Event-事件

在 Python 的 threading 模块中，Event 是一个线程同步原语，用于在多个线程之间进行简单的通信。Event 对象维护一个内部标志，线程可以使用 wait() 方法阻塞，直到另一个线程调用 set() 方法将标志设置为 True。一旦标志被设置为 True，所有等待的线程将被唤醒并继续执行。
Event 的主要方法

• set()：将事件的内部标志设置为 True，并唤醒所有等待的线程。
  
• clear()：将事件的内部标志设置为 False。
  
• is_set()：返回事件的内部标志是否为 True。
  
• wait(timeout=None)：如果事件的内部标志为 False，则阻塞当前线程，直到标志被设置为 True 或超时（如果指定了 timeout）。
  

1. import threading
2. import time
3. import random
5. class Employee(threading.Thread):
6.     def __init__(self, username: str, cond: threading.Event):
7.         self.username = username
8.         self.cond = cond
9.         super().__init__()
11.     def run(self):
12.         print(f"{time.strftime('%F %T')} {self.username} 到达公司")
13.         self.cond.wait()  # 等待事件标志为True
14.         print(f"{time.strftime('%F %T')} {self.username} 开始工作")
15.         time.sleep(random.randint(1, 5))
16.         print(f"{time.strftime('%F %T')} {self.username} 工作完成")
18. class Boss(threading.Thread):
19.     def __init__(self, username: str, cond: threading.Event):
20.         self.username = username
21.         self.cond = cond
22.         super().__init__()
24.     def run(self):
25.         print(f"{time.strftime('%F %T')} {self.username} 发出通知")
26.         self.cond.set()
28. if __name__ == "__main__":
29.     cond = threading.Event()
30.     boss = Boss("老王", cond)
31.    
32.     employees = []
33.     for i in range(5):
34.         employees.append(Employee(f"员工{i}", cond))
36.     for employee in employees:
37.         employee.start()
38.     boss.start()
40.     boss.join()
41.     for employee in employees:
42.         employee.join()

复制代码

执行输出

1. 2024-10-26 21:22:28 员工0 到达公司
2. 2024-10-26 21:22:28 员工1 到达公司
3. 2024-10-26 21:22:28 员工2 到达公司
4. 2024-10-26 21:22:28 员工3 到达公司
5. 2024-10-26 21:22:28 员工4 到达公司
6. 2024-10-26 21:22:28 老王 发出通知
7. 2024-10-26 21:22:28 员工0 开始工作
8. 2024-10-26 21:22:28 员工1 开始工作
9. 2024-10-26 21:22:28 员工3 开始工作
10. 2024-10-26 21:22:28 员工4 开始工作
11. 2024-10-26 21:22:28 员工2 开始工作
12. 2024-10-26 21:22:30 员工3 工作完成
13. 2024-10-26 21:22:31 员工4 工作完成
14. 2024-10-26 21:22:31 员工2 工作完成
15. 2024-10-26 21:22:32 员工0 工作完成
16. 2024-10-26 21:22:32 员工1 工作完成

复制代码

使用队列

Python的queue模块提供同步、线程安全的队列类。以下示例为使用queue实现的生产消费者模型

1. import threading
2. import time
3. import random
4. import queue
7. class Producer(threading.Thread):
8.     """多线程生产者类."""
10.     def __init__(
11.         self, tname: str, channel: queue.Queue, done: threading.Event
12.     ):
13.         self.tname = tname
14.         self.channel = channel
15.         self.done = done
16.         super().__init__()
18.     def run(self) -> None:
19.         """Method representing the thread's activity."""
21.         while True:
22.             if self.done.is_set():
23.                 print(
24.                     f"{time.strftime('%F %T')} {self.tname} 收到停止信号事件"
25.                 )
26.                 break
27.             if self.channel.full():
28.                 print(
29.                     f"{time.strftime('%F %T')} {self.tname} report: 队列已满, 全部停止生产"
30.                 )
31.                 self.done.set()
32.             else:
33.                 num = random.randint(100, 1000)
34.                 self.channel.put(f"{self.tname}-{num}")
35.                 print(
36.                     f"{time.strftime('%F %T')} {self.tname} 生成数据 {num}, queue size: {self.channel.qsize()}"
37.                 )
38.                 time.sleep(random.randint(1, 5))
41. class Consumer(threading.Thread):
42.     """多线程消费者类."""
44.     def __init__(
45.         self, tname: str, channel: queue.Queue, done: threading.Event
46.     ):
47.         self.tname = tname
48.         self.channel = channel
49.         self.done = done
50.         self.counter = 0
51.         super().__init__()
53.     def run(self) -> None:
54.         """Method representing the thread's activity."""
55.         while True:
56.             if self.done.is_set():
57.                 print(
58.                     f"{time.strftime('%F %T')} {self.tname} 收到停止信号事件"
59.                 )
60.                 break
61.             if self.counter >= 3:
62.                 print(
63.                     f"{time.strftime('%F %T')} {self.tname} report: 全部停止消费"
64.                 )
65.                 self.done.set()
66.                 continue
67.             if self.channel.empty():
68.                 print(
69.                     f"{time.strftime('%F %T')} {self.tname} report: 队列为空, counter: {self.counter}"
70.                 )
71.                 self.counter += 1
72.                 time.sleep(1)
73.                 continue
74.             else:
75.                 data = self.channel.get()
76.                 print(
77.                     f"{time.strftime('%F %T')} {self.tname} 消费数据 {data}, queue size: {self.channel.qsize()}"
78.                 )
79.                 time.sleep(random.randint(1, 5))
80.                 self.counter = 0
83. if __name__ == "__main__":
84.     done_p = threading.Event()
85.     done_c = threading.Event()
86.     channel = queue.Queue(30)
87.     threads_producer = []
88.     threads_consumer = []
90.     for i in range(8):
91.         threads_producer.append(Producer(f"producer-{i}", channel, done_p))
93.     for i in range(6):
94.         threads_consumer.append(Consumer(f"consumer-{i}", channel, done_c))
96.     for t in threads_producer:
97.         t.start()
99.     for t in threads_consumer:
100.         t.start()
102.     for t in threads_producer:
103.         t.join()
105.     for t in threads_consumer:
106.         t.join()

复制代码

线程池

在面向对象编程中，创建和销毁对象是很费时间的，因为创建一个对象要获取内存资源或其他更多资源。在多线程程序中，生成一个新线程之后销毁，然后再创建一个，这种方式就很低效。池化多线程，也就是线程池就为此而生。
将任务添加到线程池中，线程池会自动指定一个空闲的线程去执行任务，当超过最大线程数时，任务需要等待有新的空闲线程才会被执行。Python一般可以使用multiprocessing模块中的Pool来创建线程池。

1. import time
3. from multiprocessing.dummy import Pool as ThreadPool
5. def foo(n):
6.     time.sleep(2)
9. if __name__ == "__main__":
10.     start = time.time()
11.     for n in range(5):
12.         foo(n)
13.     print("single thread time: ", time.time() - start)
15.     start = time.time()
16.     t_pool = ThreadPool(processes=5)  # 创建线程池, 指定池中的线程数为5(默认为CPU数)
17.     rst = t_pool.map(foo, range(5))  # 使用map为每个元素应用到foo函数
18.     t_pool.close()  # 阻止任何新的任务提交到线程池
19.     t_pool.join()  # 等待所有已提交的任务完成
20.     print("thread pool time: ", time.time() - start)

复制代码

线程池执行器

python的内置模块concurrent.futures提供了ThreadPoolExecutor类。这个类结合了线程和队列的优势，可以用来平行执行任务。

1. import time
2. from random import randint
3. from concurrent.futures import ThreadPoolExecutor
5. def foo() -> None:
6.     time.sleep(2)
7.     return randint(1,100)
9. if __name__ == "__main__":
10.     start = time.time()
11.     futures = []
12.     with ThreadPoolExecutor(max_workers=5) as executor:
13.         for n in range(10):
14.             futures.append(executor.submit(foo))  # Fan out
15.             
16.     for future in futures:  # Fan in
17.         print(future.result())
18.     print("thread pool executor time: ", time.time() - start)

复制代码

执行输出

1. 44
2. 19
3. 86
4. 48
5. 35
6. 74
7. 59
8. 99
9. 58
10. 53
11. thread pool executor time:  4.001955032348633

复制代码

ThreadPoolExecutor类的最大优点在于：如果调用者通过submit方法把某项任务交给它执行，那么会获得一个与该任务相对应的Future实例，当调用者在这个实例上通过result方法获取执行结果时，ThreadPoolExecutor会把它在执行任务的过程中所遇到的异常自动抛给调用者。而ThreadPoolExecutor类的缺点是IO并行能力不高，即便把max_worker设为100，也无法高效处理任务。更高需求的IO任务可以考虑换异步协程方案。
参考

• 郑征《Python自动化运维快速入门》清华大学出版社
  
• Brett Slatkin《Effective Python》(2nd) 机械工业出版社
  

来源:https://www.cnblogs.com/XY-Heruo/p/18514316
免责声明：由于采集信息均来自互联网，如果侵犯了您的权益，请联系我们【E-Mail:cb@itdo.tech】 我们会及时删除侵权内容，谢谢合作！