当前位置：首页 > news >正文

python并发爬虫

news 来源：原创 2025/7/21 6:08:06

爬虫多线程方法生成

from threading import Threaddef func(name):for i in range(100):print(f"{name}完成了{i}项任务")if __name__ == '__main__':t1 = Thread(target=func, args=('老杨',))t2 = Thread(target=func, args=('老李',))t3 = Thread(target=func, args=('老孙',))t1.start()t2.start()t3.start()t1.join()t2.join()t3.join()print("主线程结束")

爬虫多线程类生成

from threading import Thread
from time import sleepclass MyThread(Thread):def __init__(self, name):super(MyThread, self).__init__()  # 继承MyThread的父类self.name = namedef run(self):for i in range(100):print(f"{self.name}完成了{i}项工作")sleep(0.5)if __name__ == '__main__':t1 = MyThread('老杨')t2 = MyThread('老孙')t3 = MyThread('老李')t1.start()t2.start()t3.start()t1.join()t2.join()t3.join()print("主线程结束")

线程池

from concurrent.futures import ThreadPoolExecutordef func(name):for i in range(10):print(name, i)if __name__ == '__main__':with ThreadPoolExecutor(10) as t:  # 作用是创建10个线程for i in range(100):t.submit(func, f"周杰伦{i}")

这段代码使用了 ThreadPoolExecutor 来创建一个线程池，允许并行执行多个任务。具体来说：

创建线程池：ThreadPoolExecutor(10) 创建一个能够同时管理 10 个线程的线程池。
提交任务：t.submit(func, f"周杰伦{i}") 在循环中提交了 100 个任务（i 从 0 到 99）。每个任务调用 func 函数，并传入一个字符串参数，格式为 "周杰伦{i}"（例如 "周杰伦0", "周杰伦1", ..., "周杰伦99"）。
并发执行：ThreadPoolExecutor 将会在可用的 10 个线程中并发执行这些任务。然而，由于总共有 100 个任务，线程池会轮流使用线程，确保每次都只有 10 个任务在运行。

但是这种会发生资源的争端，后续可以使用生产者消费者的模式，来确保资源不会被重复。

如果想要拿到返回值怎么弄

线程池返回值1

import time
from concurrent.futures import ThreadPoolExecutordef func(name, t):time.sleep(t)return namedef fn(res):print(res.result())if __name__ == '__main__':with ThreadPoolExecutor(10) as t:t.submit(func, '周结论', 3).add_done_callback(fn)t.submit(func, '周一', 2).add_done_callback(fn)t.submit(func, '周二', 1).add_done_callback(fn)

在这段代码中，t.submit(func, '周二', 1) 的结果是一个 Future 对象，它表示异步执行 func 函数的计算结果。add_done_callback(fn) 方法用于注册一个回调函数 fn，这个回调函数会在 Future 对象完成时被调用。

当 fn 被调用时，它会接收一个参数，该参数是已完成的 Future 对象。这个对象包含了 func 执行的结果、异常信息等。

因此，add_done_callback(fn) 会向 fn 传入这个 Future 对象作为参数。你可以在 fn 函数内通过这个对象访问你需要的信息，比如：

如果 func 执行成功，可以通过 future.result() 获取结果。
如果 func 执行失败，可以通过 future.exception() 获取抛出的异常。

在这个业务逻辑当中，add_done_callback返回会立即执行，返回call_back执行的顺序是不确定的，返回的顺序是不确定的。

线程池返回值2

import time
from concurrent.futures import ThreadPoolExecutordef func(name, t):time.sleep(t)print(f"我是", name)return nameif __name__ == '__main__':with ThreadPoolExecutor(10) as t:result = t.map(func, ['周杰伦', '老李', '小王'], [2, 1, 3])for i in result:print(i)

map的返回值是生成器，返回的内容和任务分发的顺序是一致的

我们要学会看函数

这是map的函数，要学会自己看，fn就是要传入的函数，*iterables的意思是可迭代对象，所以列表是可迭代对象吧，元组也是可迭代对象，所以这里不只是传入列表。然后看Returns an iterator equivalent to map(fn ,iter)会返回一个可迭代的对象，所以要拿到返回值我们就可以使用for循环来拿取返回值。

线程池实战案例

网址为北京新发地菜市场:新发地-价格行情

import json
import requests
from concurrent.futures import ThreadPoolExecutor
import threadingheaders = {"Accept": "*/*","Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8,en-GB;q=0.7,en-US;q=0.6","Connection": "keep-alive","Content-Type": "application/x-www-form-urlencoded; charset=UTF-8","Origin": "http://www.xinfadi.com.cn","Referer": "http://www.xinfadi.com.cn/priceDetail.html","User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/134.0.0.0 Safari/537.36 Edg/134.0.0.0","X-Requested-With": "XMLHttpRequest"
}
url = "http://www.xinfadi.com.cn/getPriceData.html"# 线程锁用于文件写入
file_lock = threading.Lock()def get_data(current):data = {"limit": "20","current": "","pubDateStartTime": "2023/01/01","pubDateEndTime": "2023/12/31",  # 设置结束时间"prodPcatid": "","prodCatid": "","prodName": ""}data["current"] = str(current)err_number = 0while True:try:proxy = get_ip()#这里插入自己的代理ipresponse = requests.post(url, headers=headers, data=data, proxies=proxy)response.raise_for_status()  # 检查请求是否成功data_list = json.loads(response.text)['list']except requests.exceptions.RequestException as e:err_number += 1if err_number == 10:print(f"{current}请求次数超过10次")print(f'{current}号请求失败: {e}')returnexcept json.JSONDecodeError as e:err_number += 1if err_number == 10:print(f"{current}请求次数超过10次")print(f'{current}号JSON解析失败: {e}')returnexcept KeyError as e:err_number += 1if err_number == 10:print(f"{current}请求次数超过10次")print(f'{current}号数据格式错误: {e}')returncontinuefor item in data_list:prodName = item.get("prodName", "")highPrice = item.get("highPrice", "")lowPrice = item.get("lowPrice", "")avgPrice = item.get("avgPrice", "")# 使用线程锁确保文件写入安全with file_lock:with open('data.csv', mode='a', encoding='utf-8') as f:f.write(f'{current}, {prodName}, {lowPrice}, {avgPrice}, {highPrice}\n')print(f"{current}号的数据爬取完成")if __name__ == '__main__':# 初始化或清理 data.csv 文件with open('data.csv', mode='w', encoding='utf-8') as f:f.write("日期, 产品名称, 最低价, 平均价, 最高价\n")with ThreadPoolExecutor(max_workers=10) as t:  # 调整线程池大小for day in range(1, 40):t.submit(get_data, day)

讲解一下，这里的ThreadPoolExecutor(max_workers=10) as t:

下面有for循环，每次爬取的数据包不一样，所以在爬取数据包上不会造成数据冲突，但是在写入数据的时候，有可能会造成数据重复，所以这里采用了数据锁，在写文件的时候，保证每次只有一个线程对文件进行写操作。

多进程多线程

多进程和多线程差不多，读者可以去看我在python收录下的并发程序这篇文章