Golang实现简单爬虫框架(3)——简单并发版
source link: https://www.tuicool.com/articles/aArUjuM
Go to the source link to view the article. You can view the picture content, updated content and better typesetting reading experience. If the link is broken, please click the button below to view the snapshot at that time.
在上篇文章 Golang实现简单爬虫框架(2)——单任务版爬虫 中我们实现了一个简单的单任务版爬虫,对于单任务版爬虫,每次都要请求页面,然后解析数据,然后才能请求下一个页面。整个过程中,获取网页数据速度比较慢,那么我们就把获取数据模块做成并发执行。在项目的基础上,实现多任务并发版爬虫。
项目github地址: https://github.com/NovemberCh... , 回滚到相应记录食用,效果更佳。
1、项目架构
首先我们把但任务版爬虫架构中的 Fetcher
模块和 Parser
模块合并成一个 Worker
模块,然后并发执行 Worker
模块
然后得到并发版的架构图:
- 在并发版爬虫中,会同时执行多个
Worker
,每个Worker
任务接受一个Request
请求,然后请求页面解析数据,输出解析出的Requests
和Item
- 因为又很多
Request
和Worker
,所以还需要Scheduler
模块,负责对请求任务的调度处理 -
Engine
模块接受Worker
发送的Requests
和Items
,当前我们先把Items
打印出,把解析出的Request
发送给调度器 - 其中
Engine
和Scheduler
是一个goruntine
,Worker
包含多个goruntine
,各个模块之间都是用channel
进行连接
先放上重构后的项目文件结构:
2、Worker实现
我们从engine.go中提取下面功能作为Worker模块,同时把engine.go 更名为simple.go。修改后的simple.go文件请自行调整,或者去github项目 源代码 回滚查看。
engine/worker.go
package engine import ( "crawler/fetcher" "log" ) // 输入 Request, 返回 ParseResult func worker(request Request) (ParseResult, error) { log.Printf("Fetching %s\n", request.Url) content, err := fetcher.Fetch(request.Url) if err != nil { log.Printf("Fetch error, Url: %s %v\n", request.Url, err) return ParseResult{}, err } return request.ParseFunc(content), nil }
对于每一个Worker接受一个请求,然后返回解析出的内容
3、并发引擎Concurrent实现
请大家根据架构图来看,效果会更好。
package engine import "log" // 并发引擎 type ConcurrendEngine struct { Scheduler Scheduler // 任务调度器 WorkerCount int // 任务并发数量 } // 任务调度器 type Scheduler interface { Submit(request Request) // 提交任务 ConfigMasterWorkerChan(chan Request) // 配置初始请求任务 } func (e *ConcurrendEngine) Run(seeds ...Request) { in := make(chan Request) // scheduler的输入 out := make(chan ParseResult) // worker的输出 e.Scheduler.ConfigMasterWorkerChan(in) // 把初始请求提交给scheduler // 创建 goruntine for i := 0; i < e.WorkerCount; i++ { createWorker(in, out) } // engine把请求任务提交给 Scheduler for _, request := range seeds { e.Scheduler.Submit(request) } itemCount := 0 for { // 接受 Worker 的解析结果 result := <-out for _, item := range result.Items { log.Printf("Got item: #%d: %v\n", itemCount, item) itemCount++ } // 然后把 Worker 解析出的 Request 送给 Scheduler for _, request := range result.Requests { e.Scheduler.Submit(request) } } } // 创建任务,调用worker,分发goruntine func createWorker(in chan Request, out chan ParseResult) { go func() { for { request := <-in result, err := worker(request) if err != nil { continue } out <- result } }() }
4、任务调度器Scheduler实现
scheduler/scheduler.go
package scheduler import "crawler/engine" type SimpleScheduler struct { workerChan chan engine.Request } func (s *SimpleScheduler) Submit(request engine.Request) { // send request down to worker chan go func() { s.workerChan <- request }() } // 把初始请求发送给 Scheduler func (s *SimpleScheduler) ConfigMasterWorkerChan(in chan engine.Request) { s.workerChan = in }
5、main函数
package main import ( "crawler/engine" "crawler/scheduler" "crawler/zhenai/parser" ) func main() { e := engine.ConcurrendEngine{ // 配置爬虫引擎 Scheduler: &scheduler.SimpleScheduler{}, WorkerCount: 50, } e.Run(engine.Request{ // 配置爬虫目标信息 Url: "http://www.zhenai.com/zhenghun", ParseFunc: parser.ParseCityList, }) }
6、小结
本次博客我们实现一个最简单的并发版爬虫,调度器源源不断的接受任务,一旦有一个worker空闲,就给其分配任务。这样子有一个缺点,就是我们不知道我们分发出那么多worker的工作情况,对worker的控制力比较弱,所以在下次博客中会用队列来实现任务调度。
如果想获取 Google工程师深度讲解go语言 视频资源的,可以在评论区留言。
项目的 源代码 已经托管到Github上,对于各个版本都有记录,欢迎大家查看,记得给个star,在此先谢谢大家了
Recommend
About Joyk
Aggregate valuable and interesting links.
Joyk means Joy of geeK