当前位置：首页 > news >正文

从零搭建高效本地代理池：设计与实现

news 来源：原创 2025/6/27 9:28:04

从零搭建高效本地代理池：设计与实现

一. 背景

XXX代理服务是一个专门拉取第三方代理IP的服务，最初仅对组内系统提供支持。随着业务规模的扩展，该服务需要面向部门内其他组提供服务。然而，由于不同业务场景对代理IP的需求量差异较大（如单次拉取100条、30条或50条），原有系统设计中每次拉取代理都需要向第三方代理服务发送一次请求。

在高并发场景下，系统频繁从第三方代理提供商获取代理IP，以满足内部应用的需求。然而，由于QPS（每秒查询数）过高，频繁访问第三方接口导致以下问题：

机器频繁重启：高并发请求对系统资源造成巨大压力。由于内部其他组应用采用单IDC部署，并启用了就近访问策略，导致流量集中到一个集群，进而引发该集群下机器频繁重启。
I/O 等待时间长：每次请求第三方接口都会引入网络延迟，接口耗时显著增加，影响整体性能。
资源浪费：内部应用单次可能仅需拉取少量代理IP（如100条、30条或50条），但每次都需要访问第三方提供商，增加了不必要的网络开销。

为了解决这些问题，我们决定采用预拉取的方式，将代理IP缓存到本地代理池中，从而减少对第三方接口的依赖，提升系统的稳定性和性能。

二. 方案选型

1. 代理缓存池的核心需求

在设计本地代理池时，我们明确了以下核心需求：
- 高效性：支持高并发场景下的快速读取和写入。
- 线程安全：多个线程可能同时访问代理池，需要保证线程安全。
- 主动更新：当缓存池中的代理数量低于单次拉取的数量时，需要主动触发拉取操作。
- 定时更新：代理池需要定期从第三方提供商拉取新的代理IP，确保代理的时效性。
- 缓存管理：需要对代理池的大小进行限制，避免内存占用过高。

2. 技术选型

在方案选型中，我们主要考虑了两种存储方式：Redis 存储 和 本地存储。

Redis 存储方案

优势：

分布式支持：Redis 是一个分布式存储系统，代理池可以在多个服务实例之间共享。
持久化能力：Redis 支持数据持久化（如 RDB 和 AOF），即使服务重启，代理池数据也不会丢失。

劣势：

额外的网络开销：使用 Redis 存储代理池需要通过网络与 Redis 服务通信，这会引入额外的网络延迟，尤其是在高并发场景下，频繁访问 Redis 会导致性能瓶颈。
依赖外部服务：Redis 是一个独立的服务，需要额外的运维成本（如部署、监控、扩容等）。如果 Redis 服务出现问题（如网络分区、节点故障），会影响代理池的可用性。
复杂性增加：引入 Redis 会增加系统的复杂性。

本地存储方案

优势：

高性能：本地存储（如 ConcurrentLinkedQueue）直接在内存中操作数据，避免了网络通信的开销，性能更高，尤其适合高并发场景。
简单易用：本地存储的实现简单，无需额外的依赖或服务，减少了系统的复杂性和运维成本。
线程安全：使用 ConcurrentLinkedQueue 等线程安全的数据结构，可以轻松实现高并发场景下的代理池管理。
低延迟：本地存储的读写操作延迟极低，适合对性能要求较高的场景。

劣势：

数据易失性：本地存储的数据存储在内存中，服务重启后数据会丢失。
内存占用：本地存储的代理池大小受限于 JVM 的内存，需要合理设置代理池的容量，避免内存溢出。

最终选择：本地存储

基于上述分析，我们最终选择了本地存储方案，主要原因如下：

高并发场景下的性能需求：本地存储的低延迟和高性能能够满足高并发场景下的快速获取代理 IP 的需求，而 Redis 的网络开销可能会成为性能瓶颈。
减少外部依赖：使用 Redis 会引入额外的外部服务依赖，增加系统的复杂性和运维成本。本地存储无需依赖外部服务，部署和维护更加简单。
避免网络延迟：Redis 的网络通信会引入额外的延迟，而本地存储直接在内存中操作数据，能够显著降低 IO 等待时间，提升整体性能。
数据易失性和内存占用：该问题可以通过合理控制本地缓存池存储的容量来规避,该场景下可以忽略。

三. 实现

1. 原有设计：远程代理获取接口

主要内容

