影刀RPA 多线程基础：并发采集入门

影刀RPA 多线程基础并发采集入门署名林焱什么情况用什么RPA采集100个网页单线程一个个打开要30分钟。如果同时开5个线程6分钟就跑完了。多线程能显著提升IO密集型任务效率。场景推荐方式特点网页批量采集threading 队列简单易用CPU密集型计算multiprocessing绕过GIL限制大量并发请求ThreadPoolExecutor线程池管理怎么做一基础多线程importthreadingimporttime# 单线程对比deffetch_page(url):模拟采集一个网页print(f开始采集:{url})time.sleep(2)# 模拟网络请求耗时print(f完成采集:{url})returnf{url}的内容urls[https://example.com/page1,https://example.com/page2,https://example.com/page3,https://example.com/page4,https://example.com/page5,]# 单线程约10秒 print( 单线程 )starttime.time()results[]forurlinurls:results.append(fetch_page(url))print(f耗时:{time.time()-start:.1f}秒)# 多线程约2秒 print(\n 多线程 )starttime.time()threads[]results[None]*len(urls)# 预分配结果列表defworker(index,url):results[index]fetch_page(url)fori,urlinenumerate(urls):tthreading.Thread(targetworker,args(i,url))threads.append(t)t.start()# 等待所有线程完成fortinthreads:t.join()print(f耗时:{time.time()-start:.1f}秒)二线程池推荐拼多多店群自动化报活动上架fromconcurrent.futuresimportThreadPoolExecutor,as_completedimporttimedeffetch_page(url):采集网页time.sleep(1)# 模拟请求return{url:url,status:200,data:f{url}的内容}urls[fhttps://example.com/page{i}foriinrange(1,21)]# 使用线程池max_workers5# 最多5个线程同时运行results[]withThreadPoolExecutor(max_workersmax_workers)asexecutor:# 提交所有任务future_to_url{executor.submit(fetch_page,url):urlforurlinurls}# 按完成顺序获取结果forfutureinas_completed(future_to_url):urlfuture_to_url[future]try:resultfuture.result()results.append(result)print(f完成:{url})exceptExceptionase:print(f失败:{url}, 错误:{e})print(f\n共完成{len(results)}个任务)三线程安全的队列importthreadingimportqueueimporttime# 任务队列task_queuequeue.Queue()result_queuequeue.Queue()# 填充任务urls[fhttps://example.com/page{i}foriinrange(1,21)]forurlinurls:task_queue.put(url)defworker(worker_id):工作线程whileTrue:try:# 非阻塞获取任务超时退出urltask_queue.get(timeout1)exceptqueue.Empty:break# 执行任务try:time.sleep(0.5)# 模拟采集resultf[线程{worker_id}] 采集完成:{url}result_queue.put({url:url,status:success,data:result})exceptExceptionase:result_queue.put({url:url,status:error,error:str(e)})finally:task_queue.task_done()# 启动多个工作线程num_workers5threads[]foriinrange(num_workers):tthreading.Thread(targetworker,args(i,))t.start()threads.append(t)# 等待所有任务完成task_queue.join()# 收集结果results[]whilenotresult_queue.empty():results.append(result_queue.get())print(f完成{len(results)}个任务)forrinresults[:5]:print(r[data])四线程安全的数据操作importthreading# 线程安全的计数器classSafeCounter:def__init__(self):self._lockthreading.Lock()self._count0defincrement(self):withself._lock:self._count1returnself._countpropertydefvalue(self):withself._lock:returnself._count# 线程安全的列表classSafeList:def__init__(self):self._lockthreading.Lock()self._data[]defappend(self,item):withself._lock:self._data.append(item)defget_all(self):withself._lock:returnlist(self._data)# 使用counterSafeCounter()resultsSafeList()defworker(url):numcounter.increment()results.append(f第{num}个任务:{url})threads[threading.Thread(targetworker,args(furl{i},))foriinrange(10)]fortinthreads:t.start()fortinthreads:t.join()print(f计数:{counter.value})foriteminresults.get_all():print(item)完整流程并发采集管理器fromconcurrent.futuresimportThreadPoolExecutor,as_completedimportthreadingimporttimefromdatetimeimportdatetime# yd_input: urls, max_workers, delayurlsyd_input.get(urls,[])max_workersyd_input.get(max_workers,5)delayyd_input.get(delay,0.5)# 统计stats{total:len(urls),success:0,failed:0,start_time:datetime.now().strftime(%Y-%m-%d %H:%M:%S)}stats_lockthreading.Lock()results[]results_lockthreading.Lock()deffetch_url(url):采集单个URLtry:time.sleep(delay)# 控制请求频率# 这里放实际的采集逻辑# 例如用requests获取网页、解析数据等importrequests resprequests.get(url,timeout15,headers{User-Agent:Mozilla/5.0 (Windows NT 10.0; Win64; x64)})ifresp.status_code200:result{url:url,status:success,code:resp.status_code,length:len(resp.text),timestamp:datetime.now().strftime(%H:%M:%S)}withstats_lock:stats[success]1withresults_lock:results.append(result)returnresultelse:raiseException(fHTTP{resp.status_code})exceptExceptionase:withstats_lock:stats[failed]1result{url:url,status:error,error:str(e),timestamp:datetime.now().strftime(%H:%M:%S)}withresults_lock:results.append(result)returnresult# 执行并发采集start_timetime.time()withThreadPoolExecutor(max_workersmax_workers)asexecutor:futures{executor.submit(fetch_url,url):urlforurlinurls}completed0forfutureinas_completed(futures):completed1urlfutures[future]try:resultfuture.result()status_icon✓ifresult[status]successelse✗print(f{status_icon}[{completed}/{len(urls)}]{url})exceptExceptionase:print(f✗ [{completed}/{len(urls)}]{url}:{e})elapsedtime.time()-start_time stats[end_time]datetime.now().strftime(%Y-%m-%d %H:%M:%S)stats[elapsed]f{elapsed:.1f}秒print(f\n 采集完成 )print(f总计:{stats[total]}, 成功:{stats[success]}, 失败:{stats[failed]})print(f耗时:{elapsed:.1f}秒, 平均:{elapsed/len(urls):.2f}秒/个)yd_output{status:ok,stats:stats,results:results}有什么坑坑一多线程共享变量导致数据错乱现象多个线程同时写同一个列表结果列表长度不对或某些数据丢失。原因Python列表的append操作不是原子操作多线程同时写会导致数据覆盖。解决用锁保护共享数据importthreading# 错误不加锁# results []# def worker(url):# results.append(url) ❌ 多线程同时append可能丢数据# 正确用锁results[]lockthreading.Lock()defworker(url):withlock:results.append(url)# ✓ 同一时刻只有一个线程能写# 更好的方案用Queue线程安全importqueue result_queuequeue.Queue()defworker(url):result_queue.put(url)# ✓ Queue自带线程安全坑二线程数过多反而变慢现象把max_workers设为100结果比5个线程还慢。原因线程太多导致频繁的上下文切换开销大于并行收益。另外大量并发请求可能触发网站反爬导致请求被拒绝后重试。解决合理设置线程数TEMU店群矩阵自动化运营核价报活动# IO密集型任务网络请求线程数 CPU核数 × 2~5importos cpu_countos.cpu_count()# CPU核数optimal_workersmin(cpu_count*3,20)# 推荐3-20个线程# 不要无脑设大# max_workers 100 ❌ 太多反而慢# 根据目标网站承受能力调整# 小网站3-5个线程# 大网站10-20个线程# API接口看API的rate limit要求坑三主线程退出导致子线程中断现象主线程跑完了子线程还没完成就被强制退出结果不完整。原因默认子线程是守护线程daemonFalse主线程会等子线程完成。但如果设了daemonTrue主线程退出时子线程会被杀掉。解决确保正确等待子线程完成importthreading# 方案1用join等待推荐threads[]foriinrange(5):tthreading.Thread(targetworker,args(i,))t.start()threads.append(t)fortinthreads:t.join()# 等待每个线程完成print(所有线程完成)# 方案2用ThreadPoolExecutor自动等待fromconcurrent.futuresimportThreadPoolExecutorwithThreadPoolExecutor(max_workers5)asexecutor:futures[executor.submit(worker,i)foriinrange(5)]# with块结束时自动等待所有任务完成# 错误设为daemonTrue后不等待# t threading.Thread(targetworker, daemonTrue)# t.start()# # 不join主线程退出时子线程被杀 ❌坑四线程中的异常不会自动传播到主线程现象子线程报错了但主线程不知道以为全部成功。原因子线程的异常不会自动抛到主线程需要主动检查。解决用Future.result()获取异常或用try-except记录fromconcurrent.futuresimportThreadPoolExecutor,as_completeddefworker(url):iferrorinurl:raiseValueError(模拟错误)returnfsuccess:{url}urls[url1,error_url,url3]withThreadPoolExecutor(max_workers3)asexecutor:futures{executor.submit(worker,url):urlforurlinurls}forfutureinas_completed(futures):urlfutures[future]try:# future.result()会重新抛出子线程中的异常resultfuture.result()print(f✓{url}:{result})exceptExceptionase:print(f✗{url}:{e})# 捕获子线程异常# 如果不用Future需要在worker函数内部catch异常defsafe_worker(url):try:# 实际逻辑returndo_work(url)exceptExceptionase:return{url:url,error:str(e)}# 返回错误信息而不是抛出