本文介绍一个简单的多线程并发爬虫，这里说的简单是指爬取的数据规模不大，单机运行，并且不使用数据库，但保证多线程下的数据的一致性，并且能让爬得正起劲的爬虫停下来，而且能保存爬取状态以备下次继续。
爬虫实现的步骤基本如下：
- 分析网页结构，选取自己感兴趣的部分;
- 建立两个Buffer，一个用于保存已经访问的URL，一个用户保存带访问的URL;
- 从待访问的Buffer中取出一个URL来爬取，保存这个URL中感兴趣的信息；并将这个URL加入已经访问的Buffer中，然后将这个URL中的所有外链URLs中没有被访问过的URL加到待访问Buffer；
- 只要待访问的Buffer不为空，重复上一步。
这次是为了给博客园的用户进行一次pagerank排名，爬取了博客园各个用户的粉丝与关注者。博客园的用户用17万多个。爬取的页面是http://home.cnblogs.com/u/+userId，每个用户的url只需要用用户的id表示就可以了，用户id按平均10B来计算，保存所有用户也只需1.7Mb内存。因此我把两个Buffer都放在内存中。
这个项目的就四个java文件，结构如下：
下面对整个爬虫的实现过程进行详细的介绍。
一、登录
要获取用户的粉丝与关注，必须先登录，博客园的模拟登陆算是比较简单，找到登录时要上传的参数，然后Pos发送即登录成功，可以使用Chrome的工具，打开登录页面，调好账号和密码后，按F12弹出工具，按登录就能看到要传的参数了，再POST一个这些参数就好了。
我之前是使用这样的方式，后来用使用Jsoup解析的参数，代码实现如下：
1 /**
2 * 使用Joup解析登录参数，然后POST发送参数实现登录
3 *
4 * @throws UnsupportedEncodingException
5 * @throws IOException
6 */
7 private static void login() throws UnsupportedEncodingException,
8 IOException {
9 CookieHandler.setDefault(new CookieManager());
10 // 获取登录页面
11 String page = getPage(LOGIN_URL);
12 // 从登录去取出参数，并填充账号和密码
13 Document doc = Jsoup.parse(page);
14 // 取登录表格
15 Element loginform = doc.getElementById("frmLogin");
16 Elements inputElements = loginform.getElementsByTag("input");
17 List<String> paramList = new ArrayList<String>();
18 for (Element inputElement : inputElements) {
19 String key = inputElement.attr("name");
20 String value = inputElement.attr("value");
21 if (key.equals("tbUserName"))
22 value = Test.Name;
23 else if (key.equals("tbPassword"))
24 value = Test.passwd;
25 paramList.add(key + "=" + URLEncoder.encode(value, "UTF-8"));
26 }
27 // 封装请求参数
28 StringBuilder para = new StringBuilder();
29 for (String param : paramList) {
30 if (para.length() == 0) {
31 para.append(param);
32 } else {
33 para.append("&" + param);
34 }
35 }
36 // POST发送登录
37 String result = sendPost(LOGIN_URL, para.toString());
38 if (!result.contains("followees")) {
39 cookies = null;
40 System.out.println("登录失败");
41 } else
42 System.out.println("登录成功");
43 }
二、获取粉丝与关注
登录成功就可以爬取粉丝和关注了，关注在http://home.cnblogs.com/u/userid/followees/链接中，而粉丝在http://home.cnblogs.com/u/userid/followers/，两个网页结构基本相同，只需要把选择一下followees（被关注者）和（followers）关注者，用Jsoup解析avatar_list中avatar_name就好了，代码如下：
1 /**
2 * 获取一页中的关注or粉丝
3 *
4 * @param pageHtml
5 * @return
6 */
7
8 private List<String> getOnePageFriends(Document doc) {
9 List<String> firends = new ArrayList<String>();
10 Elements inputElements = doc.getElementsByClass("avatar_name");
11 for (Element inputElement : inputElements) {
12 Elements links = inputElement.getElementsByTag("a");
13 for (Element link : links) {
14 //从href中解析出用户id
15 String href = link.attr("href");
16 firends.add(href.substring(3, href.length() - 1));
17 }
18 }
19 return firends;
20 }
每一页显示50个粉丝or关注者，需要分页爬取，获取下一页跟获取用户粉丝差不多，找到元素就好。
三、爬取单个用户
爬取的一个用户的过程就是先分页爬取粉丝，再爬取关注者，然后把爬过的用户放入访问Buffer中，再把爬到的用户放到未访问队列中。
1 @Override
2 public void run() {
3 while (stop.get() == false) {
4 // 取出一个待访问
5 String userId = mUserBuffer.pickOne();
6 try {
7 // 爬取粉丝
8 List<String> fans = crawUser(userId, "/followers");
9 // 爬取关注者
10 List<String> heros = crawUser(userId, "/followees");
11 // 只需要保持粉丝关系即可
12 StringBuilder sb = new StringBuilder(userId).append("\t");
13 for (String friend : fans) {
14 sb.append(friend).append("\t");
15 }
16 sb.deleteCharAt(sb.length() - 1).append("\n");
17 saver.save(sb.toString());
18 // 被关注者应该放进队列里面，以供下次爬取他的粉丝
19 fans.addAll(heros);
20 mUserBuffer.addUnCrawedUsers(fans);
21 } catch (Exception e) {
22 saver.log(e.getMessage());
23 // 访问错误时，放入访问出错的队列中，以备以后重新访问。
24 mUserBuffer.addErrorUser(userId);
25 }
26 }
一页一页爬取单个用户如下：
1 /**
2 * 爬取用户，根据tag来决定是爬该用户关注的人，还是该用户的粉丝
3 *
4 * @param userId
5 * @return
6 * @throws IOException
7 */
8 private List<String> crawUser(String userId, String tag) throws IOException {
9 //构造URL
10 StringBuilder urlBuilder = new StringBuilder(USER_HOME);
11 urlBuilder.append("/u/").append(userId).append(tag);
12 //请求页面
13 String page = getPage(urlBuilder.toString());
14 Document doc = Jsoup.parse(page);
15 List<String> friends = new ArrayList<String>();
16 //爬取第一页
17 friends.addAll(getOnePageFriends(doc));
18 String nextUrl = null;
19 //不断地爬取下一页
20 while ((nextUrl = getNextUrl(doc)) != null) {
21 page = getPage(nextUrl);
22 doc = Jsoup.parse(page);
23 friends.addAll(getOnePageFriends(doc));
24 }
25 return friends;
26 }
整个爬虫结构就是这样了:
1 public class UserCrawler implements Runnable {
2 // 停止任务标志
3 private static AtomicBoolean stop;
4 // 当前爬虫的id
5 private int id;
6 // 用户缓存
7 private UserBuffer mUserBuffer;
8 // 日志与粉丝保存工具
9 private Saver saver;
10
11 static {
12 stop = new AtomicBoolean(false);
13 try {
14 // 登录一次即可
15 login();
16 // 保存数据线程先启动
17 Saver.getInstance().start();
18 } catch (IOException e) {
19 e.printStackTrace();
20 }
21 // new Thread(new CommandListener()).start();
22 }
23
24 public UserCrawler(UserBuffer userBuffer) {
25 mUserBuffer = userBuffer;
26 mUserBuffer.crawlerCountIncrease();
27 id = c++;
28 saver = Saver.getInstance();
29 }
30
31 @Override
32 public void run() {
33 if (id > 0) {
34 // 等第一个线程启动一段时候再开始新的线程
35 try {
36 TimeUnit.SECONDS.sleep(20 + id);
37 } catch (InterruptedException e) {
38 e.printStackTrace();
39 }
40 }
41 System.out.println("UserCrawler " + id + " start");
42 int retry = 3;// 重置尝试次数
43 while (stop.get() == false) {
44 // 取出一个待访问
45 String userId = mUserBuffer.pickOne();
46 if (userId == null) {// 队列元素已经为空
47 retry--;// 重试3次
48 if (retry <= 0)
49 break;
50 continue;
51 }
52 ...//爬取用户
53 }
54 System.out.println("UserCrawler " + id + " stop");
55 // 当前线程停止了
56 mUserBuffer.crawlerCountDecrease();
57 }
58
59 private List<String> crawUser(String userId, String tag) throws IOException {
60
61 }
62
63 /**
64 * 获取一页中的关注or粉丝
65 *
66 * @param pageHtml
67 * @return
68 */
69
70 private List<String> getOnePageFriends(Document doc) {
71 ...
72 }
73
74 /**
75 * 获取下一页的地址
76 *
77 * @param doc
78 * @return
79 */
80 private String getNextUrl(Document doc) {
81
82 }
83
84 private static String getPage(String pageUrl) throws IOException {
85
86 }
87
88 /***
89 * 终止所有爬虫任务
90 */
91 public static void stop() {
92 System.out.println("正在终止...");
93 stop.compareAndSet(false, true);
94 UserBuffer.getInstance().prepareForStop();
95 }
96
97 private static void login() throws UnsupportedEncodingException,
98 IOException {
99 ...
100 }
101
102
103 private static String sendPost(String url, String postParams)
104 throws IOException {
105 ...
106
107 }
四、Buffer并发控制
在用户Buffer设置一个已经访问的用户集合、一个访问出错的用户集合和一个待访问的队列。
1 private UserBuffer() {
2 crawedUsers = new HashSet<String>();// 已经访问的用户,包括访问成功和访问出错的用户
3 errorUsers = new HashSet<String>();// 访问出错的用户
4 unCrawedUsers = new LinkedList<String>();// 未访问的用户
5 }
UserBuffer全局唯一，因此采用单例模式，使用的集合和队列都不是线程安全的数据结构，我认为没有必要使用线程安全的ConcurrentSkipListSet与ConcurrentLinkedQueue，因为将一个用户插入unCrawedUsers队列时需要先判断是否已经存在于crawedUsers用户集合中了，这需要用锁来同步访问，如果不使用锁来控制，单个线程安全的set和queque不能保证多个变量之间的test-try有效性。而使用了锁后，再使用线程安全的数据结构只会增加加锁和解锁的次数，反而降低了性能。
1 /***
2 * 添加未访问的用户
3 *
4 * @param users
5 * @return
6 */
7 public synchronized void addUnCrawedUsers(List<String> users) {
8 // 添加未访问的用户
9 for (String user : users) {
10 if (!crawedUsers.contains(user))
11 unCrawedUsers.add(user);
12 }
13 }
14
15 /**
16 * 从队列中取一个元素,并把这个元素添加到已经访问的集合中，以免重复访问。
17 *
18 *
19 * @return
20 */
21 public synchronized String pickOne() {
22 String newId = unCrawedUsers.poll();
23 // 队列中可能包含重复的id，因为插入队列时只检查是否在访问集合里，
24 // 没有检查是否已经出现在队列里
25 while (crawedUsers.contains(newId)) {
26 newId = unCrawedUsers.poll();
27 }
28 //访问前先把添加到已经访问的集合中
29 crawedUsers.add(newId);
30 return newId;
31 }
32
33 /**
34 * 添加访问出错的用户
35 *
36 * @param userId
37 */
38 public synchronized void addErrorUser(String userId) {
39 errorUsers.add(userId);
40 }
五、安全地终止爬虫
大丈夫能伸能屈，能走能停，爬虫也当如此。爬博客园的用户时，我真是提心吊胆，担心管理员封我的账号，我尽量挑凌晨的时间爬数据，另外就是爬了一会后，我就停下来，过一段时间再爬。要让一个多线程的程序平稳的停下来还真不简单，最担心的就是死锁，发送停止命令后，线程不动却也没有终止，爬了好久的Buffer空间没有保存，那个恨啊。
我的思路:
1、爬虫启动时，向Buffer注册一下，buffer记录启动的爬虫数量；
2、对爬虫设置一个全局的标志，爬虫在每次爬取一个用户前检查终止标志是否被设置；
3、当发生停止命令时，爬虫检查到停止标志被设置，于是通知Buffer自己将要停止，通知完后就结束运行；
4、Buffer收到爬虫停止通知后，将爬虫计数器减1，当计数器为0时，保存工作空间，同时通知关闭日志；
5、为了避免有些爬虫在运行异常时推出而没有通知Buffer，在发出停止命令时，同时通知Buffer准备停止，Buffer设置一个计时器，2分钟后，强制保存爬虫状态和日志。
完整的代码还是放在Github上，有兴趣的同学可以看看。
感谢阅读，转载请注明出处：http://www.cnblogs.com/fengfenggirl/