用生产者消费者模式实现爬虫批量提交网页给搜索引擎

1：爬虫：crawler4j ；

参考资料：http://blog.csdn.net/longzuyuan/article/details/8894912
http://blog.csdn.net/sadfishsc/article/details/20614105
参考书籍：自己动手写网络爬虫 --罗刚

2：搜索服务器：solr4.10 ；

3：多线程处理

参考书籍：java并发编程实战

参考相关jdk api：http://www.yq1012.com/api/，堵塞队列BlockingQueue<E> 类

业务：爬取国内部分招聘网站的职位信息。。当爬虫线程抓取到的页面数据到一定量时或者一定时间内。提交给搜索引擎solr（提高

索引性能）。循环爬行操作。。从而实现更高的资源利用率。。思路：N个线程爬虫作为Producer，提交搜索引擎作为Consumer。

部分爬虫相关代码：

package crawler;

import edu.uci.ics.crawler4j.crawler.CrawlConfig;
import edu.uci.ics.crawler4j.crawler.CrawlController;
import edu.uci.ics.crawler4j.fetcher.PageFetcher;
import edu.uci.ics.crawler4j.robotstxt.RobotstxtConfig;
import edu.uci.ics.crawler4j.robotstxt.RobotstxtServer;

public class Crawler4jTest {

public static void main(String[] args) {
try {
String crawlStorageFolder = "./tmp";
int numberOfCrawlers = 5;
CrawlConfig config = new CrawlConfig();
// 文明请求web：确保我们不发送超过1每秒请求数（1000毫秒之间的请求）。
config.setPolitenessDelay(1000);
// 深度，即从入口URL开始算，URL是第几层。如入口A是1，从A中找到了B，B中又有C，则B是2，C是3
config.setMaxDepthOfCrawling(5);

//设置最大的抓取页面数。默认值为1，页面的数量不限
config.setMaxPagesToFetch(50);

// 如果需要代理服务器的话
//config.setProxyHost("proxyserver.example.com");  //设置代理域名
//config.setProxyPort(8080);//端口

// 如果代理服务器需要认证
//config.setProxyUsername(username); config.getProxyPassword(password);  //设置代理
/*
* 此配置参数可以用来设置你的爬行是可恢复的（这意味着可以从先前中断/恢复爬行）
* 注意：如果启用恢复特征，想开始一个新的抓取，你需要删除的内容手动rootfolder。
*/
config.setResumableCrawling(false);

config.setCrawlStorageFolder(crawlStorageFolder);
PageFetcher pageFetcher = new PageFetcher(config);
RobotstxtConfig robotstxtConfig = new RobotstxtConfig();
RobotstxtServer robotstxtServer = new RobotstxtServer(robotstxtConfig, pageFetcher);
CrawlController controller = new CrawlController(config, pageFetcher, robotstxtServer);
controller.getCrawlersLocalData();
controller.addSeed("http://www.lagou.com");

CommitConsumer consumer=new CommitConsumer();
new Thread(consumer).start();
controller.start(WomiCrawler.class, numberOfCrawlers);

} catch (Exception e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}

package crawler;

import java.util.regex.Pattern;

import org.apache.solr.common.SolrInputDocument;

import edu.uci.ics.crawler4j.crawler.Page;
import edu.uci.ics.crawler4j.crawler.WebCrawler;
import edu.uci.ics.crawler4j.parser.HtmlParseData;
import edu.uci.ics.crawler4j.url.WebURL;

public class WomiCrawler extends WebCrawler{

private final static Pattern FILTERS = Pattern.compile(".*(\\.(css|js|bmp|gif|jpe?g" + "|png|tiff?|mid|mp2|mp3|mp4"
+ "|wav|avi|mov|mpeg|ram|m4v|pdf" + "|rm|smil|wmv|swf|wma|zip|rar|gz))$");
//页面前缀
private final static String URL_PREFIX = "http://www.lagou.com/jobs/";

/**
* shouldVisit是判断当前的URL是否已经应该被爬取（访问）
*/
@Override
public boolean shouldVisit(WebURL url) {
String href = url.getURL().toLowerCase();
return !FILTERS.matcher(href).matches() && href.startsWith(URL_PREFIX);
}

/**
* visit则是爬取该URL所指向的页面的数据，其传入的参数即是对该web页面全部数据的封装对象Page。
*/
@Override
public  void  visit(Page page) {
try {
SolrInputDocument doc=new SolrInputDocument();
int docid = page.getWebURL().getDocid();
String url = page.getWebURL().getURL();
String parentUrl = page.getWebURL().getParentUrl();
String anchor = page.getWebURL().getAnchor();
doc.addField("id", docid+"");
doc.addField("url", url+"");
doc.addField("host", url+"");
doc.addField("title", anchor+"");
doc.addField("author", anchor+"");
System.out.println("Docid: " + docid);
System.out.println("URL: " + url);
System.out.println("Parent page: " + parentUrl);
System.out.println("anchor: " + anchor);
if (page.getParseData() instanceof HtmlParseData) {
HtmlParseData htmlParseData = (HtmlParseData) page.getParseData();
String text = htmlParseData.getText();
doc.addField("content", text);
}
Lock lock = Lock.getInstance();
lock.lstDocument.add(doc);
lock.num++;
System.out.println("爬虫次数: num ==" + lock.num);

} catch (Exception e) {
// TODO Auto-generated catch block
e.printStackTrace();
}

}
}

部分对象锁代码

package crawler;

import java.util.concurrent.LinkedBlockingQueue;
import org.apache.solr.common.SolrInputDocument;

public class Lock {

private static Lock lock ;
public static Lock getInstance(){
if(lock==null){
synchronized (Lock.class) {
if(lock==null){
lock=new Lock();
}
}
}
return lock;
}
private Lock(){}
//爬取page数量
public int num = 0;
//提交次数
public int commitNum = 0;
//索引数据集-消费者模式
public LinkedBlockingQueue<SolrInputDocument> lstDocument = new LinkedBlockingQueue<SolrInputDocument>();
}

部分消费者代码：

package crawler;

import java.util.LinkedList;
import java.util.List;

import org.apache.solr.client.solrj.SolrServer;
import org.apache.solr.common.SolrInputDocument;

import search.solr.IndexerUtil;

public class CommitConsumer implements Runnable {

private SolrServer server = IndexerUtil.getHttpSolrServer("crawl");
private List<SolrInputDocument> list=new LinkedList<SolrInputDocument>();

private int commit=0;
public void run() {
try {
SolrInputDocument doc=null;
while((doc=Lock.getInstance().lstDocument.take())!=null){
list.add(doc);
if(list.size()==5){
commit++;
server.add(list);
server.commit();
list.clear();
System.out.println("提交次数："+commit);
}
}
} catch (Exception e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}

}