摘要：包主要實現(xiàn)類，這是一個抽象類，實現(xiàn)了通用的模板方法，并在方法內(nèi)部判斷錯誤重試去重處理等。重置重復檢查就是清空，獲取請求總數(shù)也就是獲取的。至于請求總數(shù)統(tǒng)計，就是返回中維護的的大小。

Scheduler是Webmagic中的url調(diào)度器，負責從Spider處理收集(push)需要抓取的url(Page的targetRequests)、并poll出將要被處理的url給Spider，同時還負責對url判斷是否進行錯誤重試、及去重處理、以及總頁面數(shù)、剩余頁面數(shù)統(tǒng)計等。

主要接口：

Scheduler，定義了基本的push和poll方法?；窘涌?。

MonitorableScheduler，繼承自Scheduler的接口，定義了獲取剩余url請求數(shù)和總請求數(shù)的方法。便于監(jiān)控。

core包主要實現(xiàn)類：

DuplicateRemovedScheduler，這是一個抽象類，實現(xiàn)了通用的push模板方法，并在push方法內(nèi)部判斷錯誤重試、去重處理等。去重策略采用的是HashSetDuplicateRemover類，這個會在稍后說明。

PriorityScheduler，內(nèi)置兩個優(yōu)先級隊列(+,-)和一個非優(yōu)先級阻塞隊列的調(diào)度器。

QueueScheduler，內(nèi)置一個阻塞隊列的調(diào)度器。這是默認采用的。

URL去重策略：

DuplicateRemover：去重接口，含有判斷是否重復，重置重復檢查，獲取請求總數(shù)的方法。

HashSetDuplicateRemover：DuplicateRemover的實現(xiàn)類，內(nèi)部維護了一個并發(fā)安全的HashSet。

先說下去重策略的具體實現(xiàn)。核心代碼如下：

public class HashSetDuplicateRemover implements DuplicateRemover {
    private Set urls = Collections.newSetFromMap(new ConcurrentHashMap());
    @Override
    public boolean isDuplicate(Request request, Task task) {
        return !urls.add(getUrl(request));
    }
    。。。
    @Override
    public void resetDuplicateCheck(Task task) {
        urls.clear();
    }
    @Override
    public int getTotalRequestsCount(Task task) {
        return urls.size();
    }
}

去重策略類很簡單，就是維護一個并發(fā)安全的HashSet。然后通過add方法是否成功來判斷是否是重復的url。重置重復檢查就是清空set，獲取請求總數(shù)也就是獲取set的size。簡單明了。但是你以為去重就這么點，那么你錯了。繼續(xù)看。

public abstract class DuplicateRemovedScheduler implements Scheduler {
    private DuplicateRemover duplicatedRemover = new HashSetDuplicateRemover();
    @Override
    public void push(Request request, Task task) {
        if (shouldReserved(request) || noNeedToRemoveDuplicate(request) || !duplicatedRemover.isDuplicate(request, task)) {
            pushWhenNoDuplicate(request, task);
        }
    }
    protected boolean shouldReserved(Request request) {
        return request.getExtra(Request.CYCLE_TRIED_TIMES) != null;
    }

    protected boolean noNeedToRemoveDuplicate(Request request) {
        return HttpConstant.Method.POST.equalsIgnoreCase(request.getMethod());
    }
    protected void pushWhenNoDuplicate(Request request, Task task) {

    }
}

DuplicateRemovedScheduler是一個抽象類，提供了push的通用模板，并為子類提供了pushWhenNoDuplicate用于實現(xiàn)自己的策略。push方法用于同一處理去重和重試機制。
首先判斷是否需要進行錯誤重試，如果需要，那么就直接push到隊列中，否則判斷請求是否為POST方法，如果是直接加入隊列，(這里需要注意的是，POST請求的url不會被加入HashSetDuplicateRemover維護的urls集合，故而也不會被加入到最終的getTotalRequestsCount的統(tǒng)計中，所以最終我們獲取的統(tǒng)計信息只是針對GET請求的。),否則進行去重判斷。
根據(jù)不同調(diào)度器的實現(xiàn)，pushWhenNoDuplicate的實現(xiàn)方式不一樣。
在PriorityScheduler中內(nèi)置兩個優(yōu)先級隊列(+,-)和一個非優(yōu)先級阻塞隊列的調(diào)度器，其pushWhenNoDuplicate代碼如下：

public void pushWhenNoDuplicate(Request request, Task task) {
    if (request.getPriority() == 0) {
        noPriorityQueue.add(request);
    } else if (request.getPriority() > 0) {
        priorityQueuePlus.put(request);
    } else {
        priorityQueueMinus.put(request);
    }
}

