使用Embassy库编写异步爬虫

admin管理员组

文章数量:1033379

使用Embassy库编写异步爬虫

最近有个学员想用Embassy库写一个网络爬虫程序。首先，我需要确认Embassy是什么。Embassy是一个用于Python的异步HTTP客户端库，基于aiohttp和async/await语法，适合高性能的爬虫需求。

接下来，部分人可能对异步编程不太熟悉，所以我在写代码示例时需要简洁明了，同时包含必要的注释。可能的需求包括发送GET请求、处理响应、提取数据，以及错误处理。

以下是一个使用 Python 的 Embassy 库（基于 aiohttp 的异步 HTTP 客户端）编写的网络爬虫示例。Embassy 是一个高性能的异步 HTTP 客户端库，适合用于编写高效的网络爬虫。

先安装依赖

代码语言：javascript代码运行次数：0运行复制

pip install embassy aiohttp beautifulsoup4

示例代码：异步爬虫

代码语言：javascript代码运行次数：0运行复制

import asyncio
from typing import Optional
from urllib.parse import urlparse
​
from aiohttp import ClientSession, ClientTimeout
from bs4 import BeautifulSoup
from embassy import fetch, Request, Stats
​
​
class AsyncCrawler:
    def __init__(self, concurrency: int = 5, timeout: int = 10):
        self.concurrency = concurrency  # 并发数
        self.timeout = ClientTimeout(total=timeout)  # 超时设置
        self.visited_urls = set()       # 已访问的 URL
        self.queue = asyncio.Queue()    # 待爬取的 URL 队列
​
    async def crawl(self, url: str):
        """启动爬虫"""
        await self.queue.put(url)
        tasks = []
        # 创建并发任务
        async with ClientSession(timeout=self.timeout) as session:
            for _ in range(self.concurrency):
                task = asyncio.create_task(self.worker(session))
                tasks.append(task)
            # 等待队列清空
            await self.queue.join()
            # 取消所有 worker 任务
            for task in tasks:
                task.cancel()
​
    async def worker(self, session: ClientSession):
        """工作线程：处理队列中的 URL"""
        while True:
            url = await self.queue.get()
            try:
                # 发送 HTTP 请求
                response = await fetch(
                    session,
                    Request(
                        method="GET",
                        url=url,
                        headers={"User-Agent": "EmbassyCrawler/1.0"},
                    )
                )
                # 处理响应
                await self.process_response(response)
                # 提取新链接
                new_links = self.extract_links(response.text, url)
                for link in new_links:
                    if link not in self.visited_urls:
                        await self.queue.put(link)
                        self.visited_urls.add(link)
            except Exception as e:
                print(f"Error fetching {url}: {str(e)}")
            finally:
                self.queue.task_done()
​
    async def process_response(self, response: Stats):
        """处理响应数据"""
        print(f"Scraped: {response.url} [Status: {response.status}]")
        # 这里可以添加数据解析逻辑（例如保存到文件/数据库）
        # 使用 BeautifulSoup 解析 HTML
        soup = BeautifulSoup(response.text, 'html.parser')
        # 示例：提取页面标题
        title = soup.title.string if soup.title else "No Title"
        print(f"Title: {title}")
​
    def extract_links(self, html: str, base_url: str) -> list:
        """从 HTML 中提取所有链接"""
        soup = BeautifulSoup(html, 'html.parser')
        links = []
        for a_tag in soup.find_all('a', href=True):
            href = a_tag['href']
            # 处理相对路径
            full_url = self.normalize_url(href, base_url)
            if full_url:
                links.append(full_url)
        return links
​
    @staticmethod
    def normalize_url(url: str, base_url: str) -> Optional[str]:
        """标准化 URL"""
        parsed = urlparse(url)
        if parsed.scheme and parsedloc:
            return url  # 已经是绝对路径
        # 拼接基础 URL
        base_parsed = urlparse(base_url)
        return base_parsed.scheme + "://" + base_parsedloc + parsed.path
