HTTP状态码(记录)

HTTP状态码总的分为五类:

1开头:信息状态码

2开头:成功状态码

3开头:重定向状态码

4开头:客户端错误状态码

5开头:服务端错误状态码

 1XX:信息状态码

状态码 含义 描述
100 继续 初始的请求已经接受,请客户端继续发送剩余部分
101 切换协议 请求这要求服务器切换协议,服务器已确定切换

 2XX:成功状态码

状态码 含义 描述
200 成功 服务器已成功处理了请求
201 已创建 请求成功并且服务器创建了新的资源
202 已接受 服务器已接受请求,但尚未处理
203 非授权信息 服务器已成功处理请求,但返回的信息可能来自另一个来源
204 无内容 服务器成功处理了请求,但没有返回任何内容
205 重置内容 服务器处理成功,用户终端应重置文档视图
206 部分内容 服务器成功处理了部分GET请求

3XX:重定向状态码

状态码 含义 描述
300 多种选择 针对请求,服务器可执行多种操作
301 永久移动 请求的页面已永久跳转到新的url
302 临时移动 服务器目前从不同位置的网页响应请求,但请求仍继续使用原有位置来进行以后的请求
303 查看其他位置 请求者应当对不同的位置使用单独的GET请求来检索响应时,服务器返回此代码
304 未修改 自从上次请求后,请求的网页未修改过
305 使用代理 请求者只能使用代理访问请求的网页
307 临时重定向 服务器目前从不同位置的网页响应请求,但请求者应继续使用原有位置来进行以后的请求

4XX:客户端错误状态码

状态码 含义 描述
400 错误请求 服务器不理解请求的语法
401 未授权 请求要求用户的身份演验证
403 禁止 服务器拒绝请求
404 未找到 服务器找不到请求的页面
405 方法禁用 禁用请求中指定的方法
406 不接受 无法使用请求的内容特性响应请求的页面
407 需要代理授权 请求需要代理的身份认证
408 请求超时 服务器等候请求时发生超时
409 冲突 服务器在完成请求时发生冲突
410 已删除 客户端请求的资源已经不存在
411 需要有效长度 服务器不接受不含有效长度表头字段的请求
412 未满足前提条件 服务器未满足请求者在请求中设置的其中一个前提条件
413 请求实体过大 由于请求实体过大,服务器无法处理,因此拒绝请求
414 请求url过长 请求的url过长,服务器无法处理
415 不支持格式 服务器无法处理请求中附带媒体格式
416 范围无效 客户端请求的范围无效
417 未满足期望 服务器无法满足请求表头字段要求

 

5XX:服务端错误状态码

状态码 含义 描述
500 服务器错误 服务器内部错误,无法完成请求
501 尚未实施 服务器不具备完成请求的功能
502 错误网关 服务器作为网关或代理出现错误
503 服务不可用 服务器目前无法使用
504 网关超时 网关或代理服务器,未及时获取请求
505 不支持版本 服务器不支持请求中使用的HTTP协议版本

Go 语言 – 实战简易文章系统

go语言很简单,只要有一定的编程基础都很容易使用它来编写一些程序,学完了go lang的语法,习惯写一个小程序,这里我写了一个简易的文章系统,非常简单。

目录结构如下:

a001

1、main.go

func main()  {
   //文件系统
   //fs := http.FileSystem(http.Dir("e:/tools"))
   //http.Handle("/", http.FileServer(fs))
   //log.Fatal(http.ListenAndServe(":8080", nil))

   port := "8080"
   web := http.Server{
      Addr:           ":"+port,
      Handler:        app.HttpHandler(),
      ReadTimeout:    10 * time.Second,
      WriteTimeout:   10 * time.Second,
      MaxHeaderBytes: 1 << 20,
   }

   app.Router()

   log.Printf("Listening on port %s", port)
   log.Printf("Open http://localhost:%s in the browser", port)

   log.Fatal(web.ListenAndServe())
}

首先我们从入口类开始,main()的方法,首先实例化了一个web服务器对象,传入了port跟Handler,handler使用的是一个全局性,也就是说所有的请求都会指向app.HttpHandler()。

