#Python asynchronous crawler tutorial (2024 updated version)

Is it time for the weekly crawler optimization session again? Obviously, crawling 100 test pages with a 1-second delay requires one and a half minutes to synchronize. Multi-threading is afraid of conflicts, heaping of resources, and inter-process communication is troublesome - yes, Python coroutine asynchronous crawler is the golden solution for IO-intensive tasks in 2024!

#1. Overview of asynchronous crawlers

#1.1 Why must it be asynchronous?

The core bottleneck of the crawler is never CPU operation, but network/disk IO waiting - for example, if you type arequests.get(), for the next tens, hundreds of milliseconds or even seconds, the program will just "wait" for the server to respond, unable to do anything.

An asynchronous crawler will immediately switch to executing other runnable tasks when a certain task is waiting for IO, using up all the waiting time!

#1.2 Three core advantages

#2. Core minimalist introduction (no official version)

There is no need to engage in complex underlying scheduling algorithms, just remember these three keywords:

#2.1 Coroutine

It can be understood as a task unit that “can be paused, resumed, and actively controlled by the programmer”—analogy to watching a movie:

• Synchronous thread: watch a movie to the end and don’t answer the phone in the middle
• Coroutine: Pause watching the movie → Answer the emergency call → Hang up after processing → Return to the place where the movie was paused and continue watching the movie

#2.2 Event Loop

This is the "general dispatcher" of the coroutine, which does three things in a loop in the background:

1. Check all coroutines: Which ones have been suspended but IO is completed (can be resumed)? Which ones can be run just after startup?
2. Select a task to execute according to the rules
3. The task is executed to the pause point (await), and then return to the loop

Python 3.7+ provides a super convenient entryasyncio.run(), no need to manually create/close event loop!

#2.3 async/await syntactic sugar

is the magic that makes coroutine code look like synchronous code:

• async def: Tell Python "This is not an ordinary function, it is a coroutine function. After calling, it will return the coroutine object and will not be executed immediately."
• await: Can only be used inasync def, which means "Wait until this asynchronous operation is completed before proceeding, and during this period you can do other tasks"

#3. Mainstream asynchronous HTTP tool: aiohttp

The most used one in the Python asynchronous ecosystem is aiohttp. The latest 3.9+ version in 2024 will have a smoother experience!

#3.1 The most basic single page crawling

import aiohttp
import asyncio

async def fetch(url):
    # 异步上下文管理器管理Session（类比requests.Session）
    # 会自动复用连接池、自动清理资源
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            # 等待响应文本读取完成
            return await response.text()

async def main():
    html = await fetch('https://example.com')
    print(html[:200])  # 只打印前200字，避免刷屏

# 3.7+ 标准入口
asyncio.run(main())

#3.2 2024 version 3.9+ small updates worth using

1. HTTP/2 support is enabled by default (requires manual confirmation of dependencies)h2installed)
2. Optimized the concurrent cache of DNS resolution
3. More fine-grained connection reuse and timeout control
4. Built-in pairhttpx.ResponseFormat compatibility (facilitates migration of old code)

#4. Practical combat: high-performance URL batch crawling

Directly upload a complete script framework with exception-handling, concurrency control, and data parsing!

#4.1 Complete code

import aiohttp
import asyncio
from bs4 import BeautifulSoup
from typing import List, Optional, Dict

# 限制并发数，避免被目标网站封IP
CONCURRENT_LIMIT = 20  # 单总连接数
HOST_LIMIT = 5          # 单个域名的最大连接数（非常重要！防封）
TIMEOUT = aiohttp.ClientTimeout(total=30, connect=10)

async def fetch_single(
    session: aiohttp.ClientSession,
    url: str
) -> Optional[str]:
    """爬取单个URL并处理异常"""
    try:
        async with session.get(url, timeout=TIMEOUT) as resp:
            if resp.status == 200:
                # 可以根据需要改成await resp.json()/await resp.read()
                return await resp.text()
            # 记录非200状态码
            print(f"⚠️  {url} 返回状态码 {resp.status}")
            return None
    except asyncio.TimeoutError:
        print(f"⏱️  {url} 超时")
        return None
    except Exception as e:
        print(f"❌  {url} 未知错误: {str(e)[:100]}")
        return None

