异步与并发：利用Java多线程技术提升淘宝商品爬取效率

import org.jsoup.Jsoup;
import org.jsoup.nodes.Document;
import java.net.InetSocketAddress;
import java.net.Proxy;
import java.util.concurrent.Callable;

/**
 * 单个商品详情抓取任务
 */
public class ProductDetailFetcher implements Callable<ProductDetail> {

    private final String itemId;
    private final String proxyHost = "www.16yun.cn";
    private final int proxyPort = 5445;
    private final String proxyUser = "16QMSOML";
    private final String proxyPass = "280651";

    public ProductDetailFetcher(String itemId) {
        this.itemId = itemId;
    }

    @Override
    public ProductDetail call() throws Exception {
        // 1. 模拟或真实地构建目标URL
        // 注意：淘宝有强大的反爬虫机制，直接爬取非常困难。
        // 此处仅为演示并发结构，URL和解析逻辑已做简化处理。
        String url = "https://item.taobao.com/item.htm?id=" + itemId;
        
        // 2. 创建代理对象
        Proxy proxy = new Proxy(Proxy.Type.HTTP, new InetSocketAddress(proxyHost, proxyPort));
        
        // 3. 发送HTTP请求，获取HTML文档（使用代理）
        // 在实际项目中，这里需要配置User-Agent、代理IP、Cookie等以应对反爬。
        Document doc = Jsoup.connect(url)
                .proxy(proxy) // 设置代理
                .userAgent("Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36")
                .header("Proxy-Authorization", getProxyAuthHeader()) // 设置代理认证
                .timeout(30000) // 增加超时时间，因为代理可能会增加延迟
                .ignoreContentType(true)
                .ignoreHttpErrors(true)
                .get();

        // 4. 解析HTML，提取商品信息
        String title = doc.select("h1[class*='Title']").first().text();
        // 价格可能由动态JS渲染，此处解析静态页面可能为空，实战中需用Selenium等工具。
        String price = doc.select("span[class*='Price']").first().text(); 
        String shopName = doc.select("a[class*='ShopName']").first().text();

        // 5. 封装并返回商品详情对象
        return new ProductDetail(itemId, title, price, shopName);
    }

    /**
     * 生成代理认证头信息
     */
    private String getProxyAuthHeader() {
        String credentials = proxyUser + ":" + proxyPass;
        String base64Credentials = java.util.Base64.getEncoder().encodeToString(credentials.getBytes());
        return "Basic " + base64Credentials;
    }
}

/**
 * 商品详情数据模型
 */
public class ProductDetail {
    private String itemId;
    private String title;
    private String price;
    private String shopName;

    // 构造器、Getter和Setter方法、toString方法
    public ProductDetail(String itemId, String title, String price, String shopName) {
        this.itemId = itemId;
        this.title = title;
        this.price = price;
        this.shopName = shopName;
    }
    // ... 省略 Getter 和 Setter ...
    
    @Override
    public String toString() {
        return "ProductDetail{" +
                "itemId='" + itemId + '\'' +
                ", title='" + title + '\'' +
                ", price='" + price + '\'' +
                ", shopName='" + shopName + '\'' +
                '}';
    }
}

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;

/**
 * 并发爬虫调度引擎
 */
public class ConcurrentCrawlerEngine {

    public static void main(String[] args) {
        // 模拟一批要抓取的商品ID
        List<String> itemIds = List.of("1234567890", "2345678901", "3456789012", "..."); // 可以有很多

        // 1. 创建线程池
        // corePoolSize: 核心线程数，这里设置为5，可根据机器性能和网络情况调整。
        // maximumPoolSize: 最大线程数，设置为10。
        // keepAliveTime: 非核心线程空闲存活时间。
        // workQueue: 任务队列，使用LinkedBlockingQueue。
        ExecutorService executor = new ThreadPoolExecutor(
                5, 10, 60L, TimeUnit.SECONDS, new LinkedBlockingQueue<>()
        );

        // 2. 创建Future列表，用于保存每个任务的异步结果
        List<Future<ProductDetail>> futureList = new ArrayList<>();

        // 3. 提交所有任务
        System.out.println("开始提交抓取任务...");
        for (String itemId : itemIds) {
            ProductDetailFetcher fetcher = new ProductDetailFetcher(itemId);
            Future<ProductDetail> future = executor.submit(fetcher);
            futureList.add(future);
        }

        // 4. 关闭线程池，不再接受新任务，但会执行完已提交的任务
        executor.shutdown();

        // 5. 遍历Future列表，获取任务结果
        List<ProductDetail> results = new ArrayList<>();
        for (Future<ProductDetail> future : futureList) {
            try {
                // future.get() 会阻塞，直到该任务完成并返回结果
                ProductDetail detail = future.get();
                results.add(detail);
                System.out.println("成功抓取: " + detail);
            } catch (InterruptedException | ExecutionException e) {
                // 处理抓取过程中可能出现的异常
                System.err.println("抓取任务失败: " + e.getMessage());
            }
        }

        // 6. 所有任务完成，处理最终结果 (例如，存入数据库或文件)
        System.out.println("所有任务执行完毕。总共抓取到 " + results.size() + " 个商品。");
        // saveToDatabase(results);
    }
}

// 使用CompletableFuture的示例片段
List<CompletableFuture<Void>> futures = itemIds.stream()
    .map(itemId -> CompletableFuture.supplyAsync(() -> new ProductDetailFetcher(itemId).call(), executor)
                                   .thenApplyAsync(this::saveToDatabase) // 异步存储
                                   .exceptionally(e -> { System.err.println("Error: " + e); return null; })
    )
    .collect(Collectors.toList());

// 等待所有任务完成
CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();