接着调用app.Router()方法,初始化一些router,代码待会贴上。

2、Router.go

type route struct {
   path       string
   method     string
   authorized bool
   handler    func(http.ResponseWriter, *http.Request)
}

const (
   GET    = "GET"
   POST   = "POST"
   PUT    = "PUT"
   DELETE = "DELETE"
   OPTION = "OPTION"
   HEADER = "HEADER"
)

var (
   routes map[string]route
)

func Router() {
   //http.HandleFunc("/", indexHandler)
   routes = map[string]route{}
   routes["/"] = route{path: "/", method: GET, authorized: false, handler: indexHandler}
   routes["/view/res/*"] = route{path: "/", method: GET, authorized: false, handler: resourcesHandler}
   routes["/user"] = route{path: "/user", method: GET, authorized: true, handler: indexHandler}
   routes["/add"] = route{path: "/add", method: GET, authorized: true, handler: addHandler}
   routes["/save"] = route{path: "/edit", method: POST, authorized: true, handler: addSaveHandler}
   routes["/view"] = route{path: "/view", method: GET, authorized: false, handler: viewHandler}
   routes["/sign/in"] = route{path: "/sign/up", method: GET, authorized: false, handler: signInHandler}
   routes["/sign/up"] = route{path: "/sign/up", method: GET, authorized: false, handler: signUpHandler}
   routes["/doSignIn"] = route{path: "/doSignIn", method: POST, authorized: false, handler: signInSaveHandler}
   routes["/doSignUp"] = route{path: "/doSignUp", method: POST, authorized: false, handler: signUpSaveHandler}
}

func NewRouter(key string) (r route, ok bool) {
   if strings.Contains(key, "/view/res/") {
      key = "/view/res/*"
   }
   r, err := routes[key]
   return r, err
}

router需要一个类型来保存路由的基本信息,所以这里申明一个route类型对象,route类型:

  • path string  //路由的路径
  • method string  //方法名
  • authorized bool //是否授权
  • handler func(http.ResponseWriter, *http.Request) //处理函数

3、handler.go

const (
   ERROR_NOT_FOUND    = "ERROR_NOT_FOUND"
   ERROR_NOT_METHOD   = "ERROR_NOT_METHOD"
   ERROR_AUTH_INVALID = "ERROR_AUTH_INVALID"
)

var (
   mutex sync.Mutex
   wg    sync.WaitGroup
)

func init() {
   wg.Add(100)
}

func HttpHandler() http.HandlerFunc {
   return func(w http.ResponseWriter, r *http.Request) {
      log.Println(r.Header.Get("Accept"))
      log.Println(r.Header.Get("User-Agent"))
      log.Println(r.Header)
      log.Println(r.Proto, r.Host, r.Method, r.URL.Path)

      token, _ := r.Cookie("token")
      log.Println("token", token)

      //if strings.Index(r.URL.Path,"/view/res/") == 0 {
      // resourcesHandler(w,r)
      // return
      //}
      route, ok := NewRouter(r.URL.Path)
      if !ok {
         errorHandler(w, r, ERROR_NOT_FOUND)
         return
      }
      if r.Method != route.method {
         errorHandler(w, r, ERROR_NOT_METHOD)
         return
      }
      if route.authorized && token != nil && len(token.Value) < 32 {
         errorHandler(w, r, ERROR_AUTH_INVALID)
         return
      }

      route.handler(w, r)
   }
}

func errorHandler(w http.ResponseWriter, r *http.Request, s string) {
   if s == ERROR_NOT_FOUND {
      http.NotFound(w, r)
      return
   }
   if s == ERROR_NOT_METHOD {
      http.Error(w, http.StatusText(http.StatusMethodNotAllowed), http.StatusMethodNotAllowed)
      return
   }
   http.Error(w, http.StatusText(http.StatusNonAuthoritativeInfo), http.StatusNonAuthoritativeInfo)
   return
}

