爬虫面试题

总结一下最近面试遇到的笔试题

1、解释Python中的init方法的作用。

在Python中，__init__方法是一种特殊的构造方法，主要用于在创建类的实例时初始化对象。至少接受至少一个参数：self,它是对当前实例的引用，可以通过添加其他参数传递初始值。

class Person:
    def __init__(self, name, age):
        self.name = name  # 初始化属性
        self.age = age

# 创建实例时，__init__自动执行
person = Person("Alice", 25)
print(person.name)  
print(person.age)  

输出：
Alice
25

需要注意的是，__init__不是必须的，可以省略，且__init__应该返回None,否则会引发TypeError。

2、Python中的*args和**args分别是什么？

*args：可变位置参数，接收任意数量的位置参数，并将其打包成一个元组（tuple)。

def sum_numbers(*args):
	total = 0
	for num in args:
		total +=  num
	return total

print(sum_numbers(1, 2, 3)) 

输出：
6

**kwargs：接收任意数量的关键字参数，并将其打包成一个字典(dict)。

def get_info(**kwargs):
	for key,  value in kwargs.items():
		print(f"{key}: {value}")
get_info(name = "Alice", age = 25, city = "New York") 

输出：
name: Alice
age: 25
city: New York

3、写出一个Python生成器的示例，并解释生成器的作用。

生成器是一种惰性计算的特殊迭代器，用于按需生成数据，而不是一次性计算并存储所有数据，适合处理大数据流，节省内存，提高性能，支持无限序列（如斐波那契数列）。

（1）通过yield关键字创建生成器

def count_up_to(max_num):
    num = 1
    while num <= max_num:
        yield num  # 每次调用 yield 返回一个值，并暂停执行
        num += 1

# 创建生成器对象
counter = count_up_to(3)

# 逐个获取值
print(next(counter))  
print(next(counter))  
print(next(counter))  
print(next(counter))  

输出：
1
2
3
StopIteration异常

（2）通过表达式创建生成器

squares = (x ** 2 for x in range(3)) # 生成器表达式（类似列表推导，但返回生成器）

# 逐个获取值
print(next(squares))
print(next(squares))
print(next(squares))
print(next(squares))

输出：
0
1
4
StopIteration异常

4、Python中如何实现多线程？请写一个简单的多线程示例。

在Python中通过threading模块实现多线程编程，允许程序同时执行多个任务（并发执行）。但由于GIL(全局解释器锁)的限制，多线程更适合I/O密集型任务（如网络请求、文件读写），而不是CPU密集型任务（计算密集型任务更适合用multiprocessing）。

import threading
import time

def task(name, delay):
    print(f"线程 {name} 启动")
    time.sleep(delay)  # 模拟耗时操作
    print(f"线程 {name} 完成")

# 创建两个线程
thread1 = threading.Thread(target=task, args=("A", 2))
thread2 = threading.Thread(target=task, args=("B", 1))

# 启动线程
thread1.start()
thread2.start()

# 等待线程结束
thread1.join()
thread2.join()

print("所有线程执行完毕")

输出（由于线程并发，顺序可能不同）：
线程 A 启动
线程 B 启动
线程 B 完成 # B先结束，因为delay=1
线程 A 完成
所有线程执行完毕

5、解释Python中的GIL（全局解释器锁）及其对多线程的影响。

GIL,即全局解释器锁，是一个用于同步Python字节码执行的机制，确保同一时刻只有一个线程执行Python字节码。这一设计初衷是为了简化Python的内存管理，防止多线程同时修改数据导致的竞态条件和数据不一致问题。
然而，GIL的存在也限制了Python程序的并发性能，特别是在CPU密集型任务中，由于GIL的存在，即使有多个CPU核心可用，也只有一个线程能够执行代码，这意味着多线程程序并不会真正并行执行，而是交替执行，从而显著降低了程序的执行效率。

import threading
import time

def cpu_bound_task():
    count = 0
    for _ in range(10000000):
        count += 1

# 单线程
start = time.time()
cpu_bound_task()
cpu_bound_task()
print(f"单线程耗时: {time.time() - start:.2f}s")

# 多线程（由于 GIL，不会更快）
start = time.time()
t1 = threading.Thread(target=cpu_bound_task)
t2 = threading.Thread(target=cpu_bound_task)
t1.start()
t2.start()
t1.join()
t2.join()
print(f"多线程耗时: {time.time() - start:.2f}s")

