Erlang 分布式爬虫

下面这个是主模块,后面2段代码分别是数据库模块,和一个附属功能模块.

%%%------------------------------------------------------------------- 
%%% File    : Distributed-Crawler
%%% Designer  : free.Wang  
%%% Description :  
%%% Archieved :  Jan 28, 2009  
%%%------------------------------------------------------------------- 
-module(crawler_server). 
-behaviour(gen_server). 
%% API 
-export([start/1,login/0,crawl/0,check/0]). 

%% gen_server callbacks 
-export([init/1, handle_call/3, handle_cast/2, handle_info/2, 
    terminate/2, code_change/3]).


-define(MasterNode,'aaa@219.141.12.84'). 
-define(Downloadpath,'/Users/winfree/Project/distriawler/urlsea').
%-define(Mastername,).
%if this node is master.node , comment delow row.
%-define(Selfname,changjiangyihao).

%=====================================================================
%   gen_server : {global,node()}
%   send       : mnesia:lookup(pid_table,node)
%=====================================================================

%%==================================================================== 
%% API 
%%==================================================================== 
%%-------------------------------------------------------------------- 
%% Function: start_link() -> {ok,Pid} | ignore | {error,Error} 
%% Description: Starts the server 
%%-------------------------------------------------------------------- 

globalname(Mode) ->
if
 Mode  =:=  'slave' ->
  [Name,_] = string:tokens(atom_to_list(node()),"@");
 Mode  =:=  'master' ->
  [Name,_] = string:tokens(atom_to_list(?MasterNode),"@")
end,
list_to_atom(Name).




start(Mode) ->
if
 Mode  =:=  'master' ->
  gen_server:start_link({global, globalname(master)}, ?MODULE, [], []);
 Mode  =:=  'slave'  ->
  net_adm:ping(?MasterNode),
  gen_server:start_link({global, globalname(slave)}, ?MODULE, [], []),
  login()
end.




%%==================================================================== 
%% gen_server callbacks 
%%==================================================================== 

%%-------------------------------------------------------------------- 
%% Function: init(Args) -> {ok, State} | 
%%                         {ok, State, Timeout} | 
%%                         ignore               | 
%%                         {stop, Reason} 
%% Description: Initiates the server 
%%-------------------------------------------------------------------- 
init([]) -> 
   database:start(),
   {ok,["success"]}.

%%-------------------------------------------------------------------- 
%% Function: %% handle_call(Request, From, State) -> {reply, Reply, State} | 
%%                                      {reply, Reply, State, Timeout} | 
%%                                      {noreply, State} | 
%%                                      {noreply, State, Timeout} | 
%%                                      {stop, Reason, Reply, State} | 
%%                                      {stop, Reason, State} 
%% Description: Handling call messages 
%%-------------------------------------------------------------------- 

%handle_call(_Request, _From, State) -> 
%   Reply = term(), 
 %  {reply,ok,State}; 

login() ->
   gen_server:call({global,globalname(master)},{login,{globalname(slave),node()}}).


handle_call({login,{Globalname,Node}},_From,State) ->
io:format("starting handling form ~p~n",[_From]),
   [Id,Hostname] = string:tokens(atom_to_list(Node),"@"),
   database:insert_address(Id,Hostname,Globalname),
% database exe. in MasterNode.
   Reply = database:select_all(address),
   {reply,Reply,State};

% this will be executed by Slave Node.
handle_call({crawl,Urls},_From,State) ->
   lists:foreach(fun(Url) ->
                   io:format("~p~n",[Url]),
       spawn(fun() -> 
     Command = " curl  " ++ Url ++ " > " ++ lists:nth(2,string:tokens(Url,"//")),
%                        os:cmd("cd"++?Downloadpath),
                       os:cmd(Command),
     io:format(" downloading :~p~n",[Command])
                 end)
   end,Urls),
   {reply,ok,State}.

   
%%-------------------------------------------------------------------- 
%% Function: handle_cast(Msg, State) -> {noreply, State} | 
%%                                      {noreply, State, Timeout} | 
%%                                      {stop, Reason, State} 
%% Description: Handling cast messages 
%%-------------------------------------------------------------------- 
handle_cast(_Msg, State) -> 
   {noreply, State}. 