func resourcesHandler(w http.ResponseWriter, r *http.Request) {
   filePath := conf.ROOT + string([]byte(r.URL.Path)[1:])

   //file, err := os.OpenFile(filePath, os.O_RDONLY, 066)
   //defer file.Close()
   //if err != nil {
   // fmt.Println("not found", filePath)
   // return
   //}
   http.ServeFile(w, r, filePath)
}

这里没有用到WaitGroup,只是申明的时候忘记删除了。

主要的函数HttpHandler(),这时一个公共函数,类似于http的调度者,所有的请求都会call这个函数,然后再通过这个函数去分配控制器(route.handler(w, r))。

资源文件处理函数resourcesHandler(),这个函数是将go http服务器中的js、css、image等这些静态资源直接输出,开始不知道有http.ServeFile(w, r, string)这个函数,所以使用了最基本的os读取文件的方式把文件输出出去,其实如果全心投入到go语言,那么真的需要很好地去了解一下go语言的SDK。

3、Controller.go

func indexHandler(w http.ResponseWriter, r *http.Request) {
   util.Output(w, tmpl.Index(r), util.PUT_HTML)
}

这里我只展示了一个函数,其余的函数都是一样的,这里使用了工具类,把信息输出给用户,其中信息的处理交给了tmpl.go的文件。

4、tmpl.go

package tmpl

import (
   "book/model"
   "book/util"
   "fmt"
   "net/http"
   "strconv"
   "strings"
)

func init() {

}

func Index(r *http.Request) string {
   //return "Hello, World!"
   h := NewHtml();
   //h.body("<h1>Hello, World</h1>")
   //h.body(util.GetViewTpl("index.html"))

   list := model.GetArticles("select * from lx_article order by id desc")
   var str string
   tml := `
      <div class="row">
        <div class="col-left">
         <img src="/view/res/img/file_101.353.png" class="img128"/>
            <a href="/view?id=%d" target="_blank">%s</a>
      </div>
        <div class="col-right">%s</div>
        <div class="col-right1"><a href="#">%s</a></div>
    </div>`
   for _, s := range list {
      str += fmt.Sprintf(tml, s.Id, s.Title, s.CreateTimeF, s.User.Username)
   }
   h.body(strings.Replace(util.GetViewTpl("index.html"), "{{content}}", str, -1))
   return h.Show("首页")
}

这个文件比较负责,设计了html的代码,我没有时间去编写模板引擎,所以使用了比较简单的字符替换的方式,把模板输出出去,其实在生产环境中,我们很有必要编写一个模板引起,不过现在流行的是前后端分离,所有的请求,都是通过接口的形式去调去,那么在实际应用中这一层是用不上的,但是为了实现一个简易的文章系统,这里我还是编写出这样不人性化的代码。

核心代码还有很多比如:数据库、模型等到,这里不一一贴出,帖子的最后会附上整个项目的源代码地址,现在我们来看看截图:

用户登录

003

 

 

发表文章:

QQ截图20190507095119

首页的效果图:

001

 

查看文章:

002

项目地址:

https://github.com/AlanRo1986/go-book

Go 语言入门总结

Go 语言使用非常简单,是专门针对各种语言的痛点设计的,本人看的书是《Go语言实战》非常推荐给大家,笔记的作者是原水寒,我在查看他的笔记的时候觉得他总结的很到位,所以就拷贝了一份到本人的博客中,具体目录请参见原作者 Go 语言极速入门

实际上,还有一些关于 Go 语言的知识没有在极速入门中进行分析,例如 Go 语言的反射机制、Go 语言的测试体系、各种标准库的使用以及各种 Go 语言内建工具的使用。当然,Go 语言的表格驱动测试姿势在 Go 语言极速入门12 – 实战项目之单任务版爬虫 这一小节中做过简单的使用姿势的分析,但是性能测试没有进行分析。
使用 Go 语言实现的项目有以下这些
Docker
Kubernetes
Consul
Prometheus
etcd
istio
SOFAMosn
SOFAMesh
opentracing-go
grpc-go

