webmagic 增量爬取

　webmagic  是一個很好並且很簡單的爬蟲框架，其教程網址：http://my.oschina.net/flashsword/blog/180623

 　　webmagic參考了scrapy的模塊划分，分為Spider(整個爬蟲的調度框架)、Downloader(頁面下載)、PageProcessor(鏈接提取和頁面分析)、Scheduler(URL管理)、Pipeline(離線分析和持久化)幾部分。只不過scrapy通過middleware實現擴展，而webmagic則通過定義這幾個接口，並將其不同的實現注入主框架類Spider來實現擴展。

關於Scheduler(URL管理) 最基本的功能是實現對已經爬取的URL進行標示。

目前scheduler有三種實現方式：

　　1）內存隊列

　　2）文件隊列

　　3)redis隊列

文件隊列保存URL，能實現中斷后，繼續爬取時，實現增量爬取。

　　如果我只有一個主頁的URL，比如：http://www.cndzys.com/yundong/。如果直接引用webmagic的FileCacheQueueScheduler的話，你會發現第二次啟動的時候，什么也爬不到。可以說第二次啟動基本不爬取數據了。因為FileCacheQueueScheduler 把http://www.cndzys.com/yundong/ 記錄了，然后不再進行新的爬取。雖然是第二次增量爬取，但還是需要保留某些URL重新爬取，以保證爬取結果是我們想要的。我們可以重寫FileCacheQueueScheduler里的比較方法。

 

package com.fortunedr.crawler.expertadvice;


import java.io.BufferedReader;
import java.io.Closeable;
import java.io.File;
import java.io.FileNotFoundException;
import java.io.FileReader;
import java.io.FileWriter;
import java.io.IOException;
import java.io.PrintWriter;
import java.util.LinkedHashSet;
import java.util.Set;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.commons.io.IOUtils;
import org.apache.commons.lang3.math.NumberUtils;

import us.codecraft.webmagic.Request;
import us.codecraft.webmagic.Task;
import us.codecraft.webmagic.scheduler.DuplicateRemovedScheduler;
import us.codecraft.webmagic.scheduler.MonitorableScheduler;
import us.codecraft.webmagic.scheduler.component.DuplicateRemover;


/**
 * Store urls and cursor in files so that a Spider can resume the status when shutdown.<br>
 *增加去重的校驗，對需要重復爬取的網址進行正則過濾
 * @author [e-mail] <br>
 * @since 0.2.0
 */
public class SpikeFileCacheQueueScheduler extends DuplicateRemovedScheduler implements MonitorableScheduler,Closeable {

    private String filePath = System.getProperty("java.io.tmpdir");

    private String fileUrlAllName = ".urls.txt";

    private Task task;

    private String fileCursor = ".cursor.txt";

    private PrintWriter fileUrlWriter;

    private PrintWriter fileCursorWriter;

    private AtomicInteger cursor = new AtomicInteger();

    private AtomicBoolean inited = new AtomicBoolean(false);

    private BlockingQueue<Request> queue;

    private Set<String> urls;
    
    private ScheduledExecutorService flushThreadPool;
    
    private String regx;
    
    
    public SpikeFileCacheQueueScheduler(String filePath) {
        if (!filePath.endsWith("/") && !filePath.endsWith("\\")) {
            filePath += "/";
        }
        this.filePath = filePath;
        initDuplicateRemover();
    }

    private void flush() {
        fileUrlWriter.flush();
        fileCursorWriter.flush();
    }

    private void init(Task task) {
        this.task = task;
        File file = new File(filePath);
        if (!file.exists()) {
            file.mkdirs();
        }
        readFile();
        initWriter();
        initFlushThread();
        inited.set(true);
        logger.info("init cache scheduler success");
    }

    private void initDuplicateRemover() {
        setDuplicateRemover(
                new DuplicateRemover() {
                    @Override
                    public boolean isDuplicate(Request request, Task task) {
                        if (!inited.get()) {
                            init(task);
                        }
                        boolean temp=false;
                        String url=request.getUrl();
                        temp=!urls.add(url);//原來驗證URL是否存在
                        //正則匹配
                        if(url.matches(regx)){//二次校驗，如果符合我們需要重新爬取的，返回false。可以重新爬取
                            temp=false;
                        }
                        return temp;
                    }

                    @Override
                    public void resetDuplicateCheck(Task task) {
                        urls.clear();
                    }

                    @Override
                    public int getTotalRequestsCount(Task task) {
                        return urls.size();
                    }
                });
    }