%%-------------------------------------------------------------------- 
%% Function: handle_info(Info, State) -> {noreply, State} | 
%%                                       {noreply, State, Timeout} | 
%%                                       {stop, Reason, State} 
%% Description: Handling all non call/cast messages 
%%-------------------------------------------------------------------- 
handle_info(_Info, State) -> 
   {noreply, State}. 

%%-------------------------------------------------------------------- 
%% Function: terminate(Reason, State) -> void() 
%% Description: This function is called by a gen_server when it is about to 
%% terminate. It should be the opposite of Module:init/1 and do any necessary 
%% cleaning up. When it returns, the gen_server terminates with Reason. 
%% The return value is ignored. 
%%-------------------------------------------------------------------- 
terminate(_Reason, _State) -> 
   ok. 

%%-------------------------------------------------------------------- 
%% Func: code_change(OldVsn, State, Extra) -> {ok, NewState} 
%% Description: Convert process state when code is changed 
%%-------------------------------------------------------------------- 
code_change(_OldVsn, State, _Extra) -> 
   {ok, State}. 

%%-------------------------------------------------------------------- 
%%% Internal functions 
%%--------------------------------------------------------------------


allurls() ->
   ["http://www.g.cn","http://www.sina.com.cn","http://www.xiaonei.com",
   "http://www.erlang.org","http://www.microsoft.com","http://www.yahoo.com",
   "http://www.google.com","http://www.dell.com","http://www.doban.com"].

% communication with others.





check() ->
Nodes    = database:select_all(address),
   LoopNum  = length(Nodes),
   UrlGroups = tools:seperate_list(allurls(),LoopNum,[]),
io:format("~p~n------~p~n",[Nodes,UrlGroups]).

% this will deploy the CrawlerDuty by MasterNode.
crawl() ->
   Nodes    = database:select_all(address),
   LoopNum  = length(Nodes),
AIG   = length(allurls()) div LoopNum,
   UrlGroups = tools:seperate_list(allurls(),AIG,[]), 
   Final = lists:zip(Nodes,UrlGroups),
   lists:foreach(fun(One) ->
                     {Node,UrlGroup} = One,
                     {_ ,_ ,_ , Globalname} = Node,
      io:format("executed by ~p~n",[Globalname]),
                     spawn(fun() ->
                               gen_server:call({global,Globalname},{crawl,UrlGroup})
                        end)
              end,Final).

下面是数据库模块:

 
-module(database).
-import(lists, [foreach/2]).
-compile(export_all).
-include_lib("stdlib/include/qlc.hrl").

%% API
-export([start/0,select_all/1,delete/2,insert_address/3,
        insert_urltable/2]).



% address: {Globalname,id@hostname} to send Message.
% urltable:{"http://www.freeis.cn/new/year/","freeis.cn"}
-record(address, {id,hostname,globalname}).
-record(urltable, {url,domain}).



ready() ->
    mnesia:create_schema([node()]),
    mnesia:start(),
    mnesia:create_table(address,   [{attributes, record_info(fields, address )}]),
    mnesia:create_table(urltable,  [{attributes, record_info(fields, urltable)}]),
    mnesia:stop().


start() ->
    ready(),
    mnesia:start(),
    mnesia:wait_for_tables([address,urltable], 20000).

select_all(Table) ->
    do(qlc:q([X || X <- mnesia:table(Table)])).

do(Q) ->
    F = fun() -> qlc:e(Q) end,
    {atomic, Val} = mnesia:transaction(F),
    Val.

example_tables() ->
    [%% The address table
     {address, "jack",   "10.1.225.117",   "<0.33>"},
     {address, "allen",  "abc.example.com",  "<7.313>"},
     {address, "Foy",   "chat.gogogo.com",  "<1.214>"},
     %% The urltable table
     {urltable, "http://gogogo.sina.com",   "sina.com"},
     {urltable, "http://A.sohu.com",   "sohu.com"},
     {urltable, "http://B.avial.com",   "avial.com"},
     {urltable, "http://C.wgi.com",   "wgi.com"}
    ].

insert_address(Id, Hostname, Globalname) ->
    Row = #address{id=Id, hostname=Hostname, globalname=Globalname},
    F = fun() ->
  mnesia:write(Row)
 end,
    mnesia:transaction(F).

insert_urltable(Url,Domain) ->
    Row = #urltable{url=Url,domain=Domain},
    F = fun() ->
        mnesia:write(Row)
    end,
    mnesia:transaction(F).