Go 语言的基本使用姿势掌握后,后续我会去分析 SOFAStack 中的两个框架 – SOFAMosn 和 SOFAMesh,这两个框架由蚂蚁金服开发维护,前者是 ServiceMesh 架构下很好的 SideCar 实现,后者是 fork 自 istio,并作出了一些自己的优化,通过对这两个框架的分析,我们可以不只是从理论方面学习 ServiceMesh 架构,还可以从实际的代码实现来学习 ServiceMesh!!!

Go 语言入门14 – jsonrpc 最简姿势

服务定义

package rpcdemo

import "errors"

// 服务
type DemoService struct {
}

// 参数
type Args struct {
   A, B int
}

// jsonrpc 的服务需要定义为参入参数和传出参数的格式
// 传出参数必须为指针格式
func (e DemoService) DIV(args Args, result *float64) error {
   if args.B == 0 {
      return errors.New("division by zero")
   }

   *result = float64(args.A) / float64(args.B)
   return nil
}

 

服务端

package main

import (
   "github.com/zhaojigang/crawler/rpc"
   "log"
   "net"
   "net/rpc"
   "net/rpc/jsonrpc"
)

func main() {
   // 注册服务
   rpc.Register(rpcdemo.DemoService{})
   // 启动 server
   listener, err := net.Listen("tcp", "127.0.0.1:1234")

   if err != nil {
      panic(err)
   }

   for {
      // 不断连接服务
      conn, err := listener.Accept()
      if err != nil {
         log.Printf("accept error, %v", err)
         continue
      }
      // 使用 Goroutine:ServeConn runs the JSON-RPC server on a single connection.
      go jsonrpc.ServeConn(conn)
   }
}

客户端

package main

import (
   "fmt"
   "github.com/zhaojigang/crawler/rpc"
   "log"
   "net/rpc/jsonrpc"
)

func main() {
   // 连接server并创建jsonrpc-client
   client, err := jsonrpc.Dial("tcp", "127.0.0.1:1234")
   if err != nil {
      log.Printf("create jsonrpc client error")
   }

   var result float64
   // 发起调用
   err = client.Call("DemoService.DIV", rpcdemo.Args{3, 4}, &result)
   if err != nil {
      log.Printf("call error")
   }
   fmt.Println(result)
}

Go 语言入门13 – 实战项目之并发版爬虫

单任务版的爬虫很慢,因为只有一个 main Goroutine 在执行,最慢的地方就是爬取(fetch)和解析(parse),我们可以将同一个 url 的这两部分合并成一个任务(worker),然后使用多个 Goroutine 去并行的执行这多个 worker,这样速度就会有极大的提升。

一、并发版爬虫 – 并发调度器

1.1、架构

1556594489-4817-5842684-9885ac4881bc3a58

  1. 服务启动时,创建两个 chan,in(用于接收 Request)和 out(用于接收 ParseResult),并且将 in 赋值给 scheduler 的任务 chan(即 scheduler 后续操作的 chan 其实就是 in)
  2. 创建指定数量个 Goroutine,每一个 Goroutine 做以下几件事:

2.1. 从 in chan 中阻塞等待获取 Request
2.2. 使用 Worker 对 获取到的Request 做 fetch 和 parse 的操作,将 parseResult 阻塞发送到 out chan

  1. Engine 将 seeds 中的 Request 添加到调度器 chann,
  2. 之后开启死循环,不断从 out chan 中接收 parseResult,然后将 parseResult.items 进行打印,将 parseResult.requests 加入到 scheduler 的 chan,实际上就是 in chan(当然,每一个 Request 加入 in 的操作都是由一个新的 Goroutine 来完成的,原因为本小节末会讲)

1.2、实现

1556594496-2334-5842684-318244d050f83249

由于项目结构进行了调整,列出发生修改的所有代码。

爬取器 fetcher 和解析器 parser 与之前相同,模型类也不变。

调度器接口 scheduler.go

package scheduler

import (
    "github.com/zhaojigang/crawler/model"
)

// 调度器接口
type Scheduler interface {
    // 提交 Request 到调度器的 request 任务通道中
    Submit(request model.Request)
    // 初始化当前的调度器实例的 request 任务通道
    ConfigureMasterWorkerChan(chan model.Request)
}