输出：
单线程耗时: 0.48s
多线程耗时: 0.51s # 多线程反而更慢（线程切换开销）

6、如何避免爬虫被网站封禁？请列举至少3种策略。

（1）设置合理的请求频率，使用time.sleep()随机延迟

import requests
import time
import random

for page in range(1, 10):
    url = f"https://example.com/page/{page}"
    response = requests.get(url)
    print(f"爬取页面 {page}")
    time.sleep(random.uniform(1, 3))  # 随机延迟1~3秒

（2）使用随机User-Agent和IP,伪装不同用户

import requests
from fake_useragent import UserAgent

ua = UserAgent()
headers = {"User-Agent": ua.random}  # 随机User-Agent
proxies = {"http": "http://123.45.67.89:8080"}  # 代理IP

response = requests.get(
    "https://example.com",
    headers=headers,
    proxies=proxies
)

（3）处理cookies和session,维持回话状态

import requests

session = requests.Session()
login_data = {"username": "your_id", "password": "your_pwd"}
session.post("https://example.com/login", data=login_data)  # 模拟登录

# 后续请求自动携带Cookies
response = session.get("https://example.com/protected-page")

（4）使用selenium模拟浏览器行为

from selenium import webdriver
driver = webdriver.Chrome()
driver.get("https://example.com")
html = driver.page_source

7、什么是反爬虫机制？如何应对动态加载内容的网页？

反爬虫机制是指网站为了防止爬虫过度访问和抓取数据而采取的一系列技术手段，这些机制旨在保护网站的安全和资源，减轻服务器压力，并确保合法用户的正常访问。
常见的反爬虫技术有：

• User-Agent检测：拦截无user-agent或使用爬虫常见UA的请求。
• IP封禁：黑名单识别并封禁爬虫常用IP或代理池。
• 验证码：弹出验证码验证人类用户。
• 动态渲染：数据通过JavaScript动态加载，传统爬虫无法直接获取。
• 行为指纹检测：分析鼠标移动、点击模式等行为特征识别爬虫。
• 数据混淆：对网页关键数据加密或动态生成DOM结构。

常见应对方案有：

• 直接调用隐藏的API，分析网页，找到返回目标数据的api接口，模拟浏览器请求获取数据。

import requests

url = "https://example.com/api/data"  # 从Network面板找到的API地址
headers = {
    "User-Agent": "Mozilla/5.0",
    "X-Requested-With": "XMLHttpRequest"  # 模拟AJAX请求
}
params = {"page": 1}  # API可能需要的参数

response = requests.get(url, headers=headers, params=params)
data = response.json()  # 获取JSON数据

• 使用无头浏览器，如Selenium、Playwright、Pyppeteer

from playwright.sync_api import sync_playwright

with sync_playwright() as p:
    browser = p.chromium.launch(headless=False)  # 显示浏览器调试
    page = browser.new_page()
    page.goto("https://example.com")
    page.wait_for_selector(".dynamic-content")  # 等待动态内容加载
    html = page.content()  # 获取渲染后的HTML
    browser.close()

• 逆向JavaScript,找到数据解密逻辑，复现js代码

import execjs

js_code = """
function decrypt(data) {
    // 从网站JS中复制的解密函数
    return data.split("").reverse().join("");
}
"""
ctx = execjs.compile(js_code)
result = ctx.call("decrypt", "edoced_46esab")  # 调用JS函数
print(result)  # 输出: "base64_code"

8、什么是HTTP状态码？写出常见的状态码和意义。

HTTP状态码是服务器在响应客户端请求时返回的三位数字代码，用于表示请求的处理结果。它属于HTTP协议的一部分，帮助客户端快速判断请求是否成功或失败，便于下一步操作。

常见状态码及意义

1XX（信息性状态码）
表示请求已被接收，需要继续处理。

• 100 Continue:客户端应继续发送请求（用于大文件上传前的确认）。
• 101 Switching Protocols:服务器同意切换协议（如从HTTP升级到WebSocket）。

2XX（成功状态码）
表示请求已成功被服务器接收、理解并处理。

• 200 OK:请求成功（如网页加载成功）。
• 201 Created:资源创建成功（常见于POST请求）。
• 204 No Content:请求成功，但响应无内容（如删除操作）。

