高并发Java爬虫的瓶颈分析与动态线程优化方案

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;
import java.net.*;
import java.io.IOException;
import java.net.http.HttpClient;
import java.net.http.HttpRequest;
import java.net.http.HttpResponse;
import java.net.URI;

public class DynamicThreadPoolCrawler {

    // 代理配置信息
    private static final String proxyHost = "www.16yun.cn";
    private static final String proxyPort = "5445";
    private static final String proxyUser = "16QMSOML";
    private static final String proxyPass = "280651";

    // 自定义线程池，用于重写 beforeExecute 和 afterExecute 以进行监控
    private static class MonitorableThreadPoolExecutor extends ThreadPoolExecutor {
        // 用于记录任务执行时间的统计
        private final ThreadLocal<Long> startTime = new ThreadLocal<>();
        private final AtomicLong totalTime = new AtomicLong(0);
        private final AtomicLong numberOfTasks = new AtomicLong(0);

        public MonitorableThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue) {
            super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
        }

        @Override
        protected void beforeExecute(Thread t, Runnable r) {
            super.beforeExecute(t, r);
            startTime.set(System.currentTimeMillis());
        }

        @Override
        protected void afterExecute(Runnable r, Throwable t) {
            try {
                long endTime = System.currentTimeMillis();
                long taskTime = endTime - startTime.get();
                totalTime.addAndGet(taskTime);
                numberOfTasks.incrementAndGet();
                // 此处可以获取任务结果，根据HTTP状态码进行判断
                // 例如：if (r instanceof FutureTask) { ... get() ... if statusCode == 429 ... }
                System.out.println(String.format("Task executed in %d ms. Active: %d, Completed: %d, Queue: %d",
                        taskTime, getActiveCount(), getCompletedTaskCount(), getQueue().size()));
            } finally {
                super.afterExecute(r, t);
            }
        }

        public double getAverageTaskTime() {
            long count = numberOfTasks.get();
            return (count == 0) ? 0 : (double) totalTime.get() / count;
        }
    }

    // 创建带代理的HttpClient
    private static HttpClient createHttpClientWithProxy() {
        // 设置代理认证信息
        Authenticator authenticator = new Authenticator() {
            @Override
            protected PasswordAuthentication getPasswordAuthentication() {
                return new PasswordAuthentication(proxyUser, proxyPass.toCharArray());
            }
        };

        // 创建代理对象
        InetSocketAddress proxyAddress = new InetSocketAddress(proxyHost, Integer.parseInt(proxyPort));
        Proxy proxy = new Proxy(Proxy.Type.HTTP, proxyAddress);

        // 创建HttpClient并配置代理
        return HttpClient.newBuilder()
                .proxy(ProxySelector.of(proxyAddress))
                .authenticator(authenticator)
                .version(HttpClient.Version.HTTP_1_1)
                .build();
    }

    public static void main(String[] args) throws InterruptedException {
        // 1. 初始化动态线程池
        // 使用 SynchronousQueue 而不是无界队列
        MonitorableThreadPoolExecutor executor = new MonitorableThreadPoolExecutor(
                2,  // corePoolSize
                10, // maximumPoolSize
                60L, TimeUnit.SECONDS,
                new SynchronousQueue<>()
        );

        // 2. 创建带代理的HttpClient（单例，供所有任务使用）
        HttpClient httpClient = createHttpClientWithProxy();

        // 3. 创建实际爬虫任务（使用代理）
        Runnable crawlingTask = () -> {
            try {
                // 使用代理发送HTTP请求
                HttpRequest request = HttpRequest.newBuilder()
                        .uri(URI.create("https://httpbin.org/ip")) // 示例URL，返回请求的IP信息
                        .GET()
                        .build();

                HttpResponse<String> response = httpClient.send(request, HttpResponse.BodyHandlers.ofString());

                // 输出响应信息，验证代理是否生效
                System.out.println("Response status: " + response.statusCode());
                System.out.println("Response body: " + response.body());

                // 模拟额外的处理时间
                int processingTime = ThreadLocalRandom.current().nextInt(100, 500);
                Thread.sleep(processingTime);

                // 根据状态码处理反爬逻辑
                if(response.statusCode() == 429) {
                    System.err.println("Encountered 429 Too Many Requests! Should scale down.");
                }
            } catch (IOException | InterruptedException e) {
                System.err.println("Request failed: " + e.getMessage());
                Thread.currentThread().interrupt();
            }
        };

        // 4. 提交大量任务
        for (int i = 0; i < 100; i++) {
            executor.execute(crawlingTask);
        }

        // 5. 创建监控线程，定期调整线程池大小
        ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();
        scheduler.scheduleAtFixedRate(() -> {
            int activeCount = executor.getActiveCount();
            int queueSize = executor.getQueue().size(); // 对于 SynchronousQueue，这通常为0
            int currentCorePoolSize = executor.getCorePoolSize();
            int currentMaxPoolSize = executor.getMaximumPoolSize();

            System.out.println("\n=== Monitoring Report ===");
            System.out.println("Active Threads: " + activeCount);
            System.out.println("Queue Size: " + queueSize);
            System.out.println("Current Core Pool Size: " + currentCorePoolSize);
            System.out.println("Current Max Pool Size: " + currentMaxPoolSize);
            System.out.println("Avg Task Time: " + executor.getAverageTaskTime() + " ms");

            // 动态调整策略示例：
            if (activeCount >= currentMaxPoolSize && queueSize == 0) {
                // 线程池已满负荷运行，考虑扩容（在安全限制内）
                int newMaxPoolSize = Math.min(currentMaxPoolSize + 2, 20); // 上限设为20
                executor.setMaximumPoolSize(newMaxPoolSize);
                executor.setCorePoolSize(newMaxPoolSize); // 也增加核心线程数
                System.out.println(">>> Scaling UP to: " + newMaxPoolSize);
            } else if (activeCount < currentCorePoolSize / 2) {
                // 负载很低，考虑缩容（但不能低于初始最小值）
                int newCorePoolSize = Math.max(2, currentCorePoolSize - 1);
                executor.setCorePoolSize(newCorePoolSize);
                executor.setMaximumPoolSize(Math.max(newCorePoolSize, currentMaxPoolSize - 1));
                System.out.println(">>> Scaling DOWN to core: " + newCorePoolSize);
            }
        }, 5, 5, TimeUnit.SECONDS); // 每5秒监控一次

        // 主线程等待一段时间，然后关闭
        Thread.sleep(60000);
        scheduler.shutdown();
        executor.shutdown();
        try {
            if (!executor.awaitTermination(60, TimeUnit.SECONDS)) {
                executor.shutdownNow();
            }
        } catch (InterruptedException e) {
            executor.shutdownNow();
        }
    }
}