第十六章分布式爬虫--准备工作

分布式消息传递的方式.

REST
RPC
消息队列

都在什么情况下使用这三种方式呢?

1. 客户端和主服务器之间, 使用的是REST请求方式

2. 主服务器和其他子服务器之间通信,比如接口调用, 可以使用RPC

3. 服务器和服务器之间消息传递可以是用消息队列

对外: 使用REST

模块内部:使用RPC

模块之间: 使用中间件或REST

分布式架构 VS 微服务架构

看上面的三点:

1. 微服务架构: 是按照业务模块来划分的

2. 分布式架构: 每个业务模块部署多个节点, 同一个模块之间节点是如何通信的. 不同模块之间节点是如何通信的

3. 微服务架构是基础, 知道我们项目如何拆分, 分布式架构是实现. 二者是结合使用的.

并发版爬虫的架构

这是上一章最后做成了并发版爬虫的架构.

思考:为什么需要转成分布式架构? 并发版架构不可以么?

这个问题, 还需要从实际出发, 我们遇到了什么样的问题. 针对这些问题, 我们来思考解决方案

并发版爬虫. 我们遇到了哪些问题呢?

1. 限流: 单个节点获取流量的速度是有限的. 珍爱网限制了我们爬取的速度, 如果想要爬取资源, 就必须限速在其以下, 否则就会被拦截

2. 存储问题: 存储部分的结构, 技术栈和爬虫差别很大. 如果想要在存储上进一步优化, 就需要特殊ElasticSearch技术背景的人. 而这部分人可能对爬虫的技术架构不是特别关心. 为了方便分工协作, 我们把存储模块单独提取出来. 这部分呢叫做固有分布式, 也就是说,通常我们都按照这个进行划分的.

3. 去重问题: 我们要过滤抓取到的同一个用户的多次入库, 如果去重逻辑很复杂,这一块也会很耗时. 所以也需要提取出来单独处理.

分布式爬虫的架构

每一个方框都是一个节点

worker: 处理fetch速度慢的问题, 作为一个单独的模块. 并且部署成多个节点, 同时去并发抓取网页数据

存储问题: 保存入库也是比较浪费时间, 浪费性能的, 叶提取出来作为一个单独的服务

我们下面就来实现这样一个分布式. 我们使用docker来实现.

正好可以看看docker是如何具有良好的扩展性, 如何收,如何放的.

从Channel到分布式

并发版爬虫到分布式爬虫转换, 他的关键在哪里呢? 关键在于从channel到分布式

并发版爬虫有很多goroutine , goroutine之间通过channel进行通信. 现在我们要做的就是将使用channel进行通信的节点, 换一种机制.

RPC都有哪些

jsonRpc
GRPC
Thrift

本次我们使用jsonRpc来实现

什么是RPC?

RPC（Remote Procedure Call）是远程过程调用，它是一种通过网络从远程计算机程序上请求服务，而不需要了解底层网络技术的协议。

RPC协议假定某些传输协议的存在，如TCP或UDP，为通信程序之间携带信息数据。在OSI网络通信模型中，RPC跨越了传输层和应用层。RPC使得开发包括网络分布式多程序在内的应用程序更加容易。

RPC采用客户机/服务器模式。请求程序就是一个客户机，而服务提供程序就是一个服务器。首先，客户机调用进程发送一个有进程参数的调用信息到服务进程，然后等待应答信息。

接下来我们模拟一个rpc调用

go简单模拟RPC实现

1. 实现服务端的业务代码

代码语言：javascript复制

package rpc

import "errors"

// 首先模拟一个服务
type DemoService struct {

}

type Args struct {
    A, B int
}

// 我们来做一个除法
// rpc调用要求有两个参数
func (DemoService) Div(args Args, result *float64) error {
    if args.B == 0 {
        return  errors.New("除数不能为0")
    }

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

这里就做了一个除法
共rpc调用的方法, 必须要有两个参数, 一个是入参, 一个是返回值

2. 服务端模拟

代码语言：javascript复制

package main

import (
    rpc2 "aaa/rpc"
    "github.com/kelseyhightower/confd/log"
    "net"
    "net/rpc"
    "net/rpc/jsonrpc"
)

// 下面将模拟rpc的server, 将服务端调用封装起来
func main() {
    // 第一步: 把我们写好的DemoService注册到rpc上
    rpc.Register(rpc2.DemoService{})
    // 第二步: 开服务, 服务的端口是1234
    listener, e := net.Listen("tcp", ":1234")
    if e != nil {
        panic(e)
    }

    for  {
        // 开始等待客户端连接进来
        conn, e := listener.Accept()
        if e != nil {
            log.Info("客户端连接异常: %v", e)
        }
        // 运行一个jsonrpc服务, 这里来了goroutine, 这样就不用等待处理完成了,
        // 下一个连接来了, 就可以直接连进来
        go jsonrpc.ServeConn(conn)
    }
}

将写好的服务功能注册到服务端
定义监听服务的端口和协议
和客户端建立连接, 等待客户端连接

3. 客户端模拟

代码语言：javascript复制

package main

import (
    rpc2 "aaa/rpc"
    "fmt"
    "net"
    "net/rpc/jsonrpc"
)

func main() {

    // 拨号, 和端口号为1234的tcp连接
    conn, e := net.Dial("tcp", ":1234")
    if e != nil {
        panic(e)
    }

    // 建立连接
    client := jsonrpc.NewClient(conn)

    // 发送客户端数据
    var result float64
    e = client.Call("DemoService.Div", rpc2.Args{10, 5}, &result)
    if e != nil {
        fmt.Printf("错误信息: %v n", e)
    } else {
        fmt.Printf("result: %f n", result)
    }

    e = client.Call("DemoService.Div", rpc2.Args{3, 0}, &result)
    if e != nil {
        fmt.Printf("错误信息: %v n", e)
    } else {
        fmt.Printf("result: %f n", result)
    }

}

拨号, 向端口号为1234的服务端拨号
建立连接
发送客户端消息

服务端收到消息的处理结果

代码语言：javascript复制

result: 2.000000 
错误信息: 除数不能为0 

Process finished with exit code 0

下面将并发版爬虫, 改为分布式爬虫.

2. 将worker也改为使用rpc进行通信

第一部分, ItemSaver我们已经顺利的使用rpc进行通信了

接下来我们对第二部分worker, 也提取成rpc

我们来看看worker的实现

入参是一个Request

返回值是: ParseResult,error

Request的如下:

代码语言：javascript复制

type Request struct {
    Url string
    ParseFun func(content []byte) ParseResult
}

type ParseResult struct {
    Req []Request
    Items persist.Item
}

Request结构体有两个参数, 一个是函数, 一个是Url

Url是一个字符串, 可以在网络上传输, 网络上传输的都是字符串,整数, 能够翻译成json的报文. ParseFunc是一个函数, 函数是不能直接在网络上传输的, 于是, 我们要将函数进行序列化, 然后, 在客户端在进行反序列化

因此我们要对解析器进行序列化和反序列化