delete_address(Item) ->
    Oid = {adress, Item},
    F = fun() ->
  mnesia:delete(Oid)
 end,
    mnesia:transaction(F).

delete_urltable(Item) ->
    Oid = {urltable, Item},
    F = fun() ->
  mnesia:delete(Oid)
 end,
    mnesia:transaction(F).

delete(Table,Item) ->
    if 
        Table =:= urltable ->
            Oid = {urltable,Item};
        Table =:= address  ->
            Oid = {address,Item}
    end,
    F = fun() ->
        mnesia:delete(Oid)
    end,
    mnesia:transaction(F).
 
reset_tables() ->
    mnesia:clear_table(urltable),
    mnesia:clear_table(address),
    F = fun() ->
  foreach(fun mnesia:write/1, example_tables())
 end,
    mnesia:transaction(F).

get_plan(PlanId) ->
    F = fun() -> mnesia:read({design, PlanId}) end,
    mnesia:transaction(F)

这段是附加功能模块:

-module(tools).
-export([seperate_list/3,check/0]).

allurls() ->
    ["http://www.g.cn","http://www.sina.com.cn","http://www.xiaonei.com",
    "http://www.erlang.org","http://www.microsoft.com","http://www.yahoo.com",
    "http://www.google.com","http://www.dell.com","http://www.doban.com"].

% seperate_list([1,2,3,4,5],2,[])   ---> [ [1,2],[3,4],[5] ].
% seperate_list([1,2,3,4,5,6],2,[]) ---> [ [1,2],[3,4],[5,6] ].
seperate_list(List,NOP,LatestList) ->
    % NOP = Number of One Page.
    Rem =  length(List) rem NOP,
    Grp = lists:sublist(List,NOP+Rem),
    Ext = lists:subtract(List,Grp),   
    if 
        length(Grp) >  0 ->
            seperate_list(Ext,NOP,[Grp|LatestList]);
        length(Grp) =:= 0  ->
            LatestList
    end.
    

check() ->
    Allurl = allurls(),
    seperate_list(Allurl,2,[]).

Python2.5和python3.0性能循环测试

测试环境 Ubuntu 8.04 linux 2.6.24-23-386 i686 GNU/Linux

运行环境 python2.5/python3.0

分析命令：Top

测试结果分析：
分界线上段代码的用的无限循环。然后到某个值后跳出
1,2分别为python2.5,python3.0的执行环境
分界线下段代码的用的是遍历range
3,4分别为python2.5,python3.0的执行环境

先分析Python2.5
1,3 做横向比较, 1的CPU的消耗略小于2，而内存消耗明显高于2。
原因很简单：1 中 对i做累加，所以略占用计算资源。而2则是在内存中实现一组长序列（请原谅我未曾深入python C源码学习，range实现的数据结构我不了解）显然内存空间被大量占用。

再分析python3.0
2，4做横向比较， 内存消耗相差不大。因为python3.0中的range实质是从List->Iterator.但是CPU消耗是很大的。这个我们最后总结。

纵向比较1，2和3，4, 1的各方面都优于2。而3，4分别优于CPU和内存。

总结： python3.0 对 python2.5的数据结构做了很大改进（这里我们谈实现性能）。总体方向是很成功的。问题是，细节上，需要优化的地方太多太多，就像Guido之前所说。而至于python3.0 在实现上的改进的所有内容，请参见python.org.

就目前主流的 python2.5而言。把range作为循环的容器显然是很消耗资源的行为。
所以以后就该有个好习惯，如果循环序列很大，那么就用while做为一个好习惯。

虽然我们常说空间换时间，但在这种情况下，无非舍本逐末。

#!/usr/bin/env python
#-*- coding:utf-8 -*-

#-------------------------------------------
# Designer  : free.Won 
# Licence   : License on GPL Licence
# Archieved : Jan 1st 2009
#-------------------------------------------

import threading

class Multirun(threading.Thread):
def __init__(self):
 threading.Thread.__init__(self)
def run(self):
 i=1
 while 1:
  if i== 10**7:
      break
  print(i)
  i+=1

if __name__ == '__main__':
tset = []
for i in range(10000):
t = Multirun()
tset.append(t)
for one in tset:
 one.start()
 one.join()

1.
PID USER PR NI VIRT RES SHR S %CPU %MEM TIME+ COMMAND
17305 freefis 20 0 23152 12m 1524 S 10.3 0.7 0:03.36 python

