摘要：在上篇文章實現(xiàn)簡單爬蟲框架單任務(wù)版爬蟲中我們實現(xiàn)了一個簡單的單任務(wù)版爬蟲，對于單任務(wù)版爬蟲，每次都要請求頁面，然后解析數(shù)據(jù)，然后才能請求下一個頁面。

在上篇文章Golang實現(xiàn)簡單爬蟲框架（2）——單任務(wù)版爬蟲中我們實現(xiàn)了一個簡單的單任務(wù)版爬蟲，對于單任務(wù)版爬蟲，每次都要請求頁面，然后解析數(shù)據(jù)，然后才能請求下一個頁面。整個過程中，獲取網(wǎng)頁數(shù)據(jù)速度比較慢，那么我們就把獲取數(shù)據(jù)模塊做成并發(fā)執(zhí)行。在項目的基礎(chǔ)上，實現(xiàn)多任務(wù)并發(fā)版爬蟲。

項目github地址：github.com/NovemberCho… 回滾到相應(yīng)記錄食用，效果更佳。

首先我們把但任務(wù)版爬蟲架構(gòu)中的Fetcher模塊和Parser模塊合并成一個Worker模塊，然后并發(fā)執(zhí)行Worker模塊

然后得到并發(fā)版的架構(gòu)圖：

在并發(fā)版爬蟲中，會同時執(zhí)行多個Worker，每個Worker任務(wù)接受一個Request請求，然后請求頁面解析數(shù)據(jù)，輸出解析出的Requests和Item

因為又很多Request和Worker，所以還需要Scheduler模塊，負(fù)責(zé)對請求任務(wù)的調(diào)度處理

Engine模塊接受Worker發(fā)送的Requests和Items，當(dāng)前我們先把Items打印出，把解析出的Request發(fā)送給調(diào)度器

其中Engine和Scheduler是一個goroutine，Worker包含多個goroutine，各個模塊之間都是用channel進(jìn)行連接

先放上重構(gòu)后的項目文件結(jié)構(gòu)：

2、Worker實現(xiàn)

我們從engine.go中提取下面功能作為Worker模塊，同時把engine.go 更名為simple.go。修改后的simple.go文件請自行調(diào)整，或者去github項目源代碼回滾查看。

engine/worker.go

package engine

import (
	"crawler/fetcher"
	"log"
)

// 輸入 Request， 返回 ParseResult
func worker(request Request) (ParseResult, error) {
	log.Printf("Fetching %s
", request.Url)
	content, err := fetcher.Fetch(request.Url)
	if err != nil {
		log.Printf("Fetch error, Url: %s %v
", request.Url, err)
		return ParseResult{}, err
	}
	return request.ParseFunc(content), nil
}

對于每一個Worker接受一個請求，然后返回解析出的內(nèi)容

請大家根據(jù)架構(gòu)圖來看，效果會更好。

package engine

import "log"

// 并發(fā)引擎
type ConcurrendEngine struct {
   Scheduler   Scheduler	// 任務(wù)調(diào)度器
   WorkerCount int			// 任務(wù)并發(fā)數(shù)量
}

// 任務(wù)調(diào)度器
type Scheduler interface {
   Submit(request Request) // 提交任務(wù)
   ConfigMasterWorkerChan(chan Request)	// 配置初始請求任務(wù)
}

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

   in := make(chan Request)			// scheduler的輸入
   out := make(chan ParseResult)	// worker的輸出
   e.Scheduler.ConfigMasterWorkerChan(in)	// 把初始請求提交給scheduler

   // 創(chuàng)建 goruntine
   for i := 0; i < e.WorkerCount; i++ {
      createWorker(in, out)
   }

   // engine把請求任務(wù)提交給 Scheduler
   for _, request := range seeds {
      e.Scheduler.Submit(request)
   }

   itemCount := 0
   for {
      // 接受 Worker 的解析結(jié)果
      result := <-out
      for _, item := range result.Items {
         log.Printf("Got item: #%d: %v
", itemCount, item)
         itemCount++
      }

      // 然后把 Worker 解析出的 Request 送給 Scheduler
      for _, request := range result.Requests {
         e.Scheduler.Submit(request)
      }
   }
}

// 創(chuàng)建任務(wù)，調(diào)用worker，分發(fā)goroutine
func createWorker(in chan Request, out chan ParseResult) {
   go func() {
      for {
         request := <-in
         result, err := worker(request)
         if err != nil {
            continue
         }
         out <- result
      }
   }()
}

package scheduler

import "crawler/engine"

type SimpleScheduler struct {
	workerChan chan engine.Request
}

func (s *SimpleScheduler) Submit(request engine.Request) {
	// 為每一個 Request 創(chuàng)建 goroutine
	go func() {
		s.workerChan <- request
	}()
}

// 把初始請求發(fā)送給 Scheduler
func (s *SimpleScheduler) ConfigMasterWorkerChan(in chan engine.Request) {
	s.workerChan = in
}

package main

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

func main() {
	e := engine.ConcurrendEngine{	// 配置爬蟲引擎
		Scheduler:   &scheduler.SimpleScheduler{},
		WorkerCount: 50,
	}
	e.Run(engine.Request{		// 配置爬蟲目標(biāo)信息
		Url:       "http://www.zhenai.com/zhenghun",
		ParseFunc: parser.ParseCityList,
	})
}

本次博客我們實現(xiàn)一個最簡單的并發(fā)版爬蟲，調(diào)度器源源不斷的接受任務(wù)，一旦有一個worker空閑，就給其分配任務(wù)。這樣子有一個缺點，就是我們不知道我們分發(fā)出那么多worker的工作情況，對worker的控制力比較弱，所以在下次博客中會用隊列來實現(xiàn)任務(wù)調(diào)度。

如果想獲取Google工程師深度講解go語言視頻資源的，可以在評論區(qū)留下郵箱。

項目的源代碼已經(jīng)托管到Github上，對于各個版本都有記錄，歡迎大家查看，記得給個star，在此先謝謝大家了