高级缓存策略实现更智能的数据提取

网络搜索是数据收集、研究和竞争分析的重要工具。然而，在扫描大量数据时，效率低下可能会成为一个大问题。缓慢的加载时间、重复请求和数据检索延迟都会拖慢整个过程。要解决这些问题，缓存是一个强大的解决方案。

缓存可临时存储数据，因此无需再次获取，从而加快了刮擦过程。它能让网络刮擦程序更快地访问以前收集的数据，从而提高效率。在本文中，我们将深入探讨高级缓存策略，以提高网络刮擦任务的速度和效率，确保更顺利、更快速地收集数据。

缓存为何对网络抓取很重要？

在深入探讨具体策略之前，我们先来了解一下为什么缓存对网络刮擦操作至关重要。

冗余请求的代价

您的刮擦器发出的每个 HTTP 请求都会产生成本。这包括带宽消耗、服务器处理时间、网络延迟，最重要的是，还有被限速或禁用的风险。当你抓取成千上万个页面时，这些成本就会成倍增加。

考虑一下这种情况：您正在扫描一个电子商务网站，以跟踪价格变化。产品描述很少变化，但您每次都要获取它们。这既浪费了带宽，又增加了不必要的服务器负载。有了适当的缓存，您就可以存储这些静态元素，只获取实际变化的内容–价格。

遵守速率限制和 Robots.txt

合乎道德的网络搜刮意味着要尊重目标网站的资源。许多网站都有速率限制，过度请求可能导致 IP 禁止或法律问题。缓存可显著减少请求量，使您在收集所需数据的同时，保持在可接受的范围内。

提高刮板性能

缓存良好的搜刮程序运行速度更快，消耗的资源更少，扩展效率更高。当您的搜索器可以从本地存储中提取数据而不是进行网络请求时，响应时间就会从几秒缩短到几毫秒。在处理大型数据集或运行对时间敏感的操作时，这种性能提升至关重要。

网络抓取器的核心缓存策略

让我们来探讨一下每个网络搜刮开发人员都应该了解的基本缓存方法。

1.HTTP 响应缓存

HTTP 响应缓存是高效网络搜刮的基础。这种策略将整个 HTTP 响应存储在本地，从而消除了对相同资源的冗余网络请求。

使用 Python 和 Requests-Cache 实现：

进口请求

导入请求缓存

从 datetime 导入 timedelta

# 初始化缓存，过期时间为 24 小时

requests_cache.install_cache(

scraper_cache’、

backend=’sqlite’、

expire_after=timedelta(hours=24)

)

def scrape_product_page(url)：

“””

自动缓存获取产品页面

“””

response = requests.get(url)

# 检查响应是否来自缓存

如果 hasattr(response,’from_cache’)和 response.from_cache：

print(f “Cache hit for: {url}”)

否则

print(f “Fresh request for: {url}”)

return response.text

# 第一次调用 – 发出实际 HTTP 请求