2.
PID USER PR NI VIRT RES SHR S %CPU %MEM TIME+ COMMAND
17451 freefis 20 0 29836 18m 1720 S 28.0 1.0 0:08.92 python3.0


===========================================

#!/usr/bin/python3.0
#-*- coding:utf-8 -*-

#-------------------------------------------
# Designer  : free.Won 
# Licence   : License on GPL Licence
# Archieved : Jan 1st 2009
#-------------------------------------------

import threading

class Multirun(threading.Thread):
def __init__(self):
 threading.Thread.__init__(self)
def run(self):
 for i in range(10**7):
  print(i)

if __name__ == '__main__':
tset = []
for i in range(10000):
 t = Multirun()
 tset.append(t)
for one in tset:
 one.start()
 one.join()

3
PID USER PR NI VIRT RES SHR S %CPU %MEM TIME+ COMMAND
17389 freefis 20 0 176m 166m 1524 S 9.0 8.8 0:02.76 python

4
PID USER PR NI VIRT RES SHR S %CPU %MEM TIME+ COMMAND
17494 freefis 20 0 29836 18m 1720 S 47.9 1.0 0:10.46 python3.0

神州令

神州令 征天庭

年少梦狂吞江海 初醒摆弓射日月；傲雪炎阳铸经纬 孤舟昆仑祭轩辕。

笑谈浮生千杯过 醉悯天下苍生怜；棋点江山书韬略 峥嵘岁月惘然间。

天罡封臣凌烟阁 中原六路聚英贤；东海陈兵八千万 踏云乘风诛群仙。

怒火燎原斩鬼神 九州归定霜寒剑；古今豪杰出我辈 百年山河一朝卷。

求数组中最长递增子序列

今天睡前翻看《编程之美》，这是每晚的必读。程序与数学在这帮Microsoft的天才工程师们手中结合得通俗易懂。
当中有个题目：写一个时间复杂度尽可能低的程序，求数组最长递增子序列 的解答方式让我觉得是那么不尽人意。
原题：求一个1维数组（N个元素）中的最长递增子序列的长度。
例如 L = [1,-1,2,-3,4,-5,6,-7] 的最长子序列是 1,2,4,5.

书上的解答思路过于注重了各个元素之间相互的比较。从系统工程学来解释就是内耗过大。所以在这样的思路下，书上的最低时间复杂度为O(n*logn).

于是我干脆跳出了这样一个思路。把时间复杂度控制到了O(n).

1 取出第1个元素E1为新的序列S1,
2 取出第2个元素E2和E1做比较，如果E2>E1，则追加到S1中; 否则，把E2放到新序列S2中
3 取出第3个元素E3和所有已有序列的最后一个元素比较，判断法则如同步骤2
.......
当序列遍历完成时，程序结束。

这是一个动态规划的思想。 就拿L为例,遍历L：每个虚线之间代表一个步骤
－－－－－－－－－－－－－－－－－－－－－－－－－－－－－－
1
－－－－－－－－－－－－－－－－－－－－－－－－－－－－－－
1
－1
－－－－－－－－－－－－－－－－－－－－－－－－－－－－－－
1，2
－1，2
－－－－－－－－－－－－－－－－－－－－－－－－－－－－－－
1，2
－1，2
－3
－－－－－－－－－－－－－－－－－－－－－－－－－－－－－－
1，2，4
－1，2，4
－3，4
－－－－－－－－－－－－－－－－－－－－－－－－－－－－－－
1，2，4
－1，2，4
－3，4
－5
－－－－－－－－－－－－－－－－－－－－－－－－－－－－－－
1，2，4，6
－1，2，4，6
－3，4，6
－5 ，6
－－－－－－－－－－－－－－－－－－－－－－－－－－－－－－
1，2，4，6
－1，2，4，6
－3，4，6
－5 ，6
－7
－－－－－－－－－－－－－－－－－－－－－－－－－－－－－－