​
​
async def main():
    # 初始化爬虫（设置并发数为3，超时10秒）
    crawler = AsyncCrawler(concurrency=3, timeout=10)
    # 从种子 URL 开始爬取
    await crawler.crawl(";)
​
​
if __name__ == "__main__":
    asyncio.run(main())

代码说明

1. Embassy 集成：
   • 使用 embassy.fetch 发送异步 HTTP 请求。
   • Request 对象配置请求参数（方法、URL、请求头）。
2. 异步架构：
   • 通过 asyncio.Queue 管理待爬取 URL。
   • 使用 ClientSession 复用 TCP 连接提升性能。
   • 并发由 asyncio.create_task 和 Semaphore 控制。
3. 功能模块：
   • URL 标准化：处理相对路径和绝对路径。
   • 链接提取：使用 BeautifulSoup 解析 HTML。
   • 错误处理：捕获请求异常并记录。
4. 反爬策略：
   • 默认添加 User-Agent 请求头。
   • 建议根据目标网站调整请求频率（可添加 asyncio.sleep 延迟）。

运行注意事项

1. 遵守 robots.txt：确保目标网站允许爬取。
2. 调整并发数：根据目标网站负载能力设置 concurrency。
3. 存储数据：可在 process_response 中添加数据库或文件写入逻辑。
4. 代理支持：如需代理，可在 Request 中配置 proxy 参数。

如果需要扩展功能（如限速、自动重试、动态页面渲染），可以结合其他库（如 aiohttp-retry 或 pyppeteer）。

使用Embassy库编写异步爬虫

最近有个学员想用Embassy库写一个网络爬虫程序。首先，我需要确认Embassy是什么。Embassy是一个用于Python的异步HTTP客户端库，基于aiohttp和async/await语法，适合高性能的爬虫需求。

接下来，部分人可能对异步编程不太熟悉，所以我在写代码示例时需要简洁明了，同时包含必要的注释。可能的需求包括发送GET请求、处理响应、提取数据，以及错误处理。

以下是一个使用 Python 的 Embassy 库（基于 aiohttp 的异步 HTTP 客户端）编写的网络爬虫示例。Embassy 是一个高性能的异步 HTTP 客户端库，适合用于编写高效的网络爬虫。

先安装依赖

代码语言：javascript代码运行次数：0运行复制

pip install embassy aiohttp beautifulsoup4

示例代码：异步爬虫

代码语言：javascript代码运行次数：0运行复制

import asyncio
from typing import Optional
from urllib.parse import urlparse
​
from aiohttp import ClientSession, ClientTimeout
from bs4 import BeautifulSoup
from embassy import fetch, Request, Stats
​
​
class AsyncCrawler:
    def __init__(self, concurrency: int = 5, timeout: int = 10):
        self.concurrency = concurrency  # 并发数
        self.timeout = ClientTimeout(total=timeout)  # 超时设置
        self.visited_urls = set()       # 已访问的 URL
        self.queue = asyncio.Queue()    # 待爬取的 URL 队列
​
    async def crawl(self, url: str):
        """启动爬虫"""
        await self.queue.put(url)
        tasks = []
        # 创建并发任务
        async with ClientSession(timeout=self.timeout) as session:
            for _ in range(self.concurrency):
                task = asyncio.create_task(self.worker(session))
                tasks.append(task)
            # 等待队列清空
            await self.queue.join()
            # 取消所有 worker 任务
            for task in tasks:
                task.cancel()
​
    async def worker(self, session: ClientSession):
        """工作线程：处理队列中的 URL"""
        while True:
            url = await self.queue.get()
            try:
                # 发送 HTTP 请求
                response = await fetch(
                    session,
                    Request(
                        method="GET",
                        url=url,
                        headers={"User-Agent": "EmbassyCrawler/1.0"},
                    )
                )
                # 处理响应
                await self.process_response(response)
                # 提取新链接
                new_links = self.extract_links(response.text, url)
                for link in new_links:
                    if link not in self.visited_urls:
                        await self.queue.put(link)
                        self.visited_urls.add(link)
            except Exception as e:
                print(f"Error fetching {url}: {str(e)}")
            finally:
                self.queue.task_done()
​
    async def process_response(self, response: Stats):
        """处理响应数据"""
        print(f"Scraped: {response.url} [Status: {response.status}]")
        # 这里可以添加数据解析逻辑（例如保存到文件/数据库）
        # 使用 BeautifulSoup 解析 HTML
        soup = BeautifulSoup(response.text, 'html.parser')
        # 示例：提取页面标题
        title = soup.title.string if soup.title else "No Title"
        print(f"Title: {title}")
​
    def extract_links(self, html: str, base_url: str) -> list:
        """从 HTML 中提取所有链接"""
        soup = BeautifulSoup(html, 'html.parser')
        links = []
        for a_tag in soup.find_all('a', href=True):
            href = a_tag['href']
            # 处理相对路径
            full_url = self.normalize_url(href, base_url)
            if full_url:
                links.append(full_url)
        return links
​
    @staticmethod
    def normalize_url(url: str, base_url: str) -> Optional[str]:
        """标准化 URL"""
        parsed = urlparse(url)
        if parsed.scheme and parsedloc:
            return url  # 已经是绝对路径
        # 拼接基础 URL
        base_parsed = urlparse(base_url)
        return base_parsed.scheme + "://" + base_parsedloc + parsed.path
​
​
async def main():
    # 初始化爬虫（设置并发数为3，超时10秒）
    crawler = AsyncCrawler(concurrency=3, timeout=10)
    # 从种子 URL 开始爬取
    await crawler.crawl(";)
​
​
if __name__ == "__main__":
    asyncio.run(main())

代码说明

1. Embassy 集成：
   • 使用 embassy.fetch 发送异步 HTTP 请求。
   • Request 对象配置请求参数（方法、URL、请求头）。
2. 异步架构：
   • 通过 asyncio.Queue 管理待爬取 URL。
   • 使用 ClientSession 复用 TCP 连接提升性能。
   • 并发由 asyncio.create_task 和 Semaphore 控制。
3. 功能模块：
   • URL 标准化：处理相对路径和绝对路径。
   • 链接提取：使用 BeautifulSoup 解析 HTML。
   • 错误处理：捕获请求异常并记录。
4. 反爬策略：
   • 默认添加 User-Agent 请求头。
   • 建议根据目标网站调整请求频率（可添加 asyncio.sleep 延迟）。

运行注意事项

1. 遵守 robots.txt：确保目标网站允许爬取。
2. 调整并发数：根据目标网站负载能力设置 concurrency。
3. 存储数据：可在 process_response 中添加数据库或文件写入逻辑。
4. 代理支持：如需代理，可在 Request 中配置 proxy 参数。

如果需要扩展功能（如限速、自动重试、动态页面渲染），可以结合其他库（如 aiohttp-retry 或 pyppeteer）。

本文标签： 使用Embassy库编写异步爬虫