1. 首页
  2. 文章

Golang实现简单爬虫框架（5）——项目重构与数据存储

盐的甜 · · 1399 次点击 · · 开始浏览    
这是一个创建于 的文章，其中的信息可能已经有所发展或是发生改变。
第一次，站长亲自招 Gopher 了>>>

前言

在上一篇文章《Golang实现简单爬虫框架（4）——队列实现并发任务调度》中，我们使用用队列实现了任务调度，接下来首先对两种并发方式做一个同构，使代码统一。然后添加数据存储模块。

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

1、项目重构

（1）并发引擎

通过分析我们发现，两种不同调度的区别是每个worker一个channel还是 所有worker共用一个channel，所以我们在接口中定义一个函数WorkerChan()，用来决定这件事，即worker一个channel还是 所有worker共用一个channel。此时ConfigMasterWorkerChan就不再需要了。

在项目文件concurrent.go中我们定义一个任务调度器Scheduler，如下：

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

但是在简单并发中我们只实现了Submit和ConfigMasterWorkerChan接口，而使用队列调度中却实现了接口的所有方法，所有我们同构一下使concurrent.go文件可以适用于两种不同的调度器。

因为在createworker函数中要使用WorkerReady函数，所以要传入一个Scheduler，但是这样显得比较重，我们可以利用接口组合，新建一个接口ReadyNotifier，这样在createworker函数中传入ReadyNotifier即可。

修改后的任务调度如下：

type Scheduler interface {
	ReadyNotifier
	Submit(request Request) // 提交任务
	WorkerChan() chan Request
	Run()
}
type ReadyNotifier interface {
	WorkerReady(chan Request)
}
复制代码

此时创建goroutine修改如下：

// 创建 goroutine
for i := 0; i < e.WorkerCount; i++ {
    //任务是每个 worker 一个 channel 还是 所有 worker 共用一个 channel 由WorkerChan 来决定
	createWorker(e.Scheduler.WorkerChan(), out, e.Scheduler)
}
复制代码

修改后的concurrent.go文件如下：

package engine

import (
	"log"
)

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

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

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

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

	// 创建 goruntine
	for i := 0; i < e.WorkerCount; i++ {
		// 任务是每个 worker 一个 channel 还是 所有 worker 共用一个 channel 由WorkerChan 来决定
		createWorker(e.Scheduler.WorkerChan(), 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(in chan Request, out chan ParseResult, ready ReadyNotifier) {
	go func() {
		for {
			ready.WorkerReady(in) // 告诉调度器任务空闲
			request := <-in
			result, err := worker(request)
			if err != nil {
				continue
			}
			out <- result
		}
	}()
}
复制代码

（2）简单并发调度器

scheduler/simple.go

package scheduler

import "crawler/engine"

type SimpleScheduler struct {
	workerChan chan engine.Request
}

func (s *SimpleScheduler) WorkerChan() chan engine.Request {
	// 此时所有 worker 共用同一个 channel，直接返回即可
	return s.workerChan
}

func (s *SimpleScheduler) WorkerReady(w chan engine.Request) {

}

func (s *SimpleScheduler) Run() {
    // 创建出 workchannel
	s.workerChan = make(chan engine.Request)
}

func (s *SimpleScheduler) Submit(request engine.Request) {
	// send request down to worker chan
	go func() {
		s.workerChan <- request
	}()
}
复制代码

（3）队列实现调度器

scheduler/queued.go

添加WorkerChan()的实现即可

package scheduler

import "crawler/engine"

// 使用队列来调度任务

type QueuedScheduler struct {
	requestChan chan engine.Request
	workerChan  chan chan engine.Request
}

func (s *QueuedScheduler) WorkerChan() chan engine.Request {
	// 对于队列实现来讲，每个 worker 共用一个 channel
	return make(chan engine.Request)
}

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

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

func (s *QueuedScheduler) Run() {
	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

			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:]
			}
		}
	}()
}
复制代码

（4）main函数

经过上述同构，在main函数中如需切换不同调度器，只需要相应的配置即可。

