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

异步爬虫aiohttp使用

news 2026/7/29 12:26:52

一、背景

二、应用场景

三、备注

一、背景

最近有个爬虫需求，页面是一个下拉框一个查询按钮，但是下拉框里有大概七八百多个选项，需求是要尽可能快的爬下来所有选项的数据。问了下AI，给出了aiohttp来做异步并发请求，之前也是没有用过，所以本文记录一下aiohttp使用方法。

二、应用场景

因为我们这个属于一个爬虫项目，框架任务的调度使用的aps管理，任务通过数据库查询得到，之前都是同步任务运行，但这个aiohttp是异步的函数功能，所以是要在任务里额外创建一个异步时间循环来运行异步任务。

def task(task_id=None, station_id=None, collect_date=None, province_code=None): """ 被aps定时任务查询，根据映射执行的具体任务入口 """ # 创建独立的事件循环 loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) try: results = loop.run_until_complete(async_crawl_flow()) return results finally: loop.close() class Crawler: ...... _sem = asyncio.Semaphore(10) # 控制并发数 async def async_crawl_flow(self): """ 异步主流程： 1. 获取下拉列表（拿到所有节点ID） 2. 并发查询每个节点详情 """ timeout = ClientTimeout(total=30) # aitohttp接受的proxy跟request有区别，是一个字符串 proxy_str = self.proxy.get('http') or self.proxy.get('https') async with aiohttp.ClientSession( timeout=timeout, headers=self.headers, cookies=self.cookie, proxy=proxy_str ) as session: print("开始获取下拉列表...") node_list = await self._fetch_node_list(session) if not node_list: print("未获取到节点列表，流程终止") return [] print(f"获取到 {len(node_list)} 个节点ID") print("开始并发查询节点详情...") results = await self._fetch_all_details(session, node_list) print(f"查询完成，成功 {len(results)} 条") return results # ============================================ # 第一步：获取下拉列表 # ============================================ async def _fetch_node_list(self, session: aiohttp.ClientSession) -> list[dict]: """ 调用下拉列表接口，提取所有节点ID """ try: async with session.post( 'https://xxx.com', json={} ) as resp: data = await resp.json() node_list = data.get("data", []) return node_list except Exception as e: print(f"获取下拉列表失败: {e}") return [] # ============================================ # 第二步：并发查询详情 # ============================================ async def _fetch_all_details(self, session: aiohttp.ClientSession, node_list: list[dict]) -> dict: """ 查询所有节点的详情 """ results = [] async def fetch_one_detail(node_info: dict): """查询单个节点详情""" id= node_info.get('id') name= node_info.get('name') async with self._sem: try: async with session.post( 'https://xxx.com', json={ 'phyunitId': id, 'dataTime': self.collect_date, }, timeout=ClientTimeout(total=10), ) as resp: if resp.status == 200: detail_data = await resp.json() results[id] = detail_data print(f"节点 {name} 查询成功") else: print(f"节点 {name} 返回状态码: {resp.status}") except asyncio.TimeoutError: print(f"节点 {name} 超时") # 创建所有任务并发执行 tasks = [fetch_one_detail(nl) for nl in node_list] await asyncio.gather(*tasks, return_exceptions=True) return results