根據(jù)Request是否設置priority屬性，以及是否為正、負來決定加入到哪個隊列中。因為這影響了后續(xù)poll的先后順序。

在QueueScheduler中內(nèi)置一個阻塞隊列的調(diào)度器。其pushWhenNoDuplicate代碼如下：

    public void pushWhenNoDuplicate(Request request, Task task) {
        queue.add(request);
    }

就是簡單地將其加入隊列中。

以上就是關于URL去重及push的機制，接下來說明poll思路：

在PriorityScheduler中，poll順序為plus隊列>noPriority隊列>minus隊列。

public synchronized Request poll(Task task) {
        Request poll = priorityQueuePlus.poll();
        if (poll != null) {
            return poll;
        }
        poll = noPriorityQueue.poll();
        if (poll != null) {
            return poll;
        }
        return priorityQueueMinus.poll();
    }

在QueueScheduler中，簡單粗暴。

public Request poll(Task task) {
        return queue.poll();
    }

至于url請求總數(shù)統(tǒng)計，就是返回HashSetDuplicateRemover中維護的urls set的大小。這里再次羅嗦一次：最終我們獲取的統(tǒng)計信息只是針對GET請求的。

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

當然extensions擴展模塊中還有些Scheduler實現(xiàn)，比如RedisScheduler用作集群支持，F(xiàn)ileCacheQueueScheduler用來斷點續(xù)爬支持等。由于本系列文章是先分析核心包，后續(xù)分析擴展包，所以關于這部分，后續(xù)補充。

RedisScheduler
思路是采用set來存儲已經(jīng)抓取過的url，list來存儲待抓url隊列，hash來存儲序列化數(shù)據(jù)(哈希中的鍵為url的SHA值，值為Request的json序列化字符串)。所有數(shù)據(jù)類型的鍵都是基于Spider的UUID來生成的，也就是說每個Spider實例所擁有的都是不同的。

 @Override
    public boolean isDuplicate(Request request, Task task) {
        Jedis jedis = pool.getResource();
        try {
            return jedis.sadd(getSetKey(task), request.getUrl()) > 0;
        } finally {
            pool.returnResource(jedis);
        }

    }

    @Override
    protected void pushWhenNoDuplicate(Request request, Task task) {
        Jedis jedis = pool.getResource();
        try {
            jedis.rpush(getQueueKey(task), request.getUrl());
            if (request.getExtras() != null) {
                String field = DigestUtils.shaHex(request.getUrl());
                String value = JSON.toJSONString(request);
                jedis.hset((ITEM_PREFIX + task.getUUID()), field, value);
            }
        } finally {
            pool.returnResource(jedis);
        }
    }

    @Override
    public synchronized Request poll(Task task) {
        Jedis jedis = pool.getResource();
        try {
            String url = jedis.lpop(getQueueKey(task));
            if (url == null) {
                return null;
            }
            String key = ITEM_PREFIX + task.getUUID();
            String field = DigestUtils.shaHex(url);
            byte[] bytes = jedis.hget(key.getBytes(), field.getBytes());
            if (bytes != null) {
                Request o = JSON.parseObject(new String(bytes), Request.class);
                return o;
            }
                Request request = new Request(url);
            return request;
        } finally {
            pool.returnResource(jedis);
        }
    }

 @Override
    public int getLeftRequestsCount(Task task) {
        Jedis jedis = pool.getResource();
        try {
            Long size = jedis.llen(getQueueKey(task));
            return size.intValue();
        } finally {
            pool.returnResource(jedis);
        }
    }

    @Override
    public int getTotalRequestsCount(Task task) {
        Jedis jedis = pool.getResource();
        try {
            Long size = jedis.scard(getSetKey(task));
            return size.intValue();
        } finally {
            pool.returnResource(jedis);
        }
    }

這些代碼都很好理解，只要有點redis基礎的都沒問題，這里就不再贅述了。

至于RedisPriorityScheduler就是采用有序的zset來存儲plus、min隊列，list來存儲noprioprity隊列。

FileCacheQueueScheduler
思路是維護兩個文件.cursor.txt,.urls.txt 前者由于存儲一個數(shù)字，這個數(shù)字代表了讀取.urls.txt的行數(shù)。后者用來存儲所有的urls。初始化時從兩個文件讀取內(nèi)存中，并初始化urls集和queue隊列、同時初始化flush線程定時flush內(nèi)容到文件中。當poll和pushWhenNoDuplicate時和原來邏輯差不多，只不過加了寫文件的步驟。
需要注意的是：FileCacheQueueScheduler實現(xiàn)了自己的去重規(guī)則，而不是直接使用DuplicateRemovedScheduler父類的去重規(guī)則。不過原理都一樣，都是通過Set來去重。

以上就是關于調(diào)度器的部分，下篇主題待定。