在原有设计中，我们定义了一个接口 IRemoteProxiesProvider，用于从第三方代理提供商远程获取代理 IP。该接口的核心方法如下：

public interface IRemoteProxiesProvider {/*** 获取代理** @param limit  单次拉取的代理数量* @param channel 调用方* @return List<String> 代理 IP 列表*/List<String> fetchProxies(Integer limit, String channel);/*** 获取代理类型** @return ProxyTypeEnum 代理类型*/ProxyTypeEnum getProxyType();
}

作用

远程代理获取：该接口定义了从第三方代理提供商拉取代理 IP 的方法，支持按需拉取指定数量的代理。

2. 定义本地缓存代理池管理接口

主要内容

为了支持本地缓存代理池的功能，我们定义了一个接口 ICacheProxiesProvider，用于管理本地代理池的刷新和获取操作。该接口的核心方法如下：

public interface ICacheProxiesProvider {/*** 是否开启缓存代理** @return boolean 是否开启缓存代理*/default boolean enableCacheProxies() {return false;}/*** 缓存代理是否已满** @return boolean 缓存代理是否已满*/boolean isFull();/*** 刷新代理到缓存池*/void refreshProxies();/*** 从缓存池中获取代理** @param limit 单次拉取的代理数量* @return List<String> 代理 IP 列表*/List<String> getCacheProxies(Integer limit);
}

作用

缓存管理：通过 isFull 和 enableCacheProxies 方法，控制本地缓存池的容量和开关。
动态刷新：通过 refreshProxies 方法，支持主动刷新代理到本地缓存池。
本地获取：通过 getCacheProxies 方法，支持从本地缓存池中获取代理 IP。

3. 定义抽象类：聚合远程获取和本地获取代理的能力

主要内容

为了统一管理远程代理获取和本地缓存代理的功能，我们设计了一个抽象类 AbstractProxiesProvider，该类实现了 IRemoteProxiesProvider 和 ICacheProxiesProvider 接口。以下是核心实现：

public abstract class AbstractProxiesProvider implements IRemoteProxiesProvider, ICacheProxiesProvider {private final ILog LOGGER = LogManager.getLogger(this.getClass());// 本地缓存代理存储队列private final ConcurrentLinkedQueue<CacheProxyInfo> PROXY_POOL = new ConcurrentLinkedQueue<>();// 异步刷新缓存锁,同一时刻只允许一个线程刷新private final ReentrantLock ASYNC_REFRESH_PROXY_LOCK = new ReentrantLock();/*** @param limit* @param channel* @param fetchProxySource* @return java.util.List<java.lang.String>* 获取代理* @author ljm* @date 11:25 2025/6/17**/public List<String> fetchProxies(Integer limit, String channel, FetchProxySource fetchProxySource) {ProxyHitType proxyHitType = ProxyHitType.REMOTE;try {if (StatusUtil.statusIn(fetchProxySource, FetchProxySource.REMOTE)) {return fetchProxies(limit, channel);}// 先走本地缓存,本地缓存不足则走远程List<String> cacheProxies = getCacheProxies(limit);if (CollectionUtils.isNotEmpty(cacheProxies)) {proxyHitType = ProxyHitType.LOCAL_CACHE;LOGGER.info("useCacheProxies", String.format("代理:%s拉取缓存池中代理:%s条", this.getProxyType().getName(), cacheProxies.size()));return cacheProxies;}return fetchProxies(limit, channel);} finally {MetricUtils.metricProxyHitType(this.getProxyType().getName(), channel, proxyHitType.name(), fetchProxySource.name());}}@Overridepublic boolean enableCacheProxies() {return ProxyHelper.isProxiesRefreshToCache(this.getProxyType().getName());}@Overridepublic boolean isFull() {return PROXY_POOL.size() >= ProxyHelper.getProxiesRefreshToCacheSize(this.getProxyType().getName());}@Overridepublic void refreshProxies() {String proxyName = this.getProxyType().getName();boolean lockSuccess = ASYNC_REFRESH_PROXY_LOCK.tryLock();if (!lockSuccess) {LOGGER.info("RefreshingProxy", String.format("存在另一线程在刷新%s代理", proxyName));return;}try {// 满了或者未开启缓存代理if (isFull() || !enableCacheProxies()) {return;}List<String> proxies = fetchProxies(ProxyHelper.getProxiesRefreshToCacheSize(proxyName),"SystemRefresh");if (CollectionUtils.isNotEmpty(proxies)) {long expireSeconds = ProxyHelper.getCacheProxiesExpireSeconds(proxyName);List<CacheProxyInfo> cacheProxyInfoList = proxies.stream().map(proxy -> new CacheProxyInfo(proxy, LocalDateTime.now().plusSeconds(expireSeconds))).toList();PROXY_POOL.addAll(cacheProxyInfoList);LOGGER.info("RefreshProxiesToCacheSuccess", String.format("代理:%s刷新%s条ip到缓存成功!", this.getProxyType().getName(),proxies.size()));}} finally {ASYNC_REFRESH_PROXY_LOCK.unlock();}}@Overridepublic List<String> getCacheProxies(Integer limit) {// 未开启缓存代理if (!enableCacheProxies()) {return Collections.emptyList();}boolean refreshProxies = false;try {List<String> result = new ArrayList<>();if (PROXY_POOL.size() < limit) {refreshProxies = true;return result;}for (int i = 0; i < limit; i++) {CacheProxyInfo cacheProxyInfo = PROXY_POOL.poll();if (cacheProxyInfo == null || isExpired(cacheProxyInfo.getExpireTime())) {continue;}result.add(cacheProxyInfo.getProxy());}if (result.size() < limit) {// 如果数量不足，异步刷新refreshProxies = true;}return result;} finally {if (refreshProxies) {LOGGER.info("CacheProxiesSizeIsNotEnough", String.format("代理商:%s缓存的代理数量不足,触发异步刷新", this.getProxyType().getName()));Asyncio.run(this::refreshProxies);}}}/*** @return boolean* 是否过期* @author ljm* @date 10:21 2025/6/17**/boolean isExpired(LocalDateTime expireTime) {LocalDateTime now = LocalDateTime.now();return !now.isBefore(expireTime);}
}