并发调度器 simple.go

package scheduler

import (
    "github.com/zhaojigang/crawler/model"
)

type SimpleScheduler struct {
    workerChan chan model.Request
}

// 为什么使用指针接收者,需要改变 SimpleScheduler 内部的 workerChan
// https://stackoverflow.com/questions/27775376/value-receiver-vs-pointer-receiver-in-golang
// https://studygolang.com/articles/1113
// https://blog.csdn.net/suiban7403/article/details/78899671
func (s *SimpleScheduler) ConfigureMasterWorkerChan(in chan model.Request) {
    s.workerChan = in
}

func (s *SimpleScheduler) Submit(request model.Request) {
    // 每个 Request 一个 Goroutine
    go func() { s.workerChan <- request }()
}

注意:

  • 这里为每一个将 Request 添加到 chan 的任务都开启一个 Goroutine 来执行,为什么?
  • 方法为何使用指针接收者而不是值接收者?
package engine

import (
    "github.com/zhaojigang/crawler/fetcher"
    "github.com/zhaojigang/crawler/model"
    "github.com/zhaojigang/crawler/scheduler"
    "log"
)

// 并发引擎
type ConcurrentEngine struct {
    // 调度器
    Scheduler scheduler.Scheduler
    // 开启的 worker 数量
    WorkerCount int
}

func (e ConcurrentEngine) Run(seeds ...model.Request) {
    in := make(chan model.Request)
    out := make(chan model.ParseResult)
    // 初始化调度器的 chann
    e.Scheduler.ConfigureMasterWorkerChan(in)
    // 创建 WorkerCount 个 worker
    for i := 0; i < e.WorkerCount; i++ {
        createWorker(in, out);
    }
    // 将 seeds 中的 Request 添加到调度器 chann
    for _, r := range seeds {
        e.Scheduler.Submit(r)
    }

    for {
        result := <-out // 阻塞获取
        for _, item := range result.Items {
            log.Printf("getItems, items: %v", item)
        }

        for _, r := range result.Requests {
            // 如果 submit 内部直接是 s.workerChan <- request,则阻塞等待发送,该方法阻塞在这里
            // 如果 submit 内部直接是 go func() { s.workerChan <- request }(),则为每个Request分配了一个Goroutine,这里不会阻塞在这里
            e.Scheduler.Submit(r)
        }
    }
}

func createWorker(in chan model.Request, out chan model.ParseResult) {
    go func() {
        for {
            r := <-in // 阻塞等待获取
            result, err := worker(r)
            if err != nil {
                continue
            }
            out <- result // 阻塞发送
        }
    }()
}

func worker(r model.Request) (model.ParseResult, error) {
    log.Printf("fetching url:%s", r.Url)
    body, err := fetcher.Fetch(r.Url)
    if err != nil {
        log.Printf("fetch error, url: %s, err: %v", r.Url, err)
        return model.ParseResult{}, nil
    }
    return r.ParserFunc(body), nil
}

启动器 main.go

package main

import (
    "github.com/zhaojigang/crawler/engine"
    "github.com/zhaojigang/crawler/model"
    "github.com/zhaojigang/crawler/scheduler"
    "github.com/zhaojigang/crawler/zhenai/parser"
)

func main() {
    engine.ConcurrentEngine{
        Scheduler:   &scheduler.SimpleScheduler{},
        WorkerCount: 1000,
    }.Run(model.Request{
        // 种子 Url
        Url:        "http://www.zhenai.com/zhenghun",
        ParserFunc: parser.ParseCityList,
    })
}

疑问:

Q1. 为什么在 scheduler 中每一个将 Request 添加到 chan 的任务都开启一个 Go routine 来执行?

