在实际处理需求的过程中，博主比较偏爱使用多线程threading+队列queue的方式 去开发代码。

（本文注重的是搭建模板框架，仅供参考）

举例：

以豆瓣电影的排行为例，写个简单的demo。
豆瓣链接：https://movie.douban.com/top250?start=0&filter=

一共也就10页!

直接上代码：

importjsonimporttimeimporturllib3importthreadingfromqueueimportQueuefromloguruimportlogger urllib3.disable_warnings()classCrawlDoubanMovie(object):def__init__(self):self.max_times=5self.queue_req=Queue()# 创建塞入任务defcreate_tasks(self):foriinrange(1,11):self.queue_req.put({"page_num":i,"retry_times":0})logger.info(f"本次任务数量:{self.queue_req.qsize()}")# 请求解析操作defget_content_from_web(self,page):pass# 解析任务defparse_task(self,task_info):page_num=task_info["page_num"]retry_times=task_info.get("retry_times",0)logger.info(f"正在访问第【{page_num}】页！")ifretry_times>=self.max_times:logger.error(f"抓取失败，已达到最大重试次数！page_num:{page_num}")returnlogger.info(f"正在访问豆瓣电影第【{page_num}】页！(第{retry_times+1}次尝试)")try:self.get_content_from_web(page_num)exceptExceptionasexc:logger.warning(f"抓取失败:{exc}, 准备重试! page_num:{page_num}")time.sleep(1)# 加入延迟（防止疯狂重试）self.queue_req.put({"page_num":page_num,"retry_times":retry_times+1})else:logger.info(f"抓取成功，返回内容! page_num:{page_num}")defmain_run(self):whileTrue:task_info=self.queue_req.get()iftask_infoisNone:self.queue_req.task_done()breakelse:try:self.parse_task(task_info)finally:self.queue_req.task_done()# 放在 finally 里，防止异常导致 join 卡死if__name__=="__main__":threading_num=6crawl=CrawlDoubanMovie()crawl.create_tasks()threads=[]foriinrange(threading_num):t=threading.Thread(target=crawl.main_run)t.daemon=Truet.start()threads.append(t)crawl.queue_req.join()# 任务全部完成后，给每个线程发送退出信号 Nonefor_inrange(threading_num):crawl.queue_req.put(None)logger.info("所有任务执行完毕！")

模块和函数之间非常的清晰明了，

• create_tasks() 就负责初始化创建任务
• get_content_from_web（）负责请求和解析数据
• parse_task() 负责解析任务，这里面还包括了错误重试并重新放入队列中。
• main_run（）主函数 启动程序

搭配了多线程提升很多，也用到了队列。

运行效果图：

不仅使用方式简单实用，而且速度也是相当的快。

再重构一下：ThreadPoolExecutor

改用 Python 标准库的concurrent.futures.ThreadPoolExecutor

优势：

• 不用管 None 信号，不用管 task_done，不用管 join。代码少了三分之一。
• 天然支持获取返回结果。
• 遇到异常不会卡死。

importtimefromloguruimportloggerfromconcurrent.futuresimportThreadPoolExecutor,as_completedclassCrawlDoubanMovie(object):def__init__(self):self.max_times=5defget_content_from_web(self,page):# 模拟网络请求，假设 3 和 7 会失败ifpagein(3,7):raiseValueError("模拟网络超时")time.sleep(0.1)returnf"第{page}页内容"defcrawl_page_with_retry(self,page_num):"""带重试的单个任务处理函数"""forattemptinrange(1,self.max_times+1):logger.info(f"正在访问豆瓣电影第【{page_num}】页！(尝试{attempt}/{self.max_times})")try:content=self.get_content_from_web(page_num)logger.success(f"抓取成功! page_num:{page_num}")returncontentexceptExceptionasexc:logger.warning(f"抓取失败:{exc}, page_num:{page_num}")ifattempt==self.max_times:logger.error(f"page_num:{page_num}已达到最大重试次数，放弃！")returnNonetime.sleep(1)# 失败后等待 1 秒再试defmain_run(self,page_nums):# 设置最大并发数为 6withThreadPoolExecutor(max_workers=6)asexecutor:# 提交所有任务到线程池future_to_page={executor.submit(self.crawl_page_with_retry,page):pageforpageinpage_nums}# 等待所有任务结束并获取结果forfutureinas_completed(future_to_page):page=future_to_page[future]try:result=future.result()# 这里可以处理成功的结果，比如存入数据库ifresult:passexceptExceptionasexc:logger.error(f"任务{page}发生未知异常:{exc}")if__name__=="__main__":crawl=CrawlDoubanMovie()# 直接生成页码列表提交crawl.main_run(range(1,11))logger.info("全部任务结束")

ThreadPoolExecutor 写法，代码更清晰，维护成本更低，且不会遇到队列死锁的问题。