30 辅助任务管理：任务优先级、去重与失败处理
你好，我是郑建勋。
这节课，让我们给系统加入一些辅助功能，把爬虫流程变得更完善一些。这些功能包括：爬虫最大深度、请求不重复、优先队列、以及随机的User-Agent。
设置爬虫最大深度
当我们用深度和广度优先搜索爬取一个网站时，为了防止访问陷入到死循环，同时控制爬取的有效链接的数量，一般会给当前任务设置一个最大爬取深度。最大爬取深度是和任务有关的，因此我们要在 Request 中加上MaxDepth 这个字段，它可以标识到爬取的最大深度。Depth 则表示任务的当前深度，最初始的深度为0。
type Request struct {
Url string
Cookie string
WaitTime time.Duration
Depth int
MaxDepth int
ParseFunc func([]byte, *Request) ParseResult
}
那在异步爬取的情况下，我们怎么知道当前网站的深度呢？最好的时机是在采集引擎采集并解析爬虫数据，并将下一层的请求放到队列中的时候。以我们之前写好的ParseURL函数为例，在添加下一层的URL时，我们将Depth加1，这样就标识了下一层的深度。
func ParseURL(contents []byte, req *collect.Request) collect.ParseResult {
re := regexp.MustCompile(urlListRe)
matches := re.FindAllSubmatch(contents, -1)
result := collect.ParseResult{}
for _, m := range matches {
u := string(m[1])
result.Requesrts = append(
result.Requesrts, &collect.Request{
Url: u,
WaitTime: req.WaitTime,
Cookie: req.Cookie,
Depth: req.Depth + 1,
MaxDepth: req.MaxDepth,
ParseFunc: func(c []byte, request *collect.Request) collect.ParseResult {
return GetContent(c, u)
},
})
}
return result
}
最后一步，我们在爬取新的网页之前，判断最大深度。如果当前深度超过了最大深度，那就不再进行爬取。这部分的完整代码你可以查看分支v0.1.7。
func (r *Request) Check() error {
if r.Depth > r.MaxDepth {
return errors.New("Max depth limit reached")
}
return nil
}
func (s *Schedule) CreateWork() {
for {
r := <-s.workerCh
if err := r.Check(); err != nil {
s.Logger.Error("check failed",
zap.Error(err),
)
continue
}
...
}
}
避免请求重复
为了避免爬取时候的死循环，避免无效的爬取，我们常常需要检测请求是否重复，这时我们需要考虑3个问题：
- 用什么数据结构来存储数据才能保证快速地查找到请求的记录？
- 如何保证并发查找与写入时，不出现并发冲突问题？
- 在什么条件下，我们才能确认请求是重复的，从而停止爬取？
要解决第一个问题我们可以用一个简单高效的结构：哈希表。我们可以借助哈希表查找o(1)。另外，由于Go语言中的哈希表是不支持并发安全的，为了解决第二个问题，我们还需要在此基础上加一个互斥锁。而第三个问题我们需要在爬虫采取之前进行检查。
在解决上面的三个问题之前，我们先优化一下代码。我们之前的 Request 结构体会在每一次请求时发生变化，但是我们希望有一个字段能够表示一整个网站的爬取任务，因此我们需要抽离出一个新的结构Task作为一个爬虫任务，而 Request 则作为单独的请求存在。有些参数是整个任务共有的，例如Task中的Cookie、MaxDepth（最大深度）、WaitTime（默认等待时间）和RootReq（任务中的第一个请求）。
type Task struct {
Url string
Cookie string
WaitTime time.Duration
MaxDepth int
RootReq *Request
Fetcher Fetcher
}
// 单个请求
type Request struct {
Task *Task
Url string
Depth int
ParseFunc func([]byte, *Request) ParseResult
}
由于抽象出了Task，代码需要做对应的修改，例如我们需要把初始的Seed种子任务替换为Task结构。
for i := 0; i <= 0; i += 25 {
str := fmt.Sprintf("<https://www.douban.com/group/szsh/discussion?start=%d>", i)
seeds = append(seeds, &collect.Task{
...
Url: str,
RootReq: &collect.Request{
ParseFunc: doubangroup.ParseURL,
},
})
}
同时，在深度检查时，每一个请求的最大深度需要从Task字段中获取。
func (r *Request) Check() error {
if r.Depth > r.Task.MaxDepth {
return errors.New("Max depth limit reached")
}
return nil
}
完整代码你可以查看v0.1.8。
接下来，我们继续用一个哈希表结构来存储历史请求。由于我们希望随时访问哈希表中的历史请求，所以把它放在Request、Task中都不合适。 放在调度引擎中也不合适，因为调度引擎从功能上讲，应该只负责调度才对。所以，我们还需要完成一轮抽象，将调度引擎抽离出来作为一个接口，让它只做调度的工作，不用负责存储全局变量等任务。
所以我们就构建一个新的结构Crawler作为全局的爬取实例，将之前Schedule中的options迁移到Crawler中，Schedule只处理与调度有关的工作，并抽象为了Scheduler接口。
type Crawler struct {
out chan collect.ParseResult
options
}
type Scheduler interface {
Schedule()
Push(...*collect.Request)
Pull() *collect.Request
}
type Schedule struct {
requestCh chan *collect.Request
workerCh chan *collect.Request
reqQueue []*collect.Request
Logger *zap.Logger
}
在Scheduler中，Schedule方法负责启动调度器，Push方法会将请求放入到调度器中，而Pull方法则会从调度器中获取请求。我们也需要对代码做相应的调整，这里就不再赘述了，具体你可以参考v0.1.9。调度器抽象为接口后，如果我们有其他的调度器算法实现，也能够非常方便完成替换了。
现在，我们在Crawler中加入Visited哈希表，用它存储请求访问信息，增加VisitedLock来确保并发安全。
type Crawler struct {
out chan collect.ParseResult
Visited map[string]bool
VisitedLock sync.Mutex
options
}
Visited中的Key是请求的唯一标识，我们现在先将唯一标识设置为URL + method方法，并使用MD5生成唯一键。后面我们还会为唯一标识加上当前请求的规则条件。
// 请求的唯一识别码
func (r *Request) Unique() string {
block := md5.Sum([]byte(r.Url + r.Method))
return hex.EncodeToString(block[:])
}
接着，编写HasVisited方法，判断当前请求是否已经被访问过。StoreVisited方法用于将请求存储到Visited哈希表中。
func (e *Crawler) HasVisited(r *collect.Request) bool {
e.VisitedLock.Lock()
defer e.VisitedLock.Unlock()
unique := r.Unique()
return e.Visited[unique]
}
func (e *Crawler) StoreVisited(reqs ...*collect.Request) {
e.VisitedLock.Lock()
defer e.VisitedLock.Unlock()
for _, r := range reqs {
unique := r.Unique()
e.Visited[unique] = true
}
}
最后在Worker中，在执行request前，判断当前请求是否已被访问。如果请求没有被访问过，将request放入Visited哈希表中。
func (s *Crawler) CreateWork() {
for {
r := s.scheduler.Pull()
if err := r.Check(); err != nil {
s.Logger.Error("check failed",
zap.Error(err),
)
continue
}
// 判断当前请求是否已被访问
if s.HasVisited(r) {
s.Logger.Debug("request has visited",
zap.String("url:", r.Url),
)
continue
}
// 设置当前请求已被访问
s.StoreVisited(r)
...
}
}
最后要注意的是，哈希表需要用make进行初始化，要不然在运行时访问哈希表会直接报错。（完整的代码位于v0.2.0）。
设置优先队列
我们要给项目增加的第三个功能就是优先队列。
爬虫任务的优先级有时并不是相同的，一些任务需要优先处理。因此，接下来我们就来设置一个任务的优先队列。优先队列还可以分成多个等级，不过在这里我将它简单地分为了两个等级，即优先队列和普通队列。优先级更高的请求会存储到priReqQueue优先队列中。
type Schedule struct {
requestCh chan *collect.Request
workerCh chan *collect.Request
priReqQueue []*collect.Request
reqQueue []*collect.Request
Logger *zap.Logger
}
在调度函数Schedule中，我们会优先从优先队列中获取请求。而在放入请求时，如果请求的优先级更高，也会单独放入优先级队列。
最后我们还修复了之前遗留的一个Bug，将变量req、ch放置到for循环外部，防止丢失请求的可能性。
func (s *Schedule) Schedule() {
var req *collect.Request
var ch chan *collect.Request
for {
if req == nil && len(s.priReqQueue) > 0 {
req = s.priReqQueue[0]
s.priReqQueue = s.priReqQueue[1:]
ch = s.workerCh
}
if req == nil && len(s.reqQueue) > 0 {
req = s.reqQueue[0]
s.reqQueue = s.reqQueue[1:]
ch = s.workerCh
}
select {
case r := <-s.requestCh:
if r.Priority > 0 {
s.priReqQueue = append(s.priReqQueue, r)
} else {
s.reqQueue = append(s.reqQueue, r)
}
case ch <- req:
req = nil
ch = nil
}
}
}
执行后输出结果为：
{"level":"INFO","ts":"2022-11-05T21:40:18.339+0800","caller":"crawler/main.go:19","msg":"log init end"}
{"level":"INFO","ts":"2022-11-05T21:40:22.067+0800","caller":"engine/schedule.go:163","msg":"get result: <https://www.douban.com/group/topic/278041246/>"}
{"level":"INFO","ts":"2022-11-05T21:40:22.150+0800","caller":"engine/schedule.go:163","msg":"get result: <https://www.douban.com/group/topic/278040957/>"}
...
完整的代码位于v0.2.1。
设置随机 User-Agent
我们给项目增加的第四个功能是 User-Agent 随机性。 为了避免服务器检测到我们使用了同一个User-Agent，继而判断出是同一个客户端在发出请求，我们可以为发送的User-Agent加入随机性。这个操作的本质就是将浏览器的不同型号与不同版本拼接起来，组成一个新的User-Agent。
随机生成User-Agent的逻辑位于extensions/randomua.go中，里面枚举了不同型号的浏览器和不同型号的版本，并且通过排列组合产生了不同的User-Agent。
最后一步，我们要在采集引擎中调用GenerateRandomUA函数，将请求头设置为随机的User-Agent，如下所示：
func (b BrowserFetch) Get(request *spider.Request) ([]byte, error) {
...
req.Header.Set("User-Agent", extensions.GenerateRandomUA())
resp, err := client.Do(req)
完整的代码你可以参考v0.2.2分支。
进行失败处理
在课程的最后，我们来看一看失败处理。
我们在爬取网站时，网络超时等诸多潜在风险都可能导致爬取失败。这时，我们可以对失败的任务进行重试。但是如果网站多次失败，那就没有必要反复重试了，我们可以将它们放入单独的队列中。为了防止失败请求日积月久导致的内存泄露，同时也为了在程序崩溃后能够再次加载这些失败网站，我们最后还需要将这些失败网站持久化到数据库或文件中。
这节课我们先完成前半部分，即失败重试。后半部分会在第32讲存储引擎中详细介绍。我们要在全局Crawler中存储failures哈希表，设置Key为请求的唯一键，用于快速查找。failureLock互斥锁用于并发安全。
type Crawler struct {
...
failures map[string]*collect.Request // 失败请求id -> 失败请求
failureLock sync.Mutex
}
当请求失败之后，调用SetFailure方法将请求加入到failures哈希表中，并且把它重新交由调度引擎进行调度。这里我们为任务Task引入了一个新的字段Reload，标识当前任务的网页是否可以重复爬取。如果不可以重复爬取，我们需要在失败重试前删除Visited中的历史记录。
func (e *Crawler) SetFailure(req *collect.Request) {
if !req.Task.Reload {
e.VisitedLock.Lock()
unique := req.Unique()
delete(e.Visited, unique)
e.VisitedLock.Unlock()
}
e.failureLock.Lock()
defer e.failureLock.Unlock()
if _, ok := e.failures[req.Unique()]; !ok {
// 首次失败时，再重新执行一次
e.failures[req.Unique()] = req
e.scheduler.Push(req)
}
// todo: 失败2次，加载到失败队列中
}
如果失败两次，就将请求单独加载到失败队列中，并在后续进行持久化，这一步操作我们之后再进行。失败重试的完整代码位于v0.2.3。
总结
这节课，我们为爬虫系统增加了丰富的辅助任务，包括：设置爬虫的最大深度，避免了重复爬取、设置优先队列、设置随机的User-Agent，另外我们还进行了任务失败的处理。
随着爬虫系统的演进，我们还会有更加复杂的功能与需求。后面的内容，我们还会介绍限速器，并借助JS虚拟机实现更加灵活的爬虫系统。
课后题
学完这节课也给你留一道思考题。
之前我们实现了任务的优先队列，但我们目前只支持两种优先级。如果要支持多种优先级，你知道如何实现吗？
欢迎你在留言区与我交流讨论，我们下节课见。
- Realm 👍（4） 💬（0）
https://shimo.im/docs/5rk9dVyblnFzZLqx 根据课程讲解以及原代码，自己理解整理的调度过程。
2022-12-18 - Geek_crazydaddy 👍（2） 💬（1）
把worker获取任务的channel换成channel切片，索引值就是优先级，然后用多个select按序监听这些channel，而且要加default，没读到就立即跳过？
2022-12-17 - 顷 👍（0） 💬（0）
if !req.Task.Reload { ... } // 这里我们为任务 Task 引入了一个新的字段 Reload，标识当前任务的网页是否可以重复爬取。如果不可以重复爬取，我们需要在失败重试前删除 Visited 中的历史记录。 这里逻辑是不是反了？如果能重复爬，才需要再重新调度之前删掉记录吧。
2023-01-13 - 翡翠虎 👍（0） 💬（1）
用哈希表结构有什么好处？这样的话是不是就显得单机了？如果用类似redis这样的存储，加上布谷鸟算法，能够做到既省空间又支持多机协同，会不会更好？
2022-12-20