A:在 Go 语言学习9 – Channel 一节描述过,对于无缓冲的 channel,如果两个 go routine 没有同时准备好,通道会导致先执行发送或接收操作的 go routine 阻塞等待,假设使用 s.workerChan <- request 而不是 go func() { s.workerChan <- request }(),假设开启了 10 个 Worker Go routine,这 10 个 go routine 阻塞在 r := <-in 阻塞等待获取 Request 上,假设 seeds 大于 10,例如 11,那么当 Engine 的这个循环执行到底 11 个的时候,将陷入等待,因为所有的10个 Worker go routine 此时都可能也处于等待中,即 in chan 没有接收方准备好接收数据,所以 engine 作为发送方也要阻塞等待;那么为什么10个 Worker go routine 都会处于等待中呢?

for _, r := range seeds {
        e.Scheduler.Submit(r)
}

 

 go func() {
        for {
            r := <-in // 阻塞等待获取
            result, err := worker(r)
            if err != nil {
                continue
            }
            out <- result // 阻塞发送
        }
    }()

Q2. scheduler 方法为何使用指针接收者而不是值接收者?

A:在 Go 语言学习5 – 面向接口 中我们详细的介绍了什么时候使用指针接收者,什么时候使用值接收者,其中最重要的两条就是 “1. 如果要改变接收者内部的属性值,必须使用指针接收者,因为值接收者是对接收者副本的操作;2. 如果 struct 内一个方法是指针接收者,那么其全部方法都是用指针接收者”,在 scheduler 中,我们要将外界的 in chan 赋值给 scheduler 的 workChann,所以需要改变 workChann 的值,需要使用指针接收者。

二、并发版爬虫架构 – 队列调度器

1.1、架构

1556594510-9572-5842684-bb0118a739fafe94

服务启动时,初始化 Scheduler 的 requestChann(用于接收 Request)和 workerChan(用于接收 Worker 的 chan Request,注意:每一个 Worker 都有一个 chan Request),然后 Scheduler 启动一个 go routine,不断的将 requestChann 接收到的 Request 存储到 requestQ 切片中,将 workerChan 接收到的 chan Request 存储到 workerQ 切片中,并将 requestQ 中的 Request 分配给 workerQ 中的 chan Request(即由某一个 Worker 来处理)。
注意:在 Scheduler 中有两个 Go routine。

一个是主 Go routine,用于接收 Engine 传来的 Request 到 requestChann 以及接收 Worker 传来的 chan Request 到 workerChan 中
另一个是 Scheduler 显示启动的一个 Go routine,用于将 requestChann 接收到的 Request 存储到 requestQ 切片中,将 workerChan 接收到的 chan Request 存储到 workerQ 切片中,并将 requestQ 中的 Request 分配给 workerQ 中的 chan Request

创建指定数量个 Go routine,每一个 Go routine 做以下几件事:

  • 1. 创建 Worker 自己的 w chan Request,并发送到 Scheduler 的 workerChan 中
  • 2. 从 w 中阻塞等待获取 Request(如果 Scheduler 的分配 Go routine 分配了 Request 到该 Worker 的 w,则获取成功)
  • 3. 使用 Worker 对 获取到的 Request 做 fetch 和 parse 的操作,将 parseResult 阻塞发送到 out chan

Engine 将 seeds 中的 Request 添加到 Scheduler 的 requestChann
之后开启死循环,不断从 out chan 中接收 parseResult,然后将 parseResult.items 进行打印,将 parseResult.requests 加入到 scheduler 的 requestChann

1.2、实现

1556594517-3253-5842684-ef18b36002d77a36

调度器 scheduler.go

package scheduler

import (
    "github.com/zhaojigang/crawler/model"
)

// 调度器接口
type Scheduler interface {
    ReadyNotifier
    Submit(request model.Request)
    WorkerChann() chan model.Request
    Run()
}

通知器 readyNotifier.go

package scheduler

import "github.com/zhaojigang/crawler/model"

type ReadyNotifier interface {
    WorkerReady(chan model.Request)
}

队列调度器 queued.go

package scheduler

import (
   "github.com/zhaojigang/crawler/model"
)

type QueuedScheduler struct {
   requestChann chan model.Request
   // 每一个Worker都有一个自己的chan Request
   // workerChan中存放的是Worker们的chan
   workerChan chan chan model.Request
}

func (s *QueuedScheduler) WorkerChann() chan model.Request {
   return make(chan model.Request)
}

