RPC（四）| 来自Google的gRPC

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

// 文件：grpc_demo/pro/spider.proto
syntax = "proto3";  // 协议为proto3

option go_package = ".;pro";
// package spider;  // 包名

// 发送请求
message SendAddress {
    // 发送的参数字段
    // 参数类型 参数名 标识号(不可重复)
    string address = 1;  // 要请求的地址
    string method = 2;  // 请求方式
}

// 返回响应
message GetResponse {
    // 接收的参数字段
    // 参数类型 参数名 标识号
    int32 httpCode = 1;  // http状态码
    string response = 2;  // 返回体
}

// 定义服务,可定义多个服务,每个服务可多个接口
service GoSpider {
    // rpc请求 请求的函数 (发送请求参数) returns (返回响应的参数)
    rpc GetAddressResponse (SendAddress) returns (GetResponse);
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77

package server

import (
	"context"
	"google.golang.org/grpc"
	"google.golang.org/grpc/reflection"
	"io/ioutil"
	"net"
	"net/http"
	spider "yufa/demo/grpc_demo/pro"
)

type server struct{}

const (
	// Address 监听地址
	Address string = "localhost:8080"
	// Method 通信方法
	Method string = "tcp"
)

// 接收client端的请求,函数名需保持一致
// ctx参数必传
// 参数二为自定义的参数,需从pb文件导入,因此pb文件必须可导入,文件放哪里随意
// 返回值同参数二,为pb文件的返回结构体指针
func (s *server) GetAddressResponse(ctx context.Context, a *spider.SendAddress) 
	(*spider.GetResponse, error) {
	// 逻辑写在这里
	switch a.Method {
	case "get", "Get", "GET":
		// 演示微服务用,故只写get示例
		status, body, err := get(a.Address)
		if err != nil {
			return nil, err
		}
		res := spider.GetResponse{
			HttpCode: int32(status),
			Response: body,
		}
		return &res, nil
	}
	return nil, nil
}

func get(address string) (s int, r string, err error) {
	// get请求
	resp, err := http.Get(address)
	if err != nil {
		return
	}
	defer resp.Body.Close()
	s = resp.StatusCode
	body, err := ioutil.ReadAll(resp.Body)
	if err != nil {
		return
	}
	r = string(body)
	return
}

func GRPCSpiderServer() {
	// 监听本地端口
	listener, err := net.Listen(Method, Address)
	if err != nil {
		return
	}
	s := grpc.NewServer()                       // 创建GRPC
	spider.RegisterGoSpiderServer(s, &server{}) // 在GRPC服务端注册服务

	reflection.Register(s) // 在GRPC服务器注册服务器反射服务
	// Serve方法接收监听的端口,每到一个连接创建一个ServerTransport和server的grroutine
	// 这个goroutine读取GRPC请求,调用已注册的处理程序进行响应
	err = s.Serve(listener)
	if err != nil {
		return
	}
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40

package client

import (
	"context"
	"google.golang.org/grpc"
	spider "yufa/demo/grpc_demo/pro"
)

import "fmt"

const (
	// Address server端地址
	Address string = "localhost:8080"
)

func GRPCSpiderClient() {
	// 连接服务器
	conn, err := grpc.Dial(Address, grpc.WithInsecure())
	if err != nil {
		fmt.Println(err)
		return
	}
	defer conn.Close()

	// 连接GRPC
	c := spider.NewGoSpiderClient(conn)
	// 创建要发送的结构体
	req := spider.SendAddress{
		Address: "http://www.baidu.com",
		Method:  "get",
	}
	// 调用server的注册方法
	r, err := c.GetAddressResponse(context.Background(), &req)
	if err != nil {
		fmt.Println(err)
		return
	}
	// 打印返回值
	fmt.Println(r)
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

import (
	"context"
	"google.golang.org/grpc"
	"log"
	"net"
	"yufa/demo/grpc_demo/pro"
)

type serverHello struct{}

func (s *serverHello) SayHello(ctx context.Context, in *pro.HelloRequest)
	(*pro.HelloReplay, error) {
	log.Printf("Received Name: %v", in.GetName())
	return &pro.HelloReplay{Message: "Hello " + in.Name}, nil
}

const (
	port = ":8081"
)

func GRPCServerHello() {
	lis, err := net.Listen("tcp", port)
	if err != nil {
		log.Fatalf("Failed to listen: %v", err)
	}
	s := grpc.NewServer()
	pro.RegisterGreeterServer(s, &serverHello{})
	if err := s.Serve(lis); err != nil {
		log.Fatalf("Failed to serve: %v", err)
	}
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

import (
	"context"
	"google.golang.org/grpc"
	"log"
	"os"
	"time"
	"yufa/demo/grpc_demo/pro"
)

const (
	address     = ":8081"
	defaultName = "chende"
)

func GRPCClientHello() {
	conn, err := grpc.Dial(address, grpc.WithInsecure(), grpc.WithBlock())
	if err != nil {
		log.Fatalf("Did not connect: %v", err)
	}
	defer conn.Close()
	c := pro.NewGreeterClient(conn)

	name := defaultName
	if len(os.Args) > 1 {
		name = os.Args[1]
	}
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()

	r, err := c.SayHello(ctx, &pro.HelloRequest{Name: name})
	if err != nil {
		log.Fatalf("Could not great: %v", err)
	}
	log.Printf("Greeting: %s", r.GetMessage())
}

1
2
3
4
5
6
7

func init() {
	c := &compressor{}
	c.poolCompressor.New = func() interface{} {
		return &writer{Writer: gzip.NewWriter(ioutil.Discard), pool: &c.poolCompressor}
	}
	encoding.RegisterCompressor(c)
}