3XX（重定向状态码）
表示需要客户端进一步操作以完成请求

• 301 Moved Permanently:资源已永久重定向（SEO会将权重转移到新URL）。
• 302 Found:资源临时重定向（下次请求可能仍用原URL）。
• 304 Not Modified:资源未修改（客户端可使用缓存）。

4XX（客户端错误状态码）
表示客户端请求有误，服务器无法处理。

• 400 Bad Request:请求语法错误（如参数格式错误）。
• 401 Unauthorized:未授权（需身份验证，如未登录）。
• 403 Forbidden:禁止访问（无权限，即使登录也可能触发）。
• 404 Not Found:资源不存在（URL错误或页面已删除）。
• 429 Too Many Requests:请求过于频繁（触发反爬机制）。

5XX（服务器错误状态码）
表示服务器处理请求时出错。

• 500 Internal Server Error:服务器内部错误（如代码崩溃）。
• 502 Bad Gateway:网关错误（代理服务器无法从上游获取响应）。
• 503 Service Unavailable:服务不可用（服务器过载或维护）。
• 504 Gateway Timeout:网关超时（代理服务器等待上游响应超时）。

9、设计一个爬虫系统，要求支持分布式爬取，数据存储和去重。请描述你的设计方案。

本系统采用主从式分布式架构，包含以下核心组件：

（1）调度中心(Master)：负责任务分配、节点管理和URL去重

class Scheduler:
    def __init__(self):
        self.task_queue = RedisQueue('crawler_tasks')  # 待抓取URL队列
        self.visited_urls = BloomFilter()  # 已访问URL布隆过滤器
        self.workers = {}  # 活跃Worker节点
    
    def add_task(self, url, priority=0):
        if not self.visited_urls.contains(url):
            self.task_queue.push(url, priority)
            self.visited_urls.add(url)
    
    def assign_task(self):
        while True:
            task = self.task_queue.pop()
            available_worker = self.get_available_worker()
            available_worker.assign_task(task)

（2） 爬虫节点(Worker)：执行实际的网页抓取和数据处理

class CrawlerWorker:
    def __init__(self, worker_id):
        self.worker_id = worker_id
        self.http_client = AsyncHTTPClient()
        self.parser = HtmlParser()
    
    async def crawl(self, url):
        try:
            # 1. 下载网页
            response = await self.http_client.fetch(url)
            
            # 2. 解析内容
            data = self.parser.parse(response.body)
            
            # 3. 提取新URL
            new_urls = self.parser.extract_links(response.body)
            
            # 4. 返回结果
            return {
                'url': url,
                'data': data,
                'new_urls': new_urls
            }
        except Exception as e:
            log_error(f"Failed to crawl {url}: {str(e)}")
            raise

（3）消息队列：实现任务分发和节点间通信

# RabbitMQ配置
RABBITMQ_CONFIG = {
    'host': 'mq.master',
    'port': 5672,
    'user': 'crawler',
    'password': 'password',
    'task_queue': 'crawler_tasks',
    'result_queue': 'crawler_results'
}

（4） 分布式存储：存储爬取结果

# MySQL表结构
CREATE TABLE crawled_pages (
    id BIGINT PRIMARY KEY AUTO_INCREMENT,
    url VARCHAR(1024) NOT NULL,
    title VARCHAR(512),
    content TEXT,
    html LONGTEXT,
    crawl_time DATETIME,
    md5 CHAR(32),  # 内容指纹
    UNIQUE KEY (md5),
    INDEX (url(255))
);

（5）去重服务：实现URL和内容去重

class DeduplicationService:
    def __init__(self):
        self.redis = RedisClient()
        self.bloom = ScalableBloomFilter()
    
    def is_url_duplicate(self, url):
        # 先查布隆过滤器，减少Redis访问
        if not self.bloom.contains(url):
            return False
        # 再确认Redis精确去重
        return self.redis.sismember('visited_urls', url)
    
    def is_content_duplicate(self, content):
        content_md5 = hashlib.md5(content.encode()).hexdigest()
        return self.redis.sismember('content_fingerprints', content_md5)

（6） 监控系统：监控系统运行状态

10、列举常用的shell命令。

• 文件与目录操作

• 文件内容处理

• 系统信息与进程管理

• 网络操作

• 压缩与解压

• 权限管理

• 其他实用命令