Queue 配合Thread使用

Queue 配合Thread使用

生产消费者模型

结果

线程【3】开始下载https://picsum.photos/200/300线程【0】开始下载https://picsum.photos/300/300线程【4】开始下载https://picsum.photos/400/300
线程【1】开始下载https://picsum.photos/500/300线程【2】开始下载https://picsum.photos/600/300
>线程【0】 >>状态码为200
线程【0】下载https://picsum.photos/300/300成功
>线程【3】 >>状态码为200
线程【3】下载https://picsum.photos/200/300成功
>线程【1】 >>状态码为200
线程【1】下载https://picsum.photos/500/300成功
>线程【2】 >>状态码为200
线程【2】下载https://picsum.photos/600/300成功
>线程【4】 >>状态码为200
线程【4】下载https://picsum.photos/400/300成功
线程【0】队列已空,没有任务了，退出
线程【3】队列已空,没有任务了，退出
线程【1】队列已空,没有任务了，退出
线程【2】队列已空,没有任务了，退出
线程【4】队列已空,没有任务

示例代码

# -*- coding: utf-8 -*-
# 生产者 producer / 消费者 worker 模型
import queue
import threading
import timeimport requests# producer 【生产者】生产出的数据
image_urls = ["https://picsum.photos/200/300","https://picsum.photos/300/300","https://picsum.photos/400/300","https://picsum.photos/500/300","https://picsum.photos/600/300",
]q = queue.Queue()  # 创建一个队列def worker(thread_id):while True:try:url = q.get(timeout=3)  # 从队列中取出一个url,3秒没有取到就会抛出异常except queue.Empty:print(f"线程【{thread_id}】队列已空,没有任务了，退出")breakprint(f"线程【{thread_id}】开始下载{url}")try:r = requests.get(url, timeout=3)r.raise_for_status()print(f"线程【{thread_id}】 >>状态码为{r.status_code}")filename = f"images/img_{thread_id}_{int(time.time() * 1000)}.jpg"with open(filename, "wb") as f:f.write(r.content)print(f"线程【{thread_id}】下载{url}成功")except  Exception as e:print(f"线程【{thread_id}】下载{url}失败：{e}")q.task_done()  # 告诉队列，这个任务已经完成def run():threads = []for i in range(5):t = threading.Thread(target=worker, args=(i,))t.start()threads.append(t)# 主线程作为生产者，把任务塞进度队列中for url in image_urls:q.put(url)# 等待所有任务处理完成q.join()# 等待线程退出for tt in threads:tt.join()print("下载图片任务完成！")if __name__ == '__main__':run()

知识点

queue.Queue(maxsize=0) 是线程安全的队列

• 多线程操作同一个队列，不会崩溃，底层用锁保护了
• put() 是阻塞，队列满了会等
• get() 是阻塞，队列空了会等
• task_done() 和join() 搭配使用，表示任务已经处理完了。

timeout =3 ？

• 防止线程卡在 队列的get() 处。 任务都处理完了还在等
• 增加超时限制，让线程自动退出

多线程+Queue的场景

• 图片/视频/文件批量下载
• 爬虫调度URL抓去
• 异步日志写入工具
• 后台任务（上传，转码，消息消费）
• 实时采集数据和处理

queue.Queue 是多线程世界中的任务通道

参考

https://zhuanlan.zhihu.com/p/1923309703196288836