作用

聚合能力：抽象类统一管理远程代理获取和本地缓存代理的功能，避免重复实现。
线程安全：ConcurrentLinkedQueue，保证多线程环境下拉取和缓存代理的线程安全。
保护下游：通过 ReentrantLock 保证同一代理商同一时刻内只有一个线程在刷新代理操作，减少对第三方的调用。
主动刷新：当本地缓存池中的代理数量不足时，自动触发刷新操作。
过期清理：通过 isExpired 方法，移除已过期的代理。

4. 代理管理器：获取代理和定时刷新代理

主要内容

为了统一管理所有代理提供商的代理获取和刷新操作，我们设计了一个代理管理器 ProxyManager，该类的核心实现如下：

@Service
public class ProxyManager {private final ILog LOGGER = LogManager.getLogger(this.getClass());private final Map<ProxyTypeEnum, AbstractProxiesProvider> FETCH_PROXY_MAP = new HashMap<>();private static final ScheduledExecutorService SCHEDULED_EXECUTOR_SERVICE = new ScheduledThreadPoolExecutor(1, new ThreadFactoryBuilder().setNameFormat("ScheduleProxyRefresher-%d").build(), new ThreadPoolExecutor.DiscardPolicy());@Autowiredpublic ProxyManager(List<AbstractProxiesProvider> proxyProviders) {for (AbstractProxiesProvider proxyProvider : proxyProviders) {FETCH_PROXY_MAP.put(proxyProvider.getProxyType(), proxyProvider);}}public List<String> getProxyListByProxyType(ProxyTypeEnum proxyTypeEnum, Integer limit, String channel) {AbstractProxiesProvider iFetchProxy = FETCH_PROXY_MAP.get(proxyTypeEnum);if (iFetchProxy == null) {LOGGER.info("getProxyListByProxyType", String.format("未找到代理提供商,代理类型%s", proxyTypeEnum));return new ArrayList<>();}return iFetchProxy.fetchProxies(limit, channel, FetchProxySource.LOCAL_REMOTE);}/*** @author ljm* @date 10:28 2025/6/25* 监听容器刷新事件,触发代理定时刷新**/@EventListener(value = ContextRefreshedEvent.class)public void refreshProxies() {// 定时任务线程池抛出异常时任务会停止调度,采用UncaughtExceptionHandler捕获异常任务仍会停止调度,所有采用手动捕获异常SCHEDULED_EXECUTOR_SERVICE.scheduleWithFixedDelay(() -> {try {// 刷新每个代理的IP到缓存池FETCH_PROXY_MAP.values().forEach(AbstractProxiesProvider::refreshProxies);} catch (Exception e) {LOGGER.error("RefreshProxyError", e);}}, 3, 1, TimeUnit.SECONDS);}
}