遍历结束。这样就列举出了所有的递增子序列，自然最长序列一目了然。但是这样的空间消耗又大了。 那么我们可以在这个基础上优化空间。
我们观察上面的各个步骤可以发现：
如果各个序列组的最后一个元素相同，那么非最长序列就一定不可能是后结果。（相关证明请参见文章末尾）
这样的话，我们重新遍历L。
－－－－－－－－－－－－－－－－－－－－－－－－－－－－－－
1
－－－－－－－－－－－－－－－－－－－－－－－－－－－－－－
1
－1
－－－－－－－－－－－－－－－－－－－－－－－－－－－－－－
1，2
－1，2
－－－－－－－－－－－－－－－－－－－－－－－－－－－－－－
1，2
－1，2
－3
－－－－－－－－－－－－－－－－－－－－－－－－－－－－－－
1，2，4
－1，2，4
－3，4 该序列可删除
－－－－－－－－－－－－－－－－－－－－－－－－－－－－－－
1，2，4
－1，2，4
－5
－－－－－－－－－－－－－－－－－－－－－－－－－－－－－－
1，2，4，6
－1，2，4，6
－3，4，6 该序列可删除
－5 ，6 该序列可删除
－－－－－－－－－－－－－－－－－－－－－－－－－－－－－－
1，2，4，6
－1，2，4，6
－5 ，6
－7
－－－－－－－－－－－－－－－－－－－－－－－－－－－－－－
（完）


证明： 如果各个序列组的最后一个元素相同，那么非最长序列就一定不可能是后结果。
blogspot不支持TeX公式所以这里用文字叙述。

对于S1,S2....SN满足S1.LastElement = .... = SN.LastElement
=> 当L中取出元素E作比较，
如果E>S1.LastElement,则E>Sx.LastElement， 其中x属于1......N.
如果E无论L中的元素为何值，则S1到SN之间的长度大小关系不变。
=>S1到SN中的最长序列Smax不变（注意不是长度不变，而是序号不变）。
=>如果Sx不是最长序列，则Sx必然不可能成为整个程序的最终结果。

Python3.0 Urllib体验

首先推荐大家接触python3.0. 有争议的东西毕竟是好东西.
http://python.org/ftp/python/3.0/Python-3.0.tar.bz2 可下载python3.0 可以推荐大家玩玩. 如果担心和当前python环境影响没关系.解压后进入该目录.

sudo su
mkdir /opt/python3.0
./configure --prefix=/opt/python3.0 && make && make install
ln -s /opt/python3.0/bin/python3.0  /usr/bin/python3.0

python3.0出了之后.像我这样热衷爬虫的偏执狂最关心的就是它的urllib.
可以去python.org看看文档,urllib是修改比较剧烈的模块.urllib2和urllib被整合到了urllib包.相应的方法都被归到不同模块中. 在下面的代码中,urlopen, urlencode 被定义在request,parse模块.

#!/usr/bin/python3.0
# -*- coding:utf-8 -*-

#-------------------------------------------
# Designer  : Free.Wang 
# E-mail    : freefis@Gmail.com
# Licence   : License on GPL Licence
# Archieved : Dec 27 2008
#-------------------------------------------

from urllib import request,parse

class Makesocket:
""" Build for Make a full HttpRequest via POST/GET """
  def __init__(self,loginurl,param):
      # login url to recieve param , Post/Get param
       self.loginurl = loginurl
       self.param    = param

  def cookie(self):
      """make a Container for Cookie"""
       self.cookies = request.HTTPCookieProcessor()
       # opener will be used to open Url with cookie.
       # build the cookie container.
       self.opener  = request.build_opener(self.cookies)
      request.install_opener(self.opener)

  def run(self):
     """ get Response. """
     # serialize the Param
      # Login to Get Cookie ,Which will be Push into self.opener.
     # get request with Cookie.
      self.encodeparam = parse.urlencode(self.param)
     request.urlopen(self.loginurl,self.encodeparam)
      conn = self.opener.open("http://www.hellofreefis.com/profile/")
     print(conn)

if __name__ == '__main__':
   oginurl  = "http://www.hellofreefis.com/info_oper.php?tag=signin&pageref="
   param    = {
      'name':"freefis",
      'pwd':"freefis' passwod"
   }
   conn = Makesocket(loginurl,param)
   conn.cookie()
   conn.run()

测试成功.Py3.0的改动还是很大的.但愿Python4.0出现的时候,不会再来革命.我个人角度对这次改动还是很肯定.感谢Guido把reduce函数保留到functools模块中.
希望各大公司和*nix操作系统发行者能尽快支持python3.0.

P.S:以上代码让我觉得很"悲剧".blogspot转义空格的支持不够到位.若要迁移代码的朋友,请自行缩进.