Python多线程编程指南：掌握线程基础与threading模块的使用方法
[id_4406[id_1480[id_1262558437]9195]34]
[id_1847450207]
使用 threading 模块时，能够通过创建一个新的 threading.Thread 实例，然后为其分配一个函数，以此来启动一个新的执行线程。
import threading
def foo():
print("Hello threading!")
创建了一个线程对象 my_thread，该线程的目标函数是 footarget 参数所引用的是要运行的函数（或者可调用的对象）。线程在调用 start 方法之前不会开始执行。
启动线程
my_thread.start() # 打印 'Hello threading!'现在，my_thread 已经开始运行并且已经终止了。如果再次调用 start 这个方法，就会引发一个 RuntimeError 异常。如果你希望让线程以守护线程的方式运行，那么在调用 start() 这个方法之前，可以通过传递 daemon=True 这个参数，或者将 my_thread 的 daemon 属性设置为 True，这样就能使线程在后台默默地运行。
线程同步（Joining a Thread）
在某些情形下，你把一个大任务拆分成诸多小任务，并且期望这些小任务能够并发运行，不过需要等到所有小任务都完成之后再接着进行，在这种时候可以运用 Thread.join() 方法。
假设你期望下载一个网站的多个页面并把它们编译成一个页面。你能够这样去做：
import requests
from threading import Thread
from queue import Queue
q = Queue(maxsize=20)
def put_page_to_q(page_num):
q.put(requests.get(某个网站的网址是“http://some-website.com/page_%s.html”。 % page_num))
def compile(q):
# 需要所有页面才能执行的神奇函数
if not q.full():
raise ValueError
else:
print("Done compiling!")
threads = []
for page_num in range(20):
t 是一个线程，其目标是执行 requests.get 函数，传入的参数是 page_num
t.start()
threads.append(t)
接下来，使用 join() 这个方法，以确保所有的线程都运行完成之后，再继续进行后续的操作。
join()是一个阻塞调用。除非在调用 join 时使用 blocking=False 参数，否则它是阻塞的。
for t in threads:
t.join()
# 现在所有线程已完成，调用 compile()
compile(q)关于 join() 的工作原理，可以参考这里。
创建自定义线程类
我们可以使用 threading.Thread 类来创建一个新的自定义线程类。并且需要在子类中对 run 方法进行覆盖。
from threading import Thread
import time
class Sleepy(Thread):
def run(self):
time.sleep(5)
print("Hello form Thread")
if __name__ == "__main__":
t = Sleepy()
t.start() 语言描述：“start 方法会去调用 Thread 类的 run 方法。” 可以改写为：“start 方法会调用 Thread 类的 run 方法。” 或者：“start 方法会把调用 Thread 类的 run 方法作为自身的行为。” 又或者：“start 方法会引发对 Thread 类的 run 方法的调用。”
print 会输出“主程序在前台持续运行。”
t.join()
print()2: 线程间通信
你的代码中有多个线程，你需要在它们之间安全地通信。
可以使用 queue 模块中的 Queue。
from queue import Queue
from threading import Thread
import logging
import time
# 配置日志记录
设置日志的基本配置，将日志级别设置为 INFO，设置日志的格式为%(asctime)s ，这是一个时间相关的部分；%(levelname)s ，代表着日志的级别；%(message)s ，则是具体的消息内容。)
# 创建数据生产者
def producer(output_queue, exit_event):
while not exit_event.is_set():
data = data_computation()
output_queue.put(data)
logging.info(f"Produced data: {data}")
time.sleep(1) # 模拟数据生成的延迟
# 创建消费者
def consumer(input_queue, exit_event):
while not exit_event.is_set():
try:
# 阻塞地获取数据，超时时间为1秒
data 是从输入队列中获取的，获取的超时时间为1)
# 对数据进行处理
logging.info(f"Consumed data: {data}")
# 表示数据已被处理
input_queue.task_done()
except Queue.Empty:
logging.info(没有数据在队列中，正在等待更多数据。)
# 模拟数据生成函数
def data_computation():
return "data"
# 创建生产者和消费者线程，并共享一个队列
q = Queue()
创建了一个线程事件对象 exit_event 。
# 创建消费者线程
一个名为 consumer_thread 的线程，其目标是执行 consumer 函数，参数为 q 和 exit_event，并且该线程是守护线程。True)
# 创建生产者线程
生产者线程是由一个目标为 producer 的线程创建的，并且传入了参数 (q, exit_event) 。
consumer_thread.start()
producer_thread.start()
# 主程序运行一段时间后退出
time.sleep(10)
exit_event.set() # 设置退出标志
# 等待生产者线程退出
producer_thread.join()
# 等待队列中的所有任务完成
q.join()
# 消费者线程作为守护线程，会自动退出Section 3: 创建线程池
使用 threading 和 queue：
from socket import socket, AF_INET, SOCK_STREAM
from threading import Thread
from queue import Queue
def echo_server(addr, nworkers):
print('Echo server running at', addr)
# 启动客户端工作线程
q = Queue()
for n in range(nworkers):
t 是一个线程，其目标是执行 echo_client 函数，参数为 (q,) 。
t.daemon = True
t.start()
# 运行服务器
创建一个套接字，使用的地址族是 AF_INET，套接字类型是 SOCK_STREAM，将其赋值给 sock 变量。
sock.bind(addr)
sock.listen(5)
while True:
sock 接受连接，得到 client_sock 和 client_addr
将(client_sock, client_addr)放入 q 中。
echo_server(('',15000), 128)使用
一个用于并行执行任务的线程池执行器，它可以同时处理多个任务，通过提交任务到线程池中，线程池会自动分配线程来执行这些任务，从而提高程序的执行效率。
from socket import AF_INET, SOCK_STREAM, socket
from concurrent.futures import ThreadPoolExecutor
def echo_server(addr):
print('Echo server running at', addr)
pool = ThreadPoolExecutor(128)
sock = socket(AF_INET, SOCK_STREAM)
sock.bind(addr)
sock.listen(5)
while True:
client_sock, client_addr = sock.accept()
pool 把 echo_client、client_sock 和 client_addr 作为参数进行提交
echo_server(('',15000))《Python Cookbook》（第 3 版）的作者是 David Beazley 和 Brian K. Jones（O'Reilly）。版权归 David Beazley 和 Brian Jones 所有，其 ISBN 为 978-1-449-34037-7。
4: 多线程的高级用法
本节将包含一些使用多线程实现的高级示例。
高级打印机（日志记录器）
一个线程会打印接收到的所有内容，并且会依据终端宽度来修改输出。很有趣的是，一旦终端宽度出现变化，那些已经被打印出来的输出也会随之被修改。
import threading
import Queue
import time
import sys
import subprocess
from获取终端大小的 backports.shutil 相关操作import get_terminal_size
printq = Queue.Queue()
interrupt = False
lines = []
def main():
ptt 是一个线程，其目标是 printer 。# 打开打印机
ptt.daemon = True
ptt.start()
# 一些要打印的示例内容
for i in xrange(1,100):
printq.put(' '.join([str(x) for x in range(1,i)])) # 向打印机发送内容的实际方式
time.sleep(.5)
def split_line(line, cols):
if len(line) > cols:
new_line = ''
ww = line.split()
i = 0
while len(new_line) <= (cols - len(ww[i]) - 1):
new_line += ww[i] + ' '
i += 1
print len(new_line)
if new_line == '':
return (line, '')
return (new_line, ' '.join(ww[i:]))
else:
return (line, '')
def printer():
while True:
获取终端的尺寸，将其分别赋值给 cols 和 rows 。# 获取终端尺寸
msg = '#' + '-' * (cols - 2) + '#\n' # 创建
try:
使用 printq 的 get_nowait 方法获取到的内容转换为字符串后赋值给 new_line 。
if new_line != '!@#EXIT#@!': # 优雅关闭打印机线程的好方法
lines.append(new_line)
printq.task_done()
else:
printq.task_done()
sys.exit()
except Queue.Empty:
pass
# 构建要显示的新消息，并拆分过长的行
for line in lines:
res = line # 下面是为了拆分行，如果行长于 cols
# 拆分行
while len(res) !=0:
2. 分割后的结果赋值给 res 和另一个变量（这里未明确给出另一个变量的名称）。
msg += '\n' + toprint
# 清空终端并打印新输出
subprocess.check_call('clear') # 保持终端干净
sys.stdout.write(msg)
sys.stdout.flush()
time.sleep(.5)5: 使用 while 循环的可停止线程
import threading
import time
class StoppableThread(threading.Thread):
有一个带有 stop() 方法的线程类。这个线程自身需要时不时地去检查 stopped() 这个条件。
def __init__(self):
调用父类 StoppableThread 的构造函数，并将 self 作为参数传递进去。
self 拥有一个名为 _stop_event 的对象，这个对象是 threading 模块中的 Event 类的实例。
def stop(self):
self._stop_event.set()
def join(self, *args, **kwargs):
self.stop()
调用父类 StoppableThread 的 join 方法，并传入参数 *args 和 **kwargs 。
def run():
while not self._stop_event.is_set():
print("Still running!")
time.sleep(2)
print("stopped!")基于[这个问题](
phper