在实际处理需求的过程中博主比较偏爱使用多线程threading队列queue的方式 去开发代码。本文注重的是搭建模板框架仅供参考举例以豆瓣电影的排行为例写个简单的demo。豆瓣链接https://movie.douban.com/top250?start0filter一共也就10页!直接上代码importjsonimporttimeimporturllib3importthreadingfromqueueimportQueuefromloguruimportlogger urllib3.disable_warnings()classCrawlDoubanMovie(object):def__init__(self):self.max_times5self.queue_reqQueue()# 创建塞入任务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_numtask_info[page_num]retry_timestask_info.get(retry_times,0)logger.info(f正在访问第【{page_num}】页)ifretry_timesself.max_times:logger.error(f抓取失败已达到最大重试次数page_num:{page_num})returnlogger.info(f正在访问豆瓣电影第【{page_num}】页(第{retry_times1}次尝试))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_times1})else:logger.info(f抓取成功返回内容! page_num:{page_num})defmain_run(self):whileTrue:task_infoself.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_num6crawlCrawlDoubanMovie()crawl.create_tasks()threads[]foriinrange(threading_num):tthreading.Thread(targetcrawl.main_run)t.daemonTruet.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_times5defget_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_times1):logger.info(f正在访问豆瓣电影第【{page_num}】页(尝试{attempt}/{self.max_times}))try:contentself.get_content_from_web(page_num)logger.success(f抓取成功! page_num:{page_num})returncontentexceptExceptionasexc:logger.warning(f抓取失败:{exc}, page_num:{page_num})ifattemptself.max_times:logger.error(fpage_num:{page_num}已达到最大重试次数放弃)returnNonetime.sleep(1)# 失败后等待 1 秒再试defmain_run(self,page_nums):# 设置最大并发数为 6withThreadPoolExecutor(max_workers6)asexecutor:# 提交所有任务到线程池future_to_page{executor.submit(self.crawl_page_with_retry,page):pageforpageinpage_nums}# 等待所有任务结束并获取结果forfutureinas_completed(future_to_page):pagefuture_to_page[future]try:resultfuture.result()# 这里可以处理成功的结果比如存入数据库ifresult:passexceptExceptionasexc:logger.error(f任务{page}发生未知异常:{exc})if__name____main__:crawlCrawlDoubanMovie()# 直接生成页码列表提交crawl.main_run(range(1,11))logger.info(全部任务结束)ThreadPoolExecutor 写法代码更清晰维护成本更低且不会遇到队列死锁的问题。