package main

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

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

2、数据存储

（1）Mgo的介绍安装

爬取到的数据不能仅仅在控制台打印出来，所以我们还要给爬虫添加数据存储模块。我们本次选择使用mongodb来存储我们的数据。

mgo（音mango）是MongoDB的Go语言驱动，它用基于Go语法的简单API实现了丰富的特性，并经过良好测试。

官方网址：labix.org/mgo

文档：API docs for mgo

首先我们要安装mgo，打开终端，输入下面代码完成安装

go get gopkg.in/mgo.v2
复制代码

mgo基本操作都很简单，有数据库操作经验都可以很快上手。

（2）爬虫引擎与数据格式

首先，爬虫引擎获取到数据要把数据发送给数据存储模块，而数据的传递用要用到channel，所以打开concurrent.go文件，在引擎添加ItemChan属性，如下所示：

爬取到数据需要把数据发送到数据存储模块，

package engine
// 并发引擎
type ConcurrendEngine struct {
	Scheduler   Scheduler // 任务调度器
	WorkerCount int       // 并发任务数量
	ItemChan    chan Item // 数据保存 channel
}

// ...
for {
    // 接受 Worker 的解析结果
    result := <-out
    for _, item := range result.Items {
        // 当抓取一组数据后，进行保存
        go func(item2 Item) {
			e.ItemChan <- item2
		}(item)
    }
	// ...
}
// ...
复制代码

在engine/types.go中定义Item类型：

package engine

// 请求结构
type Request struct {
	Url       string // 请求地址
	ParseFunc func([]byte) ParseResult
}

// 解析结果结构
type ParseResult struct {
	Requests []Request // 解析出的请求
	Items    []Item    // 解析出的内容
}

// 解析出的用户数据格式
type Item struct {
	Url     string      // 个人信息Url地址
	Type    string      // table
	Id      string      // Id
	Payload interface{} // 详细信息
}

func NilParseFun([]byte) ParseResult {
	return ParseResult{}
}
复制代码

（3）存储模块的实现

在根目录下创建persist文件夹，然后创建itemsaver.go文件

// persist/itemsaver.go
package persist

import (
	"context"
	"crawler/engine"
	"errors"
	"gopkg.in/mgo.v2"
	"gopkg.in/olivere/elastic.v5"
	"log"
)

func ItemSaver(index string) (chan engine.Item, error) {
	// mongodb connect
	session, err := mgo.Dial("localhost:27017")
	if err != nil {
		panic(err)
	}

	out := make(chan engine.Item)
	go func() {
		itemCount := 0
		for {
			// 接收到发送的 item
			item := <-out
			log.Printf("Item Saver: got item #%d: %v\n",
				itemCount, item)
			itemCount++

			// Save data in mongodb
			err := mongo_save(session, index, item)

			if err != nil {
				// if have err, ignore it
				log.Printf("Item Saver: error, saving item %v: %v",
					item, err)
			}
		}
	}()
	return out, nil
}

// 使用 MongoDB 保存数据
func mongo_save(session *mgo.Session, dbName string, item engine.Item) error {
	if item.Type == "" {
		return errors.New("must supply Type")
	}
	c := session.DB(dbName).C(item.Type)	// 选择要操作的数据库与集合
	err := c.Insert(item)		// 插入数据
	if err != nil {
		log.Fatal(err)
	}
	return nil
}
复制代码

（4）存储测试文件

我们把一条数据存入mongodb，然后再取出来，比对读出的数据和写入的数据是否相同

// persist/itemsaver_test.gp
package persist