async def parse_single(html: str) -> Optional[Dict]:
    """异步解析单个页面（这里用bs4是同步，但小数据量不影响）"""
    if not html:
        return None
    soup = BeautifulSoup(html, 'lxml')  # lxml比html.parser快很多
    # 👇 这里替换成你的解析逻辑，比如
    title = soup.title.string if soup.title else None
    return {"title": title}

async def batch_crawl(urls: List[str]) -> List[Dict]:
    """批量爬取+解析的主协程"""
    # 配置连接池
    connector = aiohttp.TCPConnector(
        limit=CONCURRENT_LIMIT,
        limit_per_host=HOST_LIMIT,
        force_close=False,  # 开启长连接复用
        enable_cleanup_closed=True
    )
    
    # 批量执行任务
    async with aiohttp.ClientSession(connector=connector) as session:
        # 生成所有爬取任务
        fetch_tasks = [fetch_single(session, url) for url in urls]
        # 等待所有爬取任务完成（gather会收集所有结果，即使有失败）
        raw_pages = await asyncio.gather(*fetch_tasks)
        
        # 生成所有解析任务（过滤掉None的页面）
        parse_tasks = [parse_single(page) for page in raw_pages if page]
        # 等待所有解析完成
        results = await asyncio.gather(*parse_tasks)
        # 最后过滤一下解析失败的None
        return [res for res in results if res]

if __name__ == "__main__":
    # 测试用的10个带1秒延迟的URL
    test_urls = [f"https://httpbin.org/delay/1?num={i}" for i in range(10)]
    # 爬取+计时
    import time
    start = time.time()
    final_data = asyncio.run(batch_crawl(test_urls))
    end = time.time()
    # 输出结果
    print(f"\n✅ 成功爬取+解析 {len(final_data)} 条数据")
    print(f"⏱️  总耗时: {end - start:.2f} 秒")

Tips:asyncio.gather()The results will be collected in the original order. Even if some tasks fail (throw an exception), the entire batch will not be interrupted unless you setreturn_exceptions=True。

#5. Best practice pitfall avoidance guide

#5.1 Preventing IP blocking is the first priority

In addition to the above codelimit_per_host, you can also add:

• Rate Limit: UseaiolimiterLibrary, limit requests per second

  from aiolimiter import AsyncLimiter
  limiter = AsyncLimiter(5, 1)  # 单个域名每秒5个请求
  
  async def limited_fetch(session, url):
      async with limiter:
          return await fetch_single(session, url)
• Random User-Agent: Usefake_useragent_asyncLibrary
• Random delay: inawait fetchAdd beforeawait asyncio.sleep(random.uniform(0.1, 0.5))

#5.2 Don’t mix synchronous blocking code

If called in a coroutinerequests.get()、time.sleep()、open()With this kind of synchronous blocking, the event loop will be completely stuck, and asynchronous use will be in vain!

• Change the corresponding asynchronous library:requests → aiohttp / httpx，time.sleep → asyncio.sleep，open → aiofiles

#5.3 Is asynchronous parsing also important?

If your data parsing is very complex (such as tens of thousands of words of long text, a large number of regular matches), you can use:

• asyncio.to_thread(): Throw synchronous parsing into the thread pool and run (built-in in Python 3.9+)
• Specialized asynchronous parsing library (such asselectolaxAlthough it is synchronized, it is morebs4More than 10 times faster, just use it directly for small/medium data amounts)

#6. Quick performance comparison (10 URLs with 1 second delay test)

Directly use the above actual script to simplify the synchronous/multi-threaded version test:

NOTE:HOST_LIMITAccording to the target website'srobots.txtOr the actual anti-climbing strategy adjustment, bigger is not better!

#7. Summary

In 2024, Python coroutine asynchronous crawlers are already the entry-level but high-efficiency choice - you don’t need to understand the complex underlying layers, just remember to "useasync defDefine tasks withawaitTo suspend and wait, useasyncio.run"Start", and then cooperateaiohttpWith the connection pool and exception-handling, you can write a crawler that is dozens of times faster than synchronization!

#Three core links for further study

1. aiohttp 官方中文文档（虽旧但核心内容没变）
2. Real Python 异步IO入门（英文但讲得超清楚）
3. httpx官方文档（aiohttp的替代，API和requests几乎一模一样，支持同步+异步）