Golang实现简单爬虫框架（4）——队列实现并发任务调度

前言

在上一篇文章《Golang实现简单爬虫框架（3）——简单并发版》中咱们实现了一个最简单并发爬虫，调度器为每个Request建立一个goroutine，每一个goroutine往Worker队列中分发任务，发完就结束。全部的Worker都在抢一个channel中的任务。可是这样作仍是有些许不足之处，好比控制力弱：全部的Worker在抢同一个channel中的任务，咱们没有办法控制给哪个worker任务。

其实咱们能够本身作一个任务分发的机制，咱们来决定分发给哪个Worker

注意：本次并发是在上一篇文章简单并发实现的基础上修改，因此没有贴出所有代码，只是贴出部分修改部分，要查看完整项目代码，能够查看上篇文章，或者从github下载项目源代码查看

一、项目架构

在上一篇文章实现简单并发的基础上，咱们修改下Scheduler的任务分发机制

• 当Scheduler接收到一个Request后，不能直接发给Worker，也不能为每一个Request建立一个goroutine，因此这里使用一个Request队列
• 同时咱们想对Worker实现一个更多的控制，能够决定把任务分发给哪个Worker，因此这里咱们还须要一个Worker队列
• 当有了Request和Worker，咱们就能够把选择的Request发送给选择的Worker

二、队列实现任务调度器

在scheduler目录下建立queued.go文件

package scheduler

import "crawler/engine"

// 使用队列来调度任务

type QueuedScheduler struct {
	requestChan chan engine.Request		// Request channel
    // Worker channel, 其中每个Worker是一个 chan engine.Request 类型
	workerChan  chan chan engine.Request	
}

// 提交请求任务到 requestChannel
func (s *QueuedScheduler) Submit(request engine.Request) {
	s.requestChan <- request
}

func (s *QueuedScheduler) ConfigMasterWorkerChan(chan engine.Request) {
	panic("implement me")
}

// 告诉外界有一个 worker 能够接收 request
func (s *QueuedScheduler) WorkerReady(w chan engine.Request) {
	s.workerChan <- w
}

func (s *QueuedScheduler) Run() {
    // 生成channel
	s.workerChan = make(chan chan engine.Request)
	s.requestChan = make(chan engine.Request)
	go func() {
		// 建立请求队列和工做队列
		var requestQ []engine.Request
		var workerQ []chan engine.Request
		for {
			var activeWorker chan engine.Request
			var activeRequest engine.Request
			
            // 当requestQ和workerQ同时有数据时
			if len(requestQ) > 0 && len(workerQ) > 0 {
				activeWorker = workerQ[0]
				activeRequest = requestQ[0]
			}
			
			select {
			case r := <-s.requestChan: // 当 requestChan 收到数据
				requestQ = append(requestQ, r)
			case w := <-s.workerChan: // 当 workerChan 收到数据
				workerQ = append(workerQ, w)
			case activeWorker <- activeRequest: // 当请求队列和认读队列都不为空时，给任务队列分配任务
				requestQ = requestQ[1:]
				workerQ = workerQ[1:]
			}
		}
	}()
}

复制代码

三、爬虫引擎

修改后的concurrent.go文件以下

package engine

import (
	"log"
)

// 并发引擎
type ConcurrendEngine struct {
	Scheduler   Scheduler
	WorkerCount int
}

// 任务调度器
type Scheduler interface {
	Submit(request Request) // 提交任务
	ConfigMasterWorkerChan(chan Request)
	WorkerReady(w chan Request)
	Run()
}

func (e *ConcurrendEngine) Run(seeds ...Request) {

	out := make(chan ParseResult)
	e.Scheduler.Run()

	// 建立 goruntine
	for i := 0; i < e.WorkerCount; i++ {
		createWorker(out, e.Scheduler)
	}

	// 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)
		}
	}
}

func createWorker(out chan ParseResult, s Scheduler) {
    // 为每个Worker建立一个channel
	in := make(chan Request)
	go func() {
		for {
			s.WorkerReady(in) // 告诉调度器任务空闲
			request := <-in
			result, err := worker(request)
			if err != nil {
				continue
			}
			out <- result
		}
	}()
}
复制代码

四、main函数

package main

import (
	"crawler/engine"
	"crawler/scheduler"
	"crawler/zhenai/parser"
)

func main() {
	e := engine.ConcurrendEngine{
		Scheduler:   &scheduler.QueuedScheduler{},// 这里调用并发调度器
		WorkerCount: 50,
	}
	e.Run(engine.Request{
		Url:       "http://www.zhenai.com/zhenghun",
		ParseFunc: parser.ParseCityList,
	})
}
复制代码

运行结果以下：

五、总结

在这篇文章中咱们使用队列实现对并发任务的调度，从而实现了对Worker的控制。咱们如今并发有两种实现方式，可是他们的调度方法是不一样的，为了代码的统一，因此在下一篇文章中的内容有：

• 对项目作一个同构
• 添加数据的存储模块。

若是想获取Google工程师深度讲解go语言视频资源的，能够在评论区留下邮箱。

项目的源代码已经托管到Github上，对于各个版本都有记录，欢迎你们查看，记得给个star，在此先谢谢你们

若是以为博客不错，劳烦大人给个赞，