拉取第一页网页信息，拉取网页源码分析，设计对应的正则表达式，测试拉取信息正确与否
确定网页结构，以及爬虫结构
具体的工作流程为：给定一个起始网址，加入任务队列，engine拿去url将任务发给Fetcher拉取url对应网页，调用对应层级的解析器解析数据并把新的url加入任务队列。（这里利用go的函数特性，将对应的解析器与request封装在一起加入队列，可以非常方便的调用对应的解析器）
编码测试
问题:engine需要等待单个fetcher拉取网页数据（而网络I/O耗时很久）
解决其他任务等待fetcher网络I/O速度慢的问题
知道了fetcherer这个模块最耗时，现在的解决方法就是将Fetcher抽离出去，开goroutine来处理这个网络I/O
任务与解析器是封装在一个结构体，可以直接将两个操作一起放入goroutine封装成一个worker
func Worker(r Request) (ParseResult,error) {
log.Printf("Fetching url:%v",r.Url)
body , err := fetcher.Fetch(r.Url)
if err!=nil{
log.Fatal("Fetcher err:",r.Url,err)
return ParseResult{},nil
}
return r.ParserFunc(body),nil
}所以并发版的的设计为：
增加一个调度器，管理任务的分发（如果不增加一个Scheduler那么engine和Worker就还是串行的了）
func (e *ConEngine) Run(seeds ...Request) {
out := make(chan ParseResult,10)
//创建一个分发任务给worker的channel
e.Scheduler.Run()
for i := 0;i < e.WorkerCount;i++{
//创建worker
createWorker(e.Scheduler.WorkerChan(),out,e.Scheduler)
}
//将初始任务交给Schedual
for _,r := range seeds{
e.Scheduler.Submit(r)
}
//从Worker里面读取解析好的数据并给engine
for {
result := <-out
for _,item := range result.Items{
log.Printf("Got item %v:",item)
}
//将新获得的任务交给Schedual进行调度
for _,request := range result.Requests{
e.Scheduler.Submit(request)
}
}
}
func createWorker(in chan Request,out chan ParseResult) {
go func() {
for{
//收任务
request := <-in
//工作
worker,err := Worker(request)
if err != nil{
continue
}
//发出任务
out<-worker
}
}()
}func (ssche *SimpleScheduler)Submit(r engine.Request) {
ssche.workerChan <- r
}
func (ssche *SimpleScheduler)Run() {
ssche.workerChan = make(chan engine.Request)
}
func (ssche *SimpleScheduler)WorkerChan()chan engine.Request {
return ssche.workerChan
}去重问题（如果将数据量不大，可以直接定义一个map直接在内存中就去重）或是将url存在redis 集合里面 使用redis集合去重
循坏等待，死锁问题
-
这里Scheduler要将任务发给Worker的时候要保证in这个channel的接收队列还挂着Worker（就是还有空闲的Worker）
-
但是Worker要做完一个任务的前提是需要将自己做完的结果成功通过out发送给engine
-
但是engine接收这个结果的时候又需要将获得的任务中的request交给scheduler去调度
-
但是1中的Scheduler还还有任务等着去发给Woker，这时新的任务又来了，然后就会在这卡住。
这样就产生了一个循环等待的问题。
那么怎么解决呢？
一个很简单的想法就是破坏这个循环等待，
直接开一个goroutine来发送任务就好了（gorountine的特性：会立刻在主函数里返回，然后去做自己的事情）这样就不会在上文中的4卡住，每次来一个request，都开一个goroutine去解决，然后立刻开始处理下一个request
func (ssche *SimpleScheduler)Submit(r engine.Request) {
go func() {ssche.workerChan <- r}()
}在使用并发以后，爬取数据的速度明显提高不少，但随之而来的就是网站的限流和封ip。。这俄格问题，可以通过更换代理、更换代理IP、加一个ticker来限速等方法处理。
但是这里还是有一个问题，这里Scheduler创建的子goroutine和所有的woker都是在共用一个goroutine，一边抢着发，一边也抢着收，这显然不是我们想要的。
所以我们可以给每一个worker创建它独有的channel，（这个channel就可以代表worker）然后再设计一个channel，来用来接收worker的channel（相当于一个总控的channel）（在创建worker的时候呢把自己的channel发给这个发给这个总控的channel，说明我可以开始工作了）
func (q *QueueScheduler)WorkerReady(w chan engine.Request) {
q.workerChan <- w
}这里可以设计两个队列，一个存放需要发出的Request,一个存放接收Request的Worker
func (q *QueueScheduler)Run() {
//初始化两个channel
q.workerChan = make(chan chan engine.Request)
q.requestChan = make(chan engine.Request)
go func() {
//对应的两个队列
var requestQ []engine.Request
var workerQ []chan engine.Request
for{
//获取队首的request和空闲worker
var activeRequest engine.Request
var activeWorker chan engine.Request
if len(requestQ)>0&&len(workerQ)>0{
activeWorker = workerQ[0]
activeRequest = requestQ[0]
}
select {
//如果接收到request，把request加入request队列
case r := <-q.requestChan:
requestQ = append(requestQ,r)
//如果接收到worker的channel（说明这个worker空闲下来了），把这个加入队列
case w:=<-q.workerChan:
workerQ = append(workerQ,w)
//如果可以成功将request送入wokerchannel,各自队列中的元素出队
case activeWorker<-activeRequest:
workerQ = workerQ[1:]
requestQ = requestQ[1:]
}
}
}()
}go func() {
for{
//每次收到任务，处理任务，做完任务以后，就告诉Scheduler我又可以了
ready.WorkerReady(in)
request := <-in
worker,err := Worker(request)
if err != nil{
continue
}
out<-worker
}
}()func (e *ConEngine) Run(seeds ...Request) {
out := make(chan ParseResult,10)
e.Scheduler.Run()
for i := 0;i < e.WorkerCount;i++{
createWorker(e.Scheduler.WorkerChan(),out,e.Scheduler)
}
for _,r := range seeds{
e.Scheduler.Submit(r)
}
for {
result := <-out
for _,item := range result.Items{
//将数据发出去，让对应的存储模块存储数据
go func() {
e.ItemChan<- item
}()
}
for _,request := range result.Requests{
e.Scheduler.Submit(request)
}
}
}
这里的存储使用elasticsearch（本身提供搜索功能，可以直接存储json数据）
可以看到会影响系统性能的就是 ItemSaver 和Worker（这两个都会使用I/O）所以可以很容易的想到将这两个模块独立出去，发布到其他的服务器上提供服务。
这里可以是使用 rpc 消息队列等技术进行服务。
这里先使用rpc
这里选用了较为简单的jsonrpc（官方自带的跨语言rpc）
//包装save方法为rpc服务
func (saver ItemSaverService)Save(item engine.Item,result *string)error {
err := presist.Save(item,saver.Client,saver.Index)
log.Printf("Item %v saved",item)
if err == nil {
*result = "ok"
}else {
log.Printf("Error saving %v,%s",item,err)
}
return err
}//注册初始化rpc服务
func serveRpc(host string,index string) error {
//创建elsticsearch的客户端
client, err := elastic.NewClient(elastic.SetSniff(false),
elastic.SetURL(config.Host))
if err != nil{
return err
}
return
//注册服务
rpc_support.ServiceRpc(host,&persistRpc.ItemSaverService{
Client: client,
Index: index,
})
}func ItemSaver(host string)(chan engine.Item ,error){
out := make(chan engine.Item)
client, err := rpc_support.NewClient(host)
if err !=nil {
return nil,err
}
if err !=nil {
return nil,err
}
go func() {
itemCount := 1
for {
item := <- out
log.Printf("Item Saver:got item #%d %v",itemCount,item)
itemCount++
//call rpc save item
result := ""
//调用rpc服务
err = client.Call(config.ItemSaverService,item,&result)
if err != nil || result != "ok"{
log.Printf("result: %s",result)
}
if err != nil{
log.Printf("Item save error : %v,%s",item,err)
}
}
}()
return out,nil
}待定