    private void initFlushThread() {
        flushThreadPool = Executors.newScheduledThreadPool(1);
        flushThreadPool.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                flush();
            }
        }, 10, 10, TimeUnit.SECONDS);
    }

    private void initWriter() {
        try {
            fileUrlWriter = new PrintWriter(new FileWriter(getFileName(fileUrlAllName), true));
            fileCursorWriter = new PrintWriter(new FileWriter(getFileName(fileCursor), false));
        } catch (IOException e) {
            throw new RuntimeException("init cache scheduler error", e);
        }
    }

    private void readFile() {
        try {
            queue = new LinkedBlockingQueue<Request>();
            urls = new LinkedHashSet<String>();
            readCursorFile();
            readUrlFile();
            // initDuplicateRemover();
        } catch (FileNotFoundException e) {
            //init
            logger.info("init cache file " + getFileName(fileUrlAllName));
        } catch (IOException e) {
            logger.error("init file error", e);
        }
    }

    private void readUrlFile() throws IOException {
        String line;
        BufferedReader fileUrlReader = null;
        try {
            fileUrlReader = new BufferedReader(new FileReader(getFileName(fileUrlAllName)));
            int lineReaded = 0;
            while ((line = fileUrlReader.readLine()) != null) {
                urls.add(line.trim());
                lineReaded++;
                if (lineReaded > cursor.get()) {
                    queue.add(new Request(line));
                }
            }
        } finally {
            if (fileUrlReader != null) {
                IOUtils.closeQuietly(fileUrlReader);
            }
        }
    }

    private void readCursorFile() throws IOException {
        BufferedReader fileCursorReader = null;
        try {
            fileCursorReader = new BufferedReader(new FileReader(getFileName(fileCursor)));
            String line;
            //read the last number
            while ((line = fileCursorReader.readLine()) != null) {
                cursor = new AtomicInteger(NumberUtils.toInt(line));
            }
        } finally {
            if (fileCursorReader != null) {
                IOUtils.closeQuietly(fileCursorReader);
            }
        }
    }
    
    public void close() throws IOException {
        flushThreadPool.shutdown();    
        fileUrlWriter.close();
        fileCursorWriter.close();
    }

    private String getFileName(String filename) {
        return filePath + task.getUUID() + filename;
    }

    @Override
    protected void pushWhenNoDuplicate(Request request, Task task) {
        queue.add(request);
        fileUrlWriter.println(request.getUrl());
    }

    @Override
    public synchronized Request poll(Task task) {
        if (!inited.get()) {
            init(task);
        }
        fileCursorWriter.println(cursor.incrementAndGet());
        return queue.poll();
    }

    @Override
    public int getLeftRequestsCount(Task task) {
        return queue.size();
    }

    @Override
    public int getTotalRequestsCount(Task task) {
        return getDuplicateRemover().getTotalRequestsCount(task);
    }

    public String getRegx() {
        return regx;
    }
    /**
     * 設置保留需要重復爬取url的正則表達式
     * @param regx
     */
    public void setRegx(String regx) {
        this.regx = regx;
    }
    
    
}

那么在爬蟲時就引用自己特定的FileCacheQueueScheduler就可以

spider.addRequest(requests);
        SpikeFileCacheQueueScheduler file=new SpikeFileCacheQueueScheduler(filePath);
        file.setRegx(regx);//http://www.cndzys.com/yundong/(index)?[0-9]*(.html)?
        spider.setScheduler(file );

　　這樣就實現了增量爬取。

 

優化的想法：一般某個網站的內容列表都是首頁是最新內容。上面的方式是可以實現增量爬取，但是還是需要爬取很多“無用的”列表頁面。

能不能實現，當爬取到上次"最新"URL之后就不再爬取。就是不用爬取其他多余的leib