content = scrape_product_page(‘https://example.com/product/123’)

# 第二次调用 – 即刻从缓存中提供服务

cached_content = scrape_product_page(‘https://example.com/product/123’)

这种简单的方法可将您的请求量减少 50-80%，适用于重新访问许多页面的典型刮擦情况。

高级配置：

导入请求缓存

从 requests_cache 导入 CachedSession

# 使用自定义缓存设置创建会话

session = CachedSession(

cache_name=’advanced_cache’、

backend=’redis’, # 使用 Redis 进行分布式缓存

expire_after={

‘*.example.com/api/*’: timedelta(minutes=5), # 应用程序接口响应很快过期

‘*.example.com/products/*’: timedelta(days=1), # 产品页面持续时间更长

‘*.example.com/static/*’: timedelta(days=30), # 静态资产时间最长

},

allowable_methods=(‘GET’, ‘POST’), # 也缓存 POST 请求

allowable_codes=(200, 404), # 缓存成功和 404 响应

match_headers=True, # 匹配缓存响应时考虑标题

ignored_parameters=[‘session_id’, ‘timestamp’] # 忽略这些查询参数

)

# 使用会话处理所有请求

response = session.get(‘https://example.com/product/456’)

2.基于 URL 的自定义密钥缓存

有时，您需要对缓存内容和缓存时间进行更多控制。通过自定义密钥生成基于 URL 的缓存，可以实现复杂的缓存逻辑。

导入 hashlib

导入 json

导入时间

从 pathlib 导入路径

类 URLCache：

def __init__(self, cache_dir=’./cache’, default_ttl=3600)：

self.cache_dir = Path(cache_dir)

self.cache_dir.mkdir(exist_ok=True)

self.default_ttl = default_ttl

def _generate_cache_key(self, url, params=None)：

“””

根据 URL 和参数生成唯一的缓存密钥

“””

cache_string = url

如果参数

# 对参数进行排序，以确保散列一致

cache_string += json.dumps(params, sort_keys=True)

return hashlib.md5(cache_string.encode()).hexdigest()

def get(self, url, params=None, ttl=None)：

“””

如果有效，则读取缓存响应

“””

cache_key = self._generate_cache_key(url, params)

cache_file = self.cache_dir / f”{cache_key}.json”

if not cache_file.exists()：

返回 None

with open(cache_file, ‘r’) as f：

缓存数据 = json.load(f)

# 检查缓存是否过期

ttl = ttl 或 self.default_ttl

age = time.time() – cached_data[‘timestamp’]

如果 age > ttl：

cache_file.unlink() # 删除过期缓存

返回 None

return cached_data[‘content’]

def set(self, url, content, params=None)：

“””

将响应存储在缓存中

“””

cache_key = self._generate_cache_key(url, params)

cache_file = self.cache_dir / f”{cache_key}.json”

cache_data = {

url’： 网址、

参数’： 参数、

内容’： 内容、

时间戳”： time.time()

}

with open(cache_file, ‘w’) as f：

json.dump(cache_data, f)

def clear_expired(self, max_age=None)：

“””

删除所有过期的缓存条目

“””

max_age = max_age 或 self.default_ttl

current_time = time.time()

for cache_file in self.cache_dir.glob(‘*.json’)：

with open(cache_file, ‘r’) as f：

缓存数据 = json.load(f)

如果 current_time – cached_data[‘timestamp’] > max_age：

cache_file.unlink()

# 使用示例

cache = URLCache(cache_dir=’./scraper_cache’, default_ttl=86400)

def scrape_with_cache(url, params=None)：

# 尝试从缓存中获取

cached_content = cache.get(url, params, ttl=3600)

如果 cached_content：

print(f “Serving from cache: {url}”)

返回缓存内容

# 重新提出申请

response = requests.get(url, params=params)

content = response.text

# 存储在缓存中

cache.set(url, content, params)

print(f “Cached new response: {url}”)

返回内容

3.解析数据缓存

原始 HTML 缓存非常有用，但通常情况下，您需要缓存经过解析的结构化数据。这种方法既能节省网络请求，又能节省解析时间。

导入泡菜

从 bs4 导入 BeautifulSoup

从 pathlib 导入路径

类 ParsedDataCache：

def __init__(self, cache_dir=’./parsed_cache’)：

self.cache_dir = Path(cache_dir)

self.cache_dir.mkdir(exist_ok=True)

def _get_cache_path(self, url)：

“”从 URL 生成缓存文件路径”””

safe_name = hashlib.md5(url.encode()).hexdigest()

return self.cache_dir / f”{safe_name}.pkl”

def get_parsed_data(self, url, parser_func, force_refresh=False)：

“””

从缓存中获取解析数据，或重新解析

参数

url：URL to scrape

parser_func：解析 HTML 并返回结构化数据的函数

force_refresh（强制刷新）：即使已缓存，也要强制刷新

“””

cache_path = self._get_cache_path(url)

# 如果有缓存数据，则返回缓存数据，不强制刷新

如果 cache_path.exists()，而不是 force_refresh：

with open(cache_path, ‘rb’) as f：

缓存数据 = pickle.load(f)

print(f “Loaded parsed data from cache: {url}”)

返回缓存数据

# 获取并解析新数据

response = requests.get(url)

soup = BeautifulSoup(response.text, ‘html.parser’)

解析的数据 = 解析器_func(汤)

# 缓存解析后的数据

with open(cache_path, ‘wb’) as f：

pickle.dump(parsed_data, f)

print(f “Parsed and cached new data: {url}”)

返回解析后的数据

# 剖析器函数示例

def parse_product_data(soup)：

“”从 HTML 中提取结构化产品数据”””

返回 {

title’: soup.select_one(‘.product-title’).text.strip()、

价格’： float(soup.select_one(‘.price’).text.strip().replace(‘$’, ”))、

description’： soup.select_one(‘.description’).text.strip()、

图像[img[‘src’] for img in soup.select(‘.product-image img’)]、

评分’： float(soup.select_one(‘.rating’)[‘data-rating’])、

reviews_count’: int(soup.select_one(‘.reviews-count’).text.split()[0])

}

# 使用方法

cache = ParsedDataCache()

product_data = cache.get_parsed_data(

‘https://example.com/product/789’、

解析产品数据

)

print(f “Product: {product_data[‘title’]}”)

print(f “Price: ${product_data[‘price’]}”)

4.智能缓存失效策略

知道何时使缓存失效与缓存本身一样重要。不同的内容类型需要不同的失效策略。

from enum import Enum

从 datetime 导入 datetime、timedelta

class CacheStrategy(Enum)：

STATIC = ‘static’ # 很少更改（隐私政策、关于页面）

SEMI_STATIC = ‘semi_static’ # 偶尔更改（产品说明）

动态 = ‘动态’ # 经常变化（价格、库存水平）

REAL_TIME = ‘real_time’ # 不断变化（即时比分、加密货币价格）

类 SmartCache：

def __init__(self)：

self.strategy_ttls = {

CacheStrategy.STATIC: timedelta(days=30)、

CacheStrategy.SEMI_STATIC: timedelta(days=7)、

CacheStrategy.DYNAMIC: timedelta(hours=1)、

CacheStrategy.REAL_TIME: timedelta(minutes=5)

}

self.cache = {}

def _determine_strategy(self, url)：

“””

根据 URL 模式自动确定缓存策略

“””

如果 url 中有”/static/”或 url 中有”/about “或 url 中有”/terms”：

返回 CacheStrategy.STATIC

如果 url 中有”/product/”，而 url 中没有”/reviews”：

返回 CacheStrategy.SEMI_STATIC

elif url 中的”/price “或 url 中的”/stock”：

返回 CacheStrategy.DYNAMIC

elif url 中的”/live “或”/real-time”：

返回 CacheStrategy.REAL_TIME

否则

return CacheStrategy.DYNAMIC # 默认为动态

def get(self, url, strategy=None)：

“”通过智能失效功能获取缓存内容”””

if url not in self.cache：

返回 None

缓存条目 = self.cache[url]

strategy = strategy 或 self._determine_strategy(url)

ttl = self.strategy_ttls[strategy]（策略

age = datetime.now() – cached_entry[‘timestamp’]

如果 age > ttl：

del self.cache[url]

返回 None

return cached_entry[‘content’]

def set(self, url, content, strategy=None)：

“”缓存带有策略元数据的内容”””

strategy = strategy 或 self._determine_strategy(url)

self.cache[url] = {

内容’： 内容、

timestamp’: datetime.now()、

战略”： 战略

}

def invalidate_pattern(self, pattern)：

“””

使与模式匹配的所有缓存条目失效

当您知道某些内容已更改时，该功能非常有用

“””

urls_too_remove = [

url for url in self.cache.keys()

如果 url 中有模式

]

for url in urls_to_remove：

del self.cache[url]

return len(urls_too_remove)

# 使用示例

smart_cache = SmartCache()

# 这些 URL 将自动获得适当的缓存持续时间

urls = [

https://example.com/about’, # 缓存 30 天

https://example.com/product/123′, # 缓存 7 天

https://example.com/product/123/price’, # 缓存 1 小时

https://example.com/live/stock-ticker’ # 缓存 5 分钟

]

for url in urls：

content = smart_cache.get(url)

如果没有内容：

content = requests.get(url).text

smart_cache.set(url, content)

生产型抓取机的高级缓存技术

一旦你掌握了基础知识，这些高级技术将使你的刮擦操作更上一层楼。

5.使用 Redis 进行分布式缓存

对于在多台机器上运行的大规模刮擦操作，分布式缓存可确保您的所有刮擦程序都能从共享缓存数据中获益。

导入 redis

导入 json

从 datetime 导入 timedelta

类 RedisCache：

def __init__(self, host=’localhost’, port=6379, db=0)：

self.redis_client = redis.Redis(

host=主机、

port=port、

db=db、

解码回复=True

)

def get(self, key)：

“”从 Redis 缓存读取值”””

value = self.redis_client.get(key)

if value：

return json.loads(value)

返回 None

def set(self, key, value, ttl=3600)：

“”将值存储在 Redis 中并过期”””

self.redis_client.setex(

键、

timedelta(seconds=ttl)、

json.dumps(value)

)

def get_or_fetch(self, key, fetch_func, ttl=3600)：

“””

从缓存中获取或执行获取函数并缓存结果

“””

cached_value = self.get(key)

如果 cached_value 不是 None：

返回缓存值

# 获取新数据

fresh_value = fetch_func()

self.set(key, fresh_value, ttl)

返回 fresh_value

def increment_counter(self, key, amount=1)：

“””

原子计数器增量 – 用于限制速率

“””

return self.redis_client.incr(key, amount)

def set_with_lock(self, key, value, ttl=3600, lock_timeout=10)：

“””

用分布式锁设置值，防止缓存踩踏

“””

lock_key = f “lock:{key}”

lock = self.redis_client.lock(lock_key, timeout=lock_timeout)

试试看

如果 lock.acquire(blocking=False)：

self.set(key, value, ttl)

返回 True

返回 False

终于

试试看

lock.release()

除外：

通过

# 示例：在分布式刮板中使用 Redis 缓存

redis_cache = RedisCache(host=’redis-server.example.com’)

def scrape_with_distributed_cache(url)：

“””

多个刮板实例共享缓存进行刮板搜索

“””

cache_key = f “scraper:{hashlib.md5(url.encode()).hexdigest()}”

# 尝试从分布式缓存中获取

cached_content = redis_cache.get(cache_key)

如果 cached_content：

返回缓存内容

# 使用锁定以防止多个实例同时进行刮擦

lock_key = f “lock:{cache_key}”

lock = redis_cache.redis_client.lock(lock_key, timeout=30)

如果 lock.acquire(blocking=True, blocking_timeout=10)：

试试看

# 获取锁定后再次检查缓存

cached_content = redis_cache.get(cache_key)

如果 cached_content：

返回缓存内容

# 获取新数据

response = requests.get(url)

content = response.text

# 缓存所有实例

redis_cache.set(cache_key, content, ttl=3600)

返回内容

终于

lock.release()

6.使用 ETags 和 Last-Modified 标头进行条件缓存

利用 HTTP 的内置缓存机制，最大限度地减少带宽，同时确保数据的新鲜度。

进口请求

从 datetime 导入 datetime

类 ConditionalCache：

def __init__(self)：

self.cache = {}

def fetch_with_conditional(self, url)：

“””

使用条件请求（ETags 和 Last-Modified）获取 URL

“””

cache_entry = self.cache.get(url, {})

标题 = {}

# 如果有缓存数据，则添加条件标头

如果 cache_entry 中有 “etag”，则

headers[‘If-None-Match’] = cache_entry[‘etag’]

如果 cache_entry 中有 “last_modified”（最后修改）：

headers[‘If-Modified-Since’] = cache_entry[‘last_modified’]

# 提出带有条件标头的请求

response = requests.get(url, headers=headers)

# 304 未修改 – 内容未更改

如果 response.status_code == 304：

print(f “Content not modified: {url}”)

return cache_entry[‘content’]

# 内容已更改或不存在缓存

如果 response.status_code == 200：

# 使用新内容和验证标头更新缓存

self.cache[url] = {

content”： response.text、

‘etag’： response.headers.get(‘ETag’)、

last_modified’： response.headers.get(‘Last-Modified’)、

timestamp’: datetime.now()

}

print(f “Fetched fresh content: {url}”)

return response.text

raise Exception(f “Unexpected status code: {response.status_code}”)

# 使用方法

conditional_cache = ConditionalCache()

# 第一次请求 – 获取全部内容

content1 = conditional_cache.fetch_with_conditional（’https://example.com/api/data’）

# 第二次请求 – 如果内容未变，可能返回 304 Not Modified

# 这样可以节省带宽，因为服务器不会发送完整的响应

content2 = conditional_cache.fetch_with_conditional(‘https://example.com/api/data’)

7.缓存预热和抢先刷新

无需等待缓存过期。在过期前主动刷新热门内容，确保关键数据不会错过缓存。

导入线程

导入时间

从队列导入队列

类 CacheWarmer：

def __init__(self, cache, refresh_threshold=0.8)：

“””

参数

缓存：您的缓存实例

刷新阈值：当缓存年龄达到 TTL 的这一分数时刷新

“””

self.cache = cache

self.refresh_threshold = refresh_threshold

self.refresh_queue = Queue()

self.running = False

def start_worker(self)：

“””启动后台工作者进行缓存预热”””

self.running = True

worker_thread = threading.Thread(target=self._refresh_worker)

worker_thread.daemon = True

worker_thread.start()

def _refresh_worker(self)：

“”刷新缓存条目的后台工作者”””

while self.running：

试试看

url, fetch_func, ttl = self.refresh_queue.get(timeout=1)

# 获取新数据

fresh_data = fetch_func(url)

# 更新缓存

self.cache.set(url, fresh_data, ttl)

print(f “Preemptively refreshed: {url}”)

除外：

继续

def get_with_warming(self, url, fetch_func, ttl=3600)：

“””

通过抢先刷新从缓存中获取

“””

cached_entry = self.cache.get_full_entry(url) # 获取元数据

如果缓存条目

# 检查缓存是否即将过期

age = time.time() – cached_entry[‘timestamp’]

如果 age > (ttl * self.refresh_threshold)：

# 为后台刷新排队

self.refresh_queue.put((url, fetch_func, ttl))

return cached_entry[‘content’]

# 无缓存 – 同步获取

fresh_data = fetch_func(url)

self.cache.set(url, fresh_data, ttl)

返回 fresh_data

def stop(self)：

“”停止后台工作者”””

self.running = False

8.分层缓存层

将多个缓存层结合起来，实现最佳性能–内存缓存用于热数据，磁盘缓存用于热数据，网络请求用于冷数据。

类 HierarchicalCache：

def __init__(self, memory_size=1000, disk_cache_dir=’./cache’)：

# 第 1 层：内存高速缓存（最快、最小）

self.memory_cache = {}

self.memory_size = memory_size

self.access_counts = {}

# 第 2 层：磁盘缓存（速度较慢、较大）

self.disk_cache = URLCache(cache_dir=disk_cache_dir)

# 第 3 层：Redis 缓存（分布式、持久性）

self.redis_cache = RedisCache()

def _evict_from_memory(self)：

“”从内存缓存中删除使用频率最低的项目”””

if len(self.memory_cache) >= self.memory_size：

lfu_key = min(self.access_counts, key=self.access_counts.get)

删除 self.memory_cache[lfu_key]

删除 self.access_counts[lfu_key]

def get(self, url)：

“”从缓存层级中获取数据””

# 第 1 层：检查内存缓存

if url in self.memory_cache：

self.access_counts[url] = self.access_counts.get(url, 0) + 1

print(f “Memory cache hit: {url}”)

return self.memory_cache[url]

# 第 2 层：检查磁盘缓存

disk_content = self.disk_cache.get(url)

如果 disk_content：

print(f “Disk cache hit: {url}”)

# 提升至内存缓存

self._evict_from_memory()

self.memory_cache[url] = disk_content

self.access_counts[url] = 1

return disk_content

# 第 3 层：检查 Redis 缓存

redis_content = self.redis_cache.get(f “scraper:{url}”)

if redis_content：

print(f “Redis 缓存命中：{url}”)

# 提升到磁盘和内存

self.disk_cache.set(url, redis_content)

self._evict_from_memory()

self.memory_cache[url] = redis_content

self.access_counts[url] = 1

返回 redis_content

返回 None

def set(self, url, content)：

“””跨所有缓存层存储数据”””

# 储存在所有层中

self._evict_from_memory()

self.memory_cache[url] = content

self.access_counts[url] = 1

self.disk_cache.set(url, content)

self.redis_cache.set(f “scraper:{url}”, content, ttl=86400)

# 使用方法

hierarchical_cache = HierarchicalCache(memory_size=500)

def smart_scrape(url)：

# 尝试分级缓存

content = hierarchical_cache.get(url)

如果 content：

返回内容

# 获取新数据

response = requests.get(url)

content = response.text

# 跨所有缓存层存储

hierarchical_cache.set(url, content)

返回内容

最佳实践和优化技巧

缓存键设计

缓存键策略可以决定缓存实施的成败。以下是一些行之有效的模式：

def generate_cache_key(url, method=’GET’, headers=None, body=None)：

“””

生成考虑到所有请求方面的综合缓存密钥

“””

key_components = [

方法、

url、

json.dumps(sorted(headers.items())) if headers else ”、

hashlib.md5(body.encode()).hexdigest() if body else ”

]

key_string = ‘|’.join(key_components)

return hashlib.sha256(key_string.encode()).hexdigest()

监控缓存性能

跟踪缓存效果，优化设置：

class CacheMetrics：

def __init__(self)：

self.hits = 0

self.misses = 0

self.evictions = 0

self.size = 0

@ 属性

def hit_rate(self)：

总计 = self.hits + self.misses

return self.hits / total if total > 0 else 0

def record_hit(self)：

self.hits += 1

def record_miss(self)：

self.misses += 1

def report(self)：

返回 {

hit_rate’： f”{self.hit_rate:.2%}”、

total_requests’: self.hits + self.misses、

hits’: self.hits、

失误”： self.misses、

驱逐”： self.evictions、

缓存大小”： self.size

}

尊重缓存控制标头

始终尊重目标网站的缓存首选项：

def parse_cache_control(response)：

“””

解析 Cache-Control 标头并确定适当的 TTL

“””

cache_control = response.headers.get(‘Cache-Control’, ”)

# 检查无缓存或无存储指令

如果 cache_control 中有 “no-cache “或 “no-store”：

return 0 # 不缓存

# 提取 max-age 指令

for directive in cache_control.split(‘,’)：

directive = directive.strip()

如果 directive.startswith(‘max-age=’)：

return int(directive.split(‘=’)[1])

# 检查 Expires 标头作为备用

expires = response.headers.get(‘Expires’)

如果过期：

# 计算到期前的秒数

# 实施取决于日期解析

通过

return 3600 # 默认为 1 小时

最后的话

实施有效的缓存策略可将网络搜刮从资源密集型操作转变为高效、可扩展的系统。通过结合 HTTP 响应缓存、智能失效策略、分布式缓存和分层，你可以显著降低带宽消耗，提高刮擦性能，并保持在符合道德规范的刮擦范围内。

请记住这些关键原则：

• 从简单的基本 HTTP 缓存开始，根据需要逐步提高复杂性
• 使缓存持续时间与内容更新频率相匹配
• 监控缓存性能，优化命中率
• 尊重 robots.txt和 cache-control 标头
• 使用分布式缓存进行多实例部署
• 为关键数据路径实施缓存预热

正确的缓存策略不仅仅是让你的搜刮器更快，而是要建立一个可持续的、受人尊重的、高效的数据提取系统，并能根据你的需求进行扩展。无论您是在抓取少量页面还是数百万页面，正确的缓存都将是您取得长期成功的最有价值的工具。

从这里概述的基本策略开始，测量你的缓存命中率，并根据你的具体使用情况不断改进。您未来的自己和您正在抓取的网站都会感谢您对适当缓存基础设施的投资。

常见问题