func (s *QueuedScheduler) Submit(request model.Request) {
   s.requestChann <- request
}

func (s *QueuedScheduler) WorkerReady(w chan model.Request) {
   s.workerChan <- w
}

func (s *QueuedScheduler) Run() {
   // 初始化 requestChann
   s.requestChann = make(chan model.Request)
   // 初始化 workerChan
   s.workerChan = make(chan chan model.Request)

   // 创建一个 goroutine
   // 1. 进行request以及Worker的chan的存储
   // 2. 分发request到worker的chan中
   go func() {
      var requestQ []model.Request
      var workerQ []chan model.Request
      for {
         var activeRequest model.Request
         var activeWorker chan model.Request
         if len(requestQ) > 0 && len(workerQ) > 0 {
            activeRequest = requestQ[0]
            activeWorker = workerQ[0]
         }

         select {
         case r := <-s.requestChann:
            // 如果开始requestQ=nil,append之后就是包含一个r元素的切片
            requestQ = append(requestQ, r)
         case w := <-s.workerChan:
            workerQ = append(workerQ, w)
            // 进行request的分发
         case activeWorker <- activeRequest:
            requestQ = requestQ[1:]
            workerQ = workerQ[1:]
         }
      }
   }()
}

并发引擎 ConcurrentEngine.go

package engine

import (
   "github.com/zhaojigang/crawler/fetcher"
   "github.com/zhaojigang/crawler/model"
   "github.com/zhaojigang/crawler/scheduler"
   "log"
)

// 并发引擎
type ConcurrentEngine struct {
   // 调度器
   Scheduler scheduler.Scheduler
   // 开启的 worker 数量
   WorkerCount int
}

func (e ConcurrentEngine) Run(seeds ...model.Request) {
   // 初始化 Scheduler 的队列,并启动配对 goroutine
   e.Scheduler.Run()
   out := make(chan model.ParseResult)
   for i := 0; i < e.WorkerCount; i++ {
      // 每个 Worker 都创建自己的一个 chan Request
      createWorker(e.Scheduler.WorkerChann(), out, e.Scheduler);
   }
   for _, r := range seeds {
      e.Scheduler.Submit(r)
   }

   for {
      result := <-out // 阻塞获取
      for _, item := range result.Items {
         log.Printf("getItems, items: %v", item)
      }

      for _, r := range result.Requests {
         e.Scheduler.Submit(r)
      }
   }
}

func createWorker(in chan model.Request, out chan model.ParseResult, notifier scheduler.ReadyNotifier) {
   go func() {
      for {
         notifier.WorkerReady(in)
         r := <-in // 阻塞等待获取
         result, err := worker(r)
         if err != nil {
            continue
         }
         out <- result // 阻塞发送
      }
   }()
}

func worker(r model.Request) (model.ParseResult, error) {
   log.Printf("fetching url:%s", r.Url)
   body, err := fetcher.Fetch(r.Url)
   if err != nil {
      log.Printf("fetch error, url: %s, err: %v", r.Url, err)
      return model.ParseResult{}, nil
   }
   return r.ParserFunc(body), nil
}

启动器 main.go

package main

import (
    "github.com/zhaojigang/crawler/engine"
    "github.com/zhaojigang/crawler/model"
    "github.com/zhaojigang/crawler/scheduler"
    "github.com/zhaojigang/crawler/zhenai/parser"
)

func main() {
    engine.ConcurrentEngine{
        Scheduler:   &scheduler.QueuedScheduler{},
        WorkerCount: 100,
    }.Run(model.Request{
        // 种子 Url
        Url:        "http://www.zhenai.com/zhenghun",
        ParserFunc: parser.ParseCityList,
    })
}

队列调度器与并发调度器的对比

  • 队列调度器不需要为每一个 “将 Request 添加到 chan的任务” 创建一个 Go routine 来执行
  • 队列调度器需要为每一个 Worker 创建一个 chan Request
  • 队列调度器编码较为复杂,并发调度器编码简单,且虽然会创建大量的
  • Go routine,但是由于协程的轻量性,一般而言,问题不大