下面这个是主模块,后面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,[]).
Posts