import (
	"crawler/engine"
	"crawler/model"
	"encoding/json"
	"fmt"
	"gopkg.in/mgo.v2"
	"gopkg.in/mgo.v2/bson"
	"log"
	"testing"
)
func TestMongoSave(t *testing.T) {
	// mongodb connect
	session, err := mgo.Dial("localhost:27017")
	if err != nil {
		panic(err)
	}

	expected := engine.Item{
		Url:  "http://album.zhenai.com/u/1946858930",
		Type: "zhenai",
		Id:   "1946858930",
		Payload: model.Profile{
			Name:     "為你垨候",
			Gender:   "女士",
			Age:      40,
			Height:   163,
			Weight:   54,
			Income:   "5-8千",
			Marriage: "未婚",
			Address:  "佛山顺德区",
		},
	}
	// 保存数据
	err = mongo_save(session, "crawler", expected)
	if err != nil {
		panic(err)
	}

	c := session.DB("crawler").C("zhenai")

	var result engine.Item
    // 查询数据
	err = c.Find(bson.M{"id": "1946858930"}).One(&result)
	// result 为 Json 类型
	if err != nil {
		log.Fatal(err)
	}
	fmt.Printf("%s, %s, %v\n", result.Url, result.Id, result.Payload)
}
复制代码

（5）parser模块

我们要在parse/profile.go文件中组装好需要保存到数据库的数据格式

// ...
result := engine.ParseResult{
    Items: []engine.Item{
        {
            Url:     url,
            Type:    "zhenai",
            Id:      extractString([]byte(url), idUrlRe),
            Payload: profile,
        },
    },
}
// ...
复制代码

（6）main函数

package main

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

func main() {
	itemChan, err := persist.ItemSaver()
	if err != nil {
		panic(err)
	}

	e := engine.ConcurrendEngine{
		//Scheduler: &scheduler.QueuedScheduler{},
		Scheduler:   &scheduler.SimpleScheduler{},
		WorkerCount: 100,
		ItemChan:    itemChan,
	}
	e.Run(engine.Request{
		Url:       "http://www.zhenai.com/zhenghun",
		ParseFunc: parser.ParseCityList,
	})
}
复制代码

运行项目，打开mongodb可视化工具，可以看到爬取了54410条数据

3、总结

我们首先把两种并发方式做一个同构，使代码统一，直接在main函数中使用不同的配置就可以切换调度器，简单方便。然后使用Mgo驱动操作数据，添加到mongodb中。内容有点多，很多代码没有完整的展示出来，希望大家可以下载项目源代码，回滚到对应提交记录查看，效果会更好。 别无所求，只求随手给个star

下篇博客中我们会再当前博客的基础上添加数据展示功能

如果想获取Google工程师深度讲解go语言视频资源的，可以在评论区留下邮箱。

如果觉得文章还可以，劳烦大人随手点个赞。。。

有疑问加站长微信联系（非本文作者）

本文来自：掘金

感谢作者：盐的甜

查看原文：Golang实现简单爬虫框架（5）——项目重构与数据存储

入群交流（和以上内容无关）：加入Go大咖交流群，或添加微信：liuxiaoyan-s 备注：入群；或加QQ群：692541889

1399 次点击  
加入收藏 微博
赞
收入我的专栏
上一篇：切片 - Go 语言学习笔记
下一篇：go程序基于阿里云CodePipeline的一次devops实践
  • 代码
  • mgo
  • channel
  • 任务调度
    0 回复
    暂无回复
    添加一条新回复 （您需要 登录 后才能回复 没有账号 ？）
    • 请尽量让自己的回复能够对别人有帮助
    • 支持 Markdown 格式, **粗体**、~~删除线~~、`单行代码`
    • 支持 @ 本站用户；支持表情（输入 : 提示），见 Emoji cheat sheet
    • 图片支持拖拽、截图粘贴等方式上传

    用户登录

    Go今日面试题

     今日阅读排行

      加载中

     一周阅读排行

      加载中

      关注我

    • 扫码关注领全套学习资料 关注微信公众号
    • 加入 QQ 群：
      • 192706294（已满）
      • 731990104（已满）
      • 798786647（已满）
      • 729884609（已满）
      • 977810755（已满）
      • 815126783（已满）
      • 812540095（已满）
      • 1006366459（已满）
      • 692541889

    • 关注微信公众号
    • 加入微信群：liuxiaoyan-s，备注入群
    • 也欢迎加入知识星球 Go粉丝们（免费）