Python异步爬虫：aiohttp实现百万级数据采集实战指南

import aiohttp
import asyncio
from collections import deque

class AsyncCrawler:
    def __init__(self, max_concurrency=500):
        self.semaphore = asyncio.Semaphore(max_concurrency)
        self.session = aiohttp.ClientSession()
        self.task_queue = deque()

async def fetch_url(self, url, retry_times=3):
    async with self.semaphore:
        for _ in range(retry_times):
            try:
                async with self.session.get(url, timeout=30) as resp:
                    if resp.status == 200:
                        return await resp.text()
                    await asyncio.sleep(1)  # 状态码非200时短暂等待
            except (aiohttp.ClientError, asyncio.TimeoutError):
                continue
    return None

async def worker(self, task_queue):
    while True:
        url = task_queue.popleft()
        html = await self.fetch_url(url)
        if html:
            # 处理数据并存储
            pass
        # 动态调整并发数
        if len(task_queue) < 1000 and self.semaphore._value < self.max_concurrency//2:
            self.semaphore._value += 10

connector = aiohttp.TCPConnector(
    limit_per_host=100,  # 单域名最大连接数
    ttl_dns_cache=300,   # DNS缓存时间
    force_close=False     # 保持长连接
)
session = aiohttp.ClientSession(connector=connector)

async with session.get(
    url,
    headers={'Accept-Encoding': 'gzip, deflate'},
    compress='gzip'
) as resp:
    # 自动解压处理
    data = await resp.read()

from parsel import Selector

def parse_content(html):
    try:
        sel = Selector(text=html)
        # 优先使用CSS选择器
        title = sel.css('h1::text').get()
        # 备用XPath方案
        if not title:
            title = sel.xpath('//h1/text()').get()
        return {'title': title}
    except Exception:
        return None

async def batch_insert(self, data_list):
    if not data_list:
        return
    
    # MongoDB批量插入示例
    await self.collection.insert_many(data_list, ordered=False)
    
    # ClickHouse批量插入示例
    query = "INSERT INTO products FORMAT JSONEachRow"
    async with self.pool.acquire() as conn:
        async with conn.cursor() as cursor:
            await cursor.execute(query, data_list)

import aiohttp
import asyncio
from parsel import Selector
import pymongo
from motor.motor_asyncio import AsyncIOMotorClient

class MillionScaleCrawler:
    def __init__(self):
        self.semaphore = asyncio.Semaphore(500)
        self.client = AsyncIOMotorClient('mongodb://localhost:27017')
        self.db = self.client['crawler_db']
        self.collection = self.db['products']
        
        connector = aiohttp.TCPConnector(limit_per_host=100)
        self.session = aiohttp.ClientSession(connector=connector)

    async def fetch_url(self, url):
        async with self.semaphore:
            try:
                async with self.session.get(
                    url,
                    headers={'User-Agent': 'Mozilla/5.0'},
                    timeout=30,
                    compress='gzip'
                ) as resp:
                    if resp.status == 200:
                        return await resp.text()
            except Exception:
                return None

    async def parse_product(self, html, url):
        try:
            sel = Selector(text=html)
            return {
                'url': url,
                'title': sel.css('h1::text').get().strip(),
                'price': sel.css('.price::text').get(),
                'timestamp': int(time.time())
            }
        except:
            return None

    async def process_url(self, url):
        html = await self.fetch_url(url)
        if html:
            product = await self.parse_product(html, url)
            if product:
                await self.collection.insert_one(product)

    async def run(self, url_list):
        tasks = [self.process_url(url) for url in url_list]
        await asyncio.gather(*tasks)

    async def close(self):
        await self.session.close()
        self.client.close()

# 使用示例
async def main():
    crawler = MillionScaleCrawler()
    urls = ['https://example.com/product/{}'.format(i) for i in range(1000000)]
    await crawler.run(urls[:1000])  # 测试时先处理1000条
    await crawler.close()

if __name__ == '__main__':
    asyncio.run(main())