【go工程组件】高性能RPC框架gRPC

最近更新：2024-12-09 | 字数总计：8.4k | 阅读估时：43分钟 | 阅读量：次

高性能RPC框架gRPC

高性能RPC框架gRPC

使用Protobuf定义服务

安装protoc，并将protoc放到/usr/local/bin，将google文件夹放到/usr/local/include下

安装插件

1 2	go install google.golang.org/protobuf/cmd/protoc-gen-go@v1.28 go install google.golang.org/grpc/cmd/protoc-gen-go-grpc@v1.2

写proto文件

syntax = "proto3";
option go_package = "grpc/echo";
import "google/protobuf/timestamp.proto";
package grpc.echo;

message EchoRequest {
    string message = 1;
    bytes bytes = 2;
    int32 length = 3;
    google.protobuf.Timestamp time = 4;
}

message EchoResponse {
    string message = 1;
    bytes bytes = 2;
    int32 length = 3;
    google.protobuf.Timestamp time = 4;
}

service Echo {
    // 一元请求
    rpc UnaryEcho(EchoRequest) returns(EchoResponse){}
    // 服务端流
    rpc ServerStreamingEcho(EchoRequest) returns(stream EchoResponse){}
    // 客户端流
    rpc ClientStreamingEcho(stream EchoRequest) returns(EchoResponse){}
    // 双向流
    rpc BidirectionalStreamingEcho(stream EchoRequest) returns(stream EchoResponse){}
}

生成存根

1 2	# 同时生成message的存根和grpc的存根 protoc --go-grpc_out=. --go-grpc_opt=paths=source_relative ./echo/echo.proto --go_out=. --go_opt=paths=source_relative

自动追踪一下依赖
1
go mod tidy

一元请求与流式请求

解决go1.19版本依赖问题,go.mod:

1	replace google.golang.org/grpc v1.68.1 => google.golang.org/grpc v1.40.0

服务端

入口

package main

import (
	"flag"
	"fmt"
	"grpc/echo"
	"grpc/echo-server/server"
	"log"
	"net"

	"google.golang.org/grpc"
)

var (
	port = flag.Int("port", 50051, "")
)

func main() {
	flag.Parse()
	lis, err := net.Listen("tcp", fmt.Sprintf(":%d", *port))
	if err != nil {
		log.Fatal(err)
	}
	s := grpc.NewServer()
	echo.RegisterEchoServer(s, &server.EchoServer{})
	log.Printf("server listening at :%v\n", lis.Addr())
	if err := s.Serve(lis); err != nil {
		log.Fatal(err)
	}
}

实现

package server

import (
	"context"
	"fmt"
	"grpc/echo"
	"io"
	"log"
	"os"
	"strconv"
	"sync"
	"time"

	"google.golang.org/protobuf/types/known/timestamppb"
)

type EchoServer struct {
	echo.UnimplementedEchoServer // 必须内嵌实现
}

// 一元请求
func (EchoServer) UnaryEcho(ctx context.Context, in *echo.EchoRequest) (*echo.EchoResponse, error) {
	fmt.Printf("server recv :%v\n", in.Message)
	return &echo.EchoResponse{
		Message: "server send message",
	}, nil
	// return nil, status.Errorf(codes.Unimplemented, "method UnaryEcho not implemented")
}

// 服务端流，发送文件给客户端
func (EchoServer) ServerStreamingEcho(in *echo.EchoRequest, stream echo.Echo_ServerStreamingEchoServer) error {
	fmt.Printf("server recv :%v\n", in.Message)
	filePath := "echo-server/files/server.png"
	file, err := os.Open(filePath)
	if err != nil {
		log.Fatal(err)
	}
	defer file.Close()
	buf := make([]byte, 1024)
	for {
		n, err := file.Read(buf)
		if err == io.EOF {
			break
		}
		if err != nil {
			return err
		}
		stream.Send(&echo.EchoResponse{
			Message: "server send file",
			Bytes:   buf[:n],
			Time:    timestamppb.New(time.Now()),
			Length:  int32(n),
		})
	}
	// 服务端发送nil或者error，代表流结束
	return nil
	// return status.Errorf(codes.Unimplemented, "method ServerStreamingEcho not implemented")
}

// 客户端流
func (EchoServer) ClientStreamingEcho(stream echo.Echo_ClientStreamingEchoServer) error {
	filePath := "echo-server/files/" + strconv.FormatInt(time.Now().UnixMilli(), 10) + ".jpg"
	file, err := os.OpenFile(filePath, os.O_WRONLY|os.O_CREATE|os.O_APPEND, 0644)
	if err != nil {
		log.Fatal(err)
	}
	defer file.Close()
	for {
		req, err := stream.Recv()
		if err == io.EOF {
			break
		}
		if err != nil {
			log.Println(err)
			break
		}
		file.Write(req.Bytes[:req.Length])
		fmt.Printf("server recv : %v\n", req.Message)
	}
	err = stream.SendAndClose(&echo.EchoResponse{
		Message: "server send message",
	})
	return err
	// return status.Errorf(codes.Unimplemented, "method ClientStreamingEcho not implemented")
}
func (EchoServer) BidirectionalStreamingEcho(stream echo.Echo_BidirectionalStreamingEchoServer) error {
	wg := &sync.WaitGroup{}
	wg.Add(1)
	go func() {
		defer wg.Done()
		filePath := "echo-server/files/" + strconv.FormatInt(time.Now().UnixMilli(), 10) + ".jpg"
		file, err := os.OpenFile(filePath, os.O_WRONLY|os.O_CREATE|os.O_APPEND, 0644)
		if err != nil {
			log.Fatal(err)
		}
		defer file.Close()
		for {
			req, err := stream.Recv()
			if err == io.EOF {
				break
			}
			if err != nil {
				log.Println(err)
				break
			}
			file.Write(req.Bytes[:req.Length])
			fmt.Printf("server recv : %v\n", req.Message)
		}
	}()

	wg.Add(1)
	go func() {
		defer wg.Done()
		filePath := "echo-server/files/server.png"
		file, err := os.Open(filePath)
		if err != nil {
			log.Fatal(err)
		}
		defer file.Close()
		buf := make([]byte, 1024)
		for {
			n, err := file.Read(buf)
			if err == io.EOF {
				break
			}
			if err != nil {
				log.Println(err)
				break
			}
			stream.Send(&echo.EchoResponse{
				Message: "server send file",
				Bytes:   buf[:n],
				Time:    timestamppb.New(time.Now()),
				Length:  int32(n),
			})
		}
	}()
	wg.Wait()
	return nil
	// return status.Errorf(codes.Unimplemented, "method BidirectionalStreamingEcho not implemented")
}

客户端

入口

package main

import (
	"flag"
	"grpc/echo"
	"grpc/echo-client/client"
	"log"

	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials/insecure"
)

var (
	addr = flag.String("addr", "localhost:50051", "")
)

func main() {
	flag.Parse()
	conn, err := grpc.Dial(*addr, grpc.WithTransportCredentials(insecure.NewCredentials()))
	if err != nil {
		log.Fatal(err)
	}
	defer conn.Close()
	c := echo.NewEchoClient(conn)
	// client.CallServerStream(c)
	// client.CallUnary(c)
	// client.CallClientStream(c)
	client.CallBidirectional(c)
}

实现

package client

import (
	"context"
	"fmt"
	"grpc/echo"
	"io"
	"log"
	"os"
	"strconv"
	"sync"
	"time"

	"google.golang.org/protobuf/types/known/timestamppb"
)

func CallUnary(client echo.EchoClient) {
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	in := &echo.EchoRequest{
		Message: "client send message",
		Time:    timestamppb.New(time.Now()),
	}
	res, err := client.UnaryEcho(ctx, in)
	if err != nil {
		log.Fatal(err)
	}
	fmt.Printf("client recv :%v \n", res.Message)
}

func CallServerStream(client echo.EchoClient) {
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	in := &echo.EchoRequest{
		Message: "client send message",
		Time:    timestamppb.New(time.Now()),
	}
	stream, err := client.ServerStreamingEcho(ctx, in)
	if err != nil {
		log.Fatal(err)
	}
	filename := "echo-client/files/" + strconv.FormatInt(time.Now().UnixMilli(), 10) + ".png"
	file, err := os.OpenFile(filename, os.O_CREATE|os.O_APPEND|os.O_WRONLY, 0644)
	if err != nil {
		log.Fatal(err)
	}
	defer file.Close()
	for {
		res, err := stream.Recv()
		if err == io.EOF {
			break
		}
		if err != nil {
			log.Println(err)
			break
		}
		file.Write(res.Bytes[:res.Length])
		fmt.Printf("client recv : %v\n", res.Message)
	}
	stream.CloseSend()
}

func CallClientStream(client echo.EchoClient) {
	filePath := "echo-client/files/client.jpg"
	file, err := os.Open(filePath)
	if err != nil {
		log.Fatal(err)
	}
	defer file.Close()
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	stream, err := client.ClientStreamingEcho(ctx)
	if err != nil {
		log.Fatal(err)
	}
	buf := make([]byte, 1024)
	for {
		n, err := file.Read(buf)
		if err == io.EOF {
			break
		}
		if err != nil {
			log.Fatal(err)
		}
		stream.Send(&echo.EchoRequest{
			Message: "client send file",
			Bytes:   buf[:n],
			Time:    timestamppb.New(time.Now()),
			Length:  int32(n),
		})
	}
	// 客户端一定要显式调用close，不然服务端收不到io.EOF
	res, err := stream.CloseAndRecv()
	if err != nil {
		log.Fatal(err)
	}
	fmt.Printf("client recv: %v\n", res.Message)
}

func CallBidirectional(client echo.EchoClient) {
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	stream, err := client.BidirectionalStreamingEcho(ctx)
	if err != nil {
		log.Fatal(err)
	}

	wg := &sync.WaitGroup{}
	wg.Add(1)
	go func() {
		defer wg.Done()
		filePath := "echo-client/files/client.jpg"
		file, err := os.Open(filePath)
		if err != nil {
			log.Fatal(err)
		}
		defer file.Close()
		buf := make([]byte, 1024)
		for {
			n, err := file.Read(buf)
			if err == io.EOF {
				break
			}
			if err != nil {
				log.Fatal(err)
			}
			stream.Send(&echo.EchoRequest{
				Message: "client send file",
				Bytes:   buf[:n],
				Time:    timestamppb.New(time.Now()),
				Length:  int32(n),
			})
		}
		stream.CloseSend()
	}()

	wg.Add(1)
	go func() {
		defer wg.Done()
		filename := "echo-client/files/" + strconv.FormatInt(time.Now().UnixMilli(), 10) + ".png"
		file, err := os.OpenFile(filename, os.O_CREATE|os.O_APPEND|os.O_WRONLY, 0644)
		if err != nil {
			log.Fatal(err)
		}
		defer file.Close()
		for {
			res, err := stream.Recv()
			if err == io.EOF {
				break
			}
			if err != nil {
				log.Println(err)
				break
			}
			file.Write(res.Bytes[:res.Length])
			fmt.Printf("client recv : %v\n", res.Message)
		}
		stream.CloseSend()
	}()
	wg.Wait()
}

元数据的意义：将某些信息以header或trailer（头或尾）的方式，以http2的头尾包方式传递请求头尾

客户端元数据

metadata.go

package client

import (
	"context"

	"google.golang.org/grpc/metadata"
)

func getMetadataByMap(mp map[string]string) metadata.MD {
	// 通过map初始化metadata
	md := metadata.New(mp)
	return md
}

func getMetadataByKV(kv ...string) metadata.MD {
	// 通过键值对的方式初始化metadata
	md := metadata.Pairs(kv...)
	return md
}

func getOutgoingContext(ctx context.Context, md metadata.MD) context.Context {
	// OutgoingContext 用于请求发送方，包装数据传递出去
	// IncomingContext 用于请求接收方，用于获取发送方传递的数据
	// context 通过序列化成http2 header的方式传输
	// new 方法会覆盖ctx原有的数据, 如果不希望覆盖需要用append方法
	ctx = metadata.NewOutgoingContext(ctx, md)
	return ctx
}

// 将数据附加到OutgoingContext
func appendToOutgoingContext(ctx context.Context, kv ...string) context.Context {
	return metadata.AppendToOutgoingContext(ctx, kv...)
}

服务端元数据

metadata.go

服务端作为响应方，不需要设置context，只需要得到metadata

package server

import "google.golang.org/grpc/metadata"

func getMetadataByMap(mp map[string]string) metadata.MD {
	// 通过map初始化metadata
	md := metadata.New(mp)
	return md
}

func getMetadataByKV(kv ...string) metadata.MD {
	// 通过键值对的方式初始化metadata
	md := metadata.Pairs(kv...)
	return md
}

客户端设置元数据和获取元数据

一元请求

func GetContext(ctx context.Context) context.Context {
	md := getMetadataByMap(map[string]string{
		"time":        time.Now().Format("2006-01-02T15-04-05Z07:00"),
		"header_data": "true",
	})
	ctx = getOutgoingContext(ctx, md)
	ctx = appendToOutgoingContext(ctx, "k1", "value1", "k2", "value2")
	// md1, _ := metadata.FromOutgoingContext(ctx)
	// fmt.Println(md1)
	return ctx
}
func CallUnary(client echo.EchoClient) {
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
    // 客户端设置元数据
	ctx = GetContext(ctx)
	in := &echo.EchoRequest{
		Message: "client send message",
		Time:    timestamppb.New(time.Now()),
	}
	// 客户端获取元数据
	var header, trailer metadata.MD
	res, err := client.UnaryEcho(ctx, in, grpc.Header(&header), grpc.Trailer(&trailer))
	if err != nil {
		log.Fatal(err)
	}
	fmt.Printf("client recv :%v \n", res.Message)
	fmt.Println("client recv CallUnary header:", header)
	fmt.Println("client recv CallUnary trailer:", trailer)
}

流式调用

func CallServerStream(client echo.EchoClient) {
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
    // 客户端设置元数据
	ctx = GetContext(ctx)
	in := &echo.EchoRequest{
		Message: "client send message",
		Time:    timestamppb.New(time.Now()),
	}
	stream, err := client.ServerStreamingEcho(ctx, in)
	if err != nil {
		log.Fatal(err)
	}
	// 客户端获取元数据
	header, _ := stream.Header()
	trailer := stream.Trailer() // 在这个位置是拿不到的，需要在结束位置获取
	fmt.Println("client recv CallServerStream header:", header)
	fmt.Println("client recv CallServerStream trailer:", trailer)

	filename := "echo-client/files/" + strconv.FormatInt(time.Now().UnixMilli(), 10) + ".png"
	file, err := os.OpenFile(filename, os.O_CREATE|os.O_APPEND|os.O_WRONLY, 0644)
	if err != nil {
		log.Fatal(err)
	}
	defer file.Close()
	for {
		res, err := stream.Recv()
		if err == io.EOF {
			break
		}
		if err != nil {
			log.Println(err)
			break
		}
		file.Write(res.Bytes[:res.Length])
		// fmt.Printf("client recv : %v\n", res.Message)
	}
	stream.CloseSend()
	trailer = stream.Trailer() // 在这个位置才能达到服务端设置的trailer
	fmt.Println("client recv CallServerStream trailer:", trailer)
}

服务端设置元数据和获取元数据

一元请求

// 一元请求
func (EchoServer) UnaryEcho(ctx context.Context, in *echo.EchoRequest) (*echo.EchoResponse, error) {
	// 获取客户端请求中的元数据
	md, ok := metadata.FromIncomingContext(ctx)
	if !ok {
		log.Println("server recv UnaryEcho 元数据失败")
	} else {
		fmt.Println("server recv UnaryEcho 元数据 time : ", md.Get("time"))
	}

	// 响应请求，发送服务端元数据
	header, trailer := getMetadata()
	grpc.SendHeader(ctx, header)
	defer grpc.SetTrailer(ctx, trailer) // 在发送结束的时候发送一个尾部元数据

	fmt.Printf("server recv :%v\n", in.Message)
	return &echo.EchoResponse{
		Message: "server send message",
	}, nil
}

流式调用

// 服务端流，发送文件给客户端
func (EchoServer) ServerStreamingEcho(in *echo.EchoRequest, stream echo.Echo_ServerStreamingEchoServer) error {
	// 获取流请求中的元数据
	md, ok := metadata.FromIncomingContext(stream.Context())
	if !ok {
		log.Println("server recv ServerStreamingEcho 元数据失败")
	} else {
		fmt.Println("server recv ServerStreamingEcho 元数据 time : ", md.Get("time"))
	}

	// 流式调用添加元数据，注意和一元调用时的不同
	header, trailer := getMetadata()
	stream.SendHeader(header)
	defer stream.SetTrailer(trailer)

	fmt.Printf("server recv :%v\n", in.Message)
	filePath := "echo-server/files/server.png"
	file, err := os.Open(filePath)
	if err != nil {
		log.Fatal(err)
	}
	defer file.Close()
	buf := make([]byte, 1024)
	for {
		n, err := file.Read(buf)
		if err == io.EOF {
			break
		}
		if err != nil {
			return err
		}
		stream.Send(&echo.EchoResponse{
			Message: "server send file",
			Bytes:   buf[:n],
			Time:    timestamppb.New(time.Now()),
			Length:  int32(n),
		})
	}
	// 服务端发送nil或者error，代表流结束
	return nil
}

TLS与双向TLS确保通讯安全

对称加解密：客户端和服务端保存相同的密钥，使用该密钥对消息进行加解密。缺点：密钥容易泄露，因为用户的客户端的密钥非常容易泄露；
非对称加解密：
1. 通过公钥加密和私钥解密。
  1. 客户端request服务端，服务端响应公钥。
  2. 客户端生成随机的客户端公钥，并通过服务端公钥加密，发送给服务器。服务端通过自己的私钥解密知晓了这个客户端公钥
  3. 此时客户端通过服务端的公钥进行加密发送消息，而服务端则通过客户端的公钥加密发送消息，双方都通过自己的私钥解密消息
  - 缺陷：假如客户端在第一次通信时就被劫持了，第三方拦截充当中间商沟通客户端和服务端，那么客户端和服务端的消息都会被第三方知晓
2. 通过私钥签名，通过公钥验签（证书机制）
  1. 根证书机构签发证书：
    1. 根证书机构生成自己的公钥和私钥
    2. 生成待申请的服务器的公钥和私钥
    3. 通过证书机构的私钥签名（加密）服务器公钥生成证书
    4. 将服务器的证书，密钥，根证书（根证书机构的自签名证书）发送给服务器
  2. 引入证书后的双端通信
    1. 客户端request服务端，服务端发送证书给客户端（证书包含了服务器的公钥）
    2. 对证书做验签，原理：客户端的设备本身就有预装了根证书机构（权威机构）的根证书，他里面包含的公钥可以对服务器的证书进行验签
    3. 客户端生成随机key，使用服务器公钥加密传给服务端
    4. 服务端使用私钥解密，获取这个随机key，最后通过这个key进行消息传输
数字签名与数字证书：
1. 数字签名是使用私钥对信息加密后生成的，主要目的是：只有使用对应公钥进行验签，才能验证通过，目的是确保信息没有被篡权。
2. 数字证书是CA（根证书机构）使用CA私钥进行对服务器公钥进行加密后生成的，只有使用CA证书能够验证的一种证书，目的是由第三方（根证书机构）保证服务器的身份。

生成证书

从grpc-go/examples/data/x509源码目录下拷贝create.sh和openssl.cnf，用于进行证书和公私钥的生成：
1. ca_cert.pem
2. ca_key.pem
3. client_ca_cert.pem
4. client_ca_key.pem
5. client_cert.pem
6. client_key.pem
7. server_cert.pem
8. server_key.pem

服务端

tls.go

package server

import (
	"crypto/tls"
	"crypto/x509"
	"log"
	"os"

	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials"
)

// 单项TLS
func GetTLsOpt() grpc.ServerOption {
	cred, err := credentials.NewServerTLSFromFile("x509/server_cert.pem", "x509/server_key.pem")
	if err != nil {
		log.Fatal(err)
	}
	return grpc.Creds(cred)
}

// 双向TLS
func GetMTlsOpt() grpc.ServerOption {
	cert, err := tls.LoadX509KeyPair("x509/server_cert.pem", "x509/server_key.pem")
	if err != nil {
		log.Fatal(err)
	}
	ca := x509.NewCertPool()
	caFilePath := "x509/client_ca_cert.pem" // 需要加载客户端CA
	bytes, err := os.ReadFile(caFilePath)
	if err != nil {
		log.Fatal(err)
	}
	if ok := ca.AppendCertsFromPEM(bytes); !ok {
		log.Fatal("ca append failed")
	}
	tlsConf := &tls.Config{
		ClientAuth:   tls.RequestClientCert,
		Certificates: []tls.Certificate{cert},
		ClientCAs:    ca,
	}
	return grpc.Creds(credentials.NewTLS(tlsConf))
}

main.go

package main

import (
	"flag"
	"fmt"
	"grpc/echo"
	"grpc/echo-server/server"
	"log"
	"net"

	"google.golang.org/grpc"
)

var (
	port = flag.Int("port", 50051, "")
)

func getOptions() (opts []grpc.ServerOption) {
	opts = make([]grpc.ServerOption, 0)
	// opts = append(opts, server.GetTLsOpt()) // 单向TLS
	opts = append(opts, server.GetMTlsOpt())
	return opts
}

func main() {
	flag.Parse()
	lis, err := net.Listen("tcp", fmt.Sprintf(":%d", *port))
	if err != nil {
		log.Fatal(err)
	}
	// 在新建服务时传入opts
	s := grpc.NewServer(getOptions()...)
	echo.RegisterEchoServer(s, &server.EchoServer{})
	log.Printf("server listening at :%v\n", lis.Addr())
	if err := s.Serve(lis); err != nil {
		log.Fatal(err)
	}
}

客户端

tls.go

package client

import (
	"crypto/tls"
	"crypto/x509"
	"log"
	"os"

	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials"
	"google.golang.org/grpc/credentials/insecure"
)

func GetinsecureOpt() grpc.DialOption {
	return grpc.WithTransportCredentials(insecure.NewCredentials())
}

func GetTLsOpt() grpc.DialOption {
	// 注意和服务端的区别，这里需要传入ca证书和域名
	cred, err := credentials.NewClientTLSFromFile("x509/ca_cert.pem", "echo.test.example.com")
	if err != nil {
		log.Fatal(err)
	}
	opt := grpc.WithTransportCredentials(cred)
	return opt
}

// 双向TLS
func GetMTlsOpt() grpc.DialOption {
	cert, err := tls.LoadX509KeyPair("x509/client_cert.pem", "x509/client_key.pem")
	if err != nil {
		log.Fatal(err)
	}
	ca := x509.NewCertPool()
	caFilePath := "x509/ca_cert.pem" // 需要加载服务端CA
	bytes, err := os.ReadFile(caFilePath)
	if err != nil {
		log.Fatal(err)
	}
	if ok := ca.AppendCertsFromPEM(bytes); !ok {
		log.Fatal("ca append failed")
	}
	tlsConf := &tls.Config{
		ServerName:   "echo.test.example.com",
		Certificates: []tls.Certificate{cert},
		RootCAs:      ca,
	}
	return grpc.WithTransportCredentials(credentials.NewTLS(tlsConf))
}

main.go

package main

import (
	"flag"
	"grpc/echo"
	"grpc/echo-client/client"
	"log"

	"google.golang.org/grpc"
)

var (
	addr = flag.String("addr", "localhost:50051", "")
)

func getOptions() (opts []grpc.DialOption) {
	opts = make([]grpc.DialOption, 0)
	// opts = append(opts, client.GetTLsOpt())
	opts = append(opts, client.GetMTlsOpt())
	return opts
}

func main() {
	flag.Parse()
	// conn, err := grpc.Dial(*addr, grpc.WithTransportCredentials(insecure.NewCredentials()))
	conn, err := grpc.Dial(*addr, getOptions()...)
	if err != nil {
		log.Fatal(err)
	}
	defer conn.Close()
	c := echo.NewEchoClient(conn)
	client.CallServerStream(c)
	// client.CallUnary(c)
	// client.CallClientStream(c)
	// client.CallBidirectional(c)

}

OAuth2认证与拦截器

服务端

interceptor.go

package server

import (
	"context"
	"errors"
	"fmt"
	"strings"

	"google.golang.org/grpc"
	"google.golang.org/grpc/metadata"
)

// 拦截器
func UnaryInterceptor(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (resp interface{}, err error) {
	fmt.Println("server UnaryInterceptor")
	fmt.Println(info)
	err = oauth2Valid(ctx) // 添加身份校验
	if err != nil {
		return nil, err
	}
	// 所谓的拦截器就是中间件，所以最后一定要调用handler
	return handler(ctx, req)
}

func StreamInterceptor(srv interface{}, ss grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) error {
	fmt.Println("server StreamInterceptor")
	fmt.Println(info)
	err := oauth2Valid(ss.Context()) // 添加身份校验
	if err != nil {
		return err
	}
	return handler(srv, ss)
}

func oauth2Valid(ctx context.Context) error {
	md, ok := metadata.FromIncomingContext(ctx)
	if !ok {
		return errors.New("元数据获取失败，身份认证失败")
	}
	authorization := md["authorization"]
	if !valid(authorization) {
		return errors.New("身份校验失败")
	}
	return nil
}

func valid(authorization []string) bool {
	if len(authorization) < 1 { // 压根没有带token
		fmt.Println("authorization.length < 1")
		return false
	}
	token := strings.TrimPrefix(authorization[0], "Bearer ") // 去掉前缀,不要忘记空格
	return token == fetchToken()
}

func fetchToken() string {
	return "some-secret-token" // 模拟一个
}

main.go

package main

import (
	"flag"
	"fmt"
	"grpc/echo"
	"grpc/echo-server/server"
	"log"
	"net"

	"google.golang.org/grpc"
)

var (
	port = flag.Int("port", 50051, "")
)

func getOptions() (opts []grpc.ServerOption) {
	opts = make([]grpc.ServerOption, 0)
	// opts = append(opts, server.GetTLsOpt()) // 单向TLS
	opts = append(opts, server.GetMTlsOpt())
	// 附加一元拦截器
	opts = append(opts, grpc.UnaryInterceptor(server.UnaryInterceptor))
	// 附加流式请求拦截器
	opts = append(opts, grpc.StreamInterceptor(server.StreamInterceptor))
	// 附加链式一元拦截器
	// opts = append(opts, grpc.ChainUnaryInterceptor())
	// 附加链式流式请求拦截器
	// opts = append(opts, grpc.ChainStreamInterceptor())
	return opts
}

func main() {
	flag.Parse()
	lis, err := net.Listen("tcp", fmt.Sprintf(":%d", *port))
	if err != nil {
		log.Fatal(err)
	}
	// 在新建服务时传入opts
	s := grpc.NewServer(getOptions()...)
	echo.RegisterEchoServer(s, &server.EchoServer{})
	log.Printf("server listening at :%v\n", lis.Addr())
	if err := s.Serve(lis); err != nil {
		log.Fatal(err)
	}
}

客户端

interceptor.go

package client

import (
	"context"
	"fmt"

	"google.golang.org/grpc"
)

func UnaryInterceptor(ctx context.Context, method string, req, reply interface{}, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
	fmt.Println("client UnaryInterceptor")
	var credsConfiguared bool
	for _, opt := range opts {
		_, ok := opt.(*grpc.PerRPCCredsCallOption)
		if ok {
			credsConfiguared = true //已经配置过了，不用再设置
			break
		}
	}
	if !credsConfiguared {
		opts = append(opts, grpc.PerRPCCredentials(GetPerRPCCredentails(FetchToken())))
	}
	// 所谓的拦截器就是中间件，所以最后一定要调用handler
	return invoker(ctx, method, req, reply, cc, opts...)
}

func StreamInterceptor(ctx context.Context, desc *grpc.StreamDesc, cc *grpc.ClientConn, method string, streamer grpc.Streamer, opts ...grpc.CallOption) (grpc.ClientStream, error) {
	fmt.Println("client StreamInterceptor")
	var credsConfiguared bool
	for _, opt := range opts {
		_, ok := opt.(*grpc.PerRPCCredsCallOption)
		if ok {
			credsConfiguared = true //已经配置过了，不用再设置
			break
		}
	}
	if !credsConfiguared { // main方法中没有配置，我们再这里添加oauth
		opts = append(opts, grpc.PerRPCCredentials(GetPerRPCCredentails(FetchToken())))
	}
	return streamer(ctx, desc, cc, method, opts...)
}

oauth.go

package client

import (
	"golang.org/x/oauth2"
	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials"
	"google.golang.org/grpc/credentials/oauth"
)

func GetOAuth(token string) grpc.DialOption {
	perRPC := GetPerRPCCredentails(token)
	return grpc.WithPerRPCCredentials(perRPC)
}

func GetPerRPCCredentails(token string) credentials.PerRPCCredentials {
	return oauth.TokenSource{TokenSource: oauth2.StaticTokenSource(&oauth2.Token{AccessToken: token})}
}

func FetchToken() string {
	return "some-secret-token" // 模拟一个
}

main.go

package main

import (
	"flag"
	"grpc/echo"
	"grpc/echo-client/client"
	"log"

	"google.golang.org/grpc"
)

var (
	addr = flag.String("addr", "localhost:50051", "")
)

func getOptions() (opts []grpc.DialOption) {
	opts = make([]grpc.DialOption, 0)
	// opts = append(opts, client.GetTLsOpt())
	opts = append(opts, client.GetMTlsOpt())
	opts = append(opts, grpc.WithUnaryInterceptor(client.UnaryInterceptor))
	opts = append(opts, grpc.WithStreamInterceptor(client.StreamInterceptor))
	// 在main方法中添加oauth内容，也可以参考client/interceptor中的方法，在拦截器中添加oauth
	// opts = append(opts, client.GetOAuth(client.FetchToken()))
	return opts
}

func main() {
	flag.Parse()
	// conn, err := grpc.Dial(*addr, grpc.WithTransportCredentials(insecure.NewCredentials()))
	conn, err := grpc.Dial(*addr, getOptions()...)
	if err != nil {
		log.Fatal(err)
	}
	defer conn.Close()
	c := echo.NewEchoClient(conn)
	client.CallServerStream(c)
	client.CallUnary(c)
	// client.CallClientStream(c)
	// client.CallBidirectional(c)

}

Keepalive连接保活配置

服务端

keepalive.go

package server

import (
	"time"

	"google.golang.org/grpc"
	"google.golang.org/grpc/keepalive"
)

// keepalive包括两部分：一个是服务端对客户端的强制策略；一个是服务端的配置；
func GetKeepaliveOpt() (opts []grpc.ServerOption) {
	// 服务端针对客户端的强制保活策略，客户端违反该策略将会被强制关闭
	var keep = keepalive.EnforcementPolicy{
		// 客户端ping服务器的最小时间间隔，小于该时间间隔强制关闭连接
		MinTime: 5 * time.Second,
		// 当前没有任何活动流的情况下，是否允许被ping
		PermitWithoutStream: true,
	}
	// 服务端保活配置
	var kasp = keepalive.ServerParameters{
		// 客户端空闲15s后发送goaway指令（尝试断开连接）
		MaxConnectionIdle: 15 * time.Second,
		// 最大连接时长30s，超时发送goaway
		MaxConnectionAge: 30 * time.Second,
		// 强制关闭前等待时长
		MaxConnectionAgeGrace: 5 * time.Second,
		// 客户端空闲5s，发送ping保活
		Time: 5 * time.Second,
		// ping 操作超时时间
		Timeout: 1 * time.Second,
	}
	return []grpc.ServerOption{grpc.KeepaliveEnforcementPolicy(keep), grpc.KeepaliveParams(kasp)}
}

main.go

package main

import (
	"flag"
	"fmt"
	"grpc/echo"
	"grpc/echo-server/server"
	"log"
	"net"

	"google.golang.org/grpc"
)

var (
	port = flag.Int("port", 50051, "")
)

func getOptions() (opts []grpc.ServerOption) {
	opts = make([]grpc.ServerOption, 0)
	// opts = append(opts, server.GetTLsOpt()) // 单向TLS
	opts = append(opts, server.GetMTlsOpt())
	// 附加一元拦截器
	opts = append(opts, grpc.UnaryInterceptor(server.UnaryInterceptor))
	// 附加流式请求拦截器
	opts = append(opts, grpc.StreamInterceptor(server.StreamInterceptor))
	// 附加链式一元拦截器
	// opts = append(opts, grpc.ChainUnaryInterceptor())
	// 附加链式流式请求拦截器
	// opts = append(opts, grpc.ChainStreamInterceptor())
	opts = append(opts, server.GetKeepaliveOpt()...) // keepalive的配置
	return opts
}

func main() {
	flag.Parse()
	lis, err := net.Listen("tcp", fmt.Sprintf(":%d", *port))
	if err != nil {
		log.Fatal(err)
	}
	// 在新建服务时传入opts
	s := grpc.NewServer(getOptions()...)
	echo.RegisterEchoServer(s, &server.EchoServer{})
	log.Printf("server listening at :%v\n", lis.Addr())
	if err := s.Serve(lis); err != nil {
		log.Fatal(err)
	}
}

客户端

keepalive.go

package client

import (
	"time"

	"google.golang.org/grpc"
	"google.golang.org/grpc/keepalive"
)

func GetKeepaliveOpt() (opt grpc.DialOption) {
	var kacp = keepalive.ClientParameters{
		// 如果没有活动流，则每10s发送一个ping
		Time: 10 * time.Second,
		// ping超时时长
		Timeout: time.Second,
		// 当没有活动流的情况下，是否允许被ping
		PermitWithoutStream: true,
	}
	return grpc.WithKeepaliveParams(kacp)
}

main.go

package main

import (
	"flag"
	"grpc/echo"
	"grpc/echo-client/client"
	"log"

	"google.golang.org/grpc"
)

var (
	addr = flag.String("addr", "localhost:50051", "")
)

func getOptions() (opts []grpc.DialOption) {
	opts = make([]grpc.DialOption, 0)
	// opts = append(opts, client.GetTLsOpt())
	opts = append(opts, client.GetMTlsOpt())
	opts = append(opts, grpc.WithUnaryInterceptor(client.UnaryInterceptor))
	opts = append(opts, grpc.WithStreamInterceptor(client.StreamInterceptor))
	// 在main方法中添加oauth内容，也可以参考client/interceptor中的方法，在拦截器中添加oauth
	// opts = append(opts, client.GetOAuth(client.FetchToken()))
	// 添加保活配置
	opts = append(opts, client.GetKeepaliveOpt())
	return opts
}

func main() {
	flag.Parse()
	// conn, err := grpc.Dial(*addr, grpc.WithTransportCredentials(insecure.NewCredentials()))
	conn, err := grpc.Dial(*addr, getOptions()...)
	if err != nil {
		log.Fatal(err)
	}
	defer conn.Close()
	c := echo.NewEchoClient(conn)
	client.CallServerStream(c)
	client.CallUnary(c)
	// client.CallClientStream(c)
	// client.CallBidirectional(c)

}

Resolver服务名称解析器

名称解析器 : 通过服务名称去解析出名称的地址

客户端添加简单名称解析器

name_resolver.go

package client

import (
	"log"

	"google.golang.org/grpc"
	"google.golang.org/grpc/resolver"
)

const (
	MyScheme      = "myscheme"
	MyServiceName = "myecho"
)

// 模拟的地址列表
var addrs = []string{"localhost:50001", "localhost:50002", "localhost:50003"}

func GetNameResolver() grpc.DialOption {
	return grpc.WithResolvers(&MyResolverBuilder{})
}

type MyResolverBuilder struct {
}

func (*MyResolverBuilder) Build(target resolver.Target, cc resolver.ClientConn, opts resolver.BuildOptions) (resolver.Resolver, error) {
	r := &MyResolver{
		target:       target,
		cc:           cc,
		addressStore: map[string][]string{MyServiceName: addrs},
	}
	r.start()
	return r, nil
}

func (r *MyResolver) start() {
	log.Println("Resolver start")
	addrStrs := r.addressStore[r.target.Endpoint]
	addrs := make([]resolver.Address, len(addrStrs))
	for i, s := range addrStrs {
		addrs[i] = resolver.Address{
			Addr: s,
		}
	}
	r.cc.UpdateState(resolver.State{Addresses: addrs})
}

func (*MyResolverBuilder) Scheme() string {
	return MyScheme
}

type MyResolver struct {
	target       resolver.Target
	cc           resolver.ClientConn
	addressStore map[string][]string
}

func (r *MyResolver) ResolveNow(o resolver.ResolveNowOptions) {
	// 这个函数什么时候调用？在服务中断的时候，我们需要重新进行服务解析，查找下一个地址，这时候会调用这个函数
	log.Println("ResolveNow")
	log.Println(r.cc)
	// 演示：起3个服务，终止50001，这里我们重新返回一个列表，这样就能正确跳转到50002了
	r.addressStore = map[string][]string{MyServiceName: []string{"localhost:50002", "localhost:50003"}}
	r.start()
	log.Println(r.cc)
}
func (r *MyResolver) Close() {

}

main.go

package main

import (
	"flag"
	"fmt"
	"grpc/echo"
	"grpc/echo-client/client"
	"log"

	"google.golang.org/grpc"
)

var (
	addr = flag.String("addr", "localhost:50051", "")
)

func getOptions() (opts []grpc.DialOption) {
	opts = make([]grpc.DialOption, 0)
	// opts = append(opts, client.GetTLsOpt())
	opts = append(opts, client.GetMTlsOpt())
	opts = append(opts, grpc.WithUnaryInterceptor(client.UnaryInterceptor))
	opts = append(opts, grpc.WithStreamInterceptor(client.StreamInterceptor))
	// 在main方法中添加oauth内容，也可以参考client/interceptor中的方法，在拦截器中添加oauth
	// opts = append(opts, client.GetOAuth(client.FetchToken()))
	// 添加保活配置
	opts = append(opts, client.GetKeepaliveOpt())
	// 添加名称解析器
	opts = append(opts, client.GetNameResolver())
	return opts
}

func main() {
	flag.Parse()
	// conn, err := grpc.Dial(*addr, grpc.WithTransportCredentials(insecure.NewCredentials()))
	// 根据地址访问
	// conn, err := grpc.Dial(*addr, getOptions()...)
	// 根据协议 + 服务名，通过名称服务解析，访问服务器
	conn, err := grpc.Dial(fmt.Sprintf("%s:///%s", client.MyScheme, client.MyServiceName), getOptions()...)
	if err != nil {
		log.Fatal(err)
	}
	defer conn.Close()
	c := echo.NewEchoClient(conn)
	client.CallServerStream(c)
	client.CallUnary(c)
	// client.CallClientStream(c)
	// client.CallBidirectional(c)

}

自定义实现名称解析服务器

实现名称解析服务器

定义服务

name.proto

syntax = "proto3";
option go_package = "grpc/name";

package grpc.name;

message NameRequest {
    string service_name = 1;
    repeated string address = 2;
}

message NameResponse {
    string service_name = 1;
    repeated string address = 2;
}

service Name {
    // 注册
    rpc Register(NameRequest) returns (NameResponse) {}
    // 删除
    rpc Delete(NameRequest) returns (NameResponse) {}
    // 保活
    rpc Keepalive(stream NameRequest) returns (NameResponse) {}
    // 获取
    rpc GetAddress(NameRequest) returns (NameResponse) {}
}

实现名称服务器

name.go

package server

import (
	"sync"
	"time"
)

// 使用内存存储服务地址
type Address struct {
	serviceName string
	addr        string
	expireAt    int64
}

type nameStore struct {
	data         map[string]map[string]*Address
	dataLocker   sync.RWMutex
	expireAtData map[int64][]*Address
	expireLocker sync.RWMutex
}

var serviceNameData *nameStore
var expireTs = time.Second * 10

func init() {
	serviceNameData = &nameStore{
		data:         map[string]map[string]*Address{},
		expireAtData: map[int64][]*Address{},
	}
	// 异步删除过期的
	go serviceNameData.clearData()
}

// 将服务名与地址注册到集合
func Register(serviceName, address string) {
	ns := serviceNameData
	addr := &Address{
		serviceName: serviceName,
		addr:        address,
		expireAt:    time.Now().Add(expireTs).Unix(),
	}
	ns.expireLocker.Lock()
	expireAtKey := addr.expireAt
	_, ok := ns.expireAtData[expireAtKey]
	if !ok {
		ns.expireAtData[expireAtKey] = make([]*Address, 0)
	}
	// 为什么先添加到过期时间的位置?
	// 因为我们的数据分了两个地方存储，有两个锁，假如其中一个锁的中间内容发生问题，会导致两个地方的内容不同步。
	// 假如将存储位置的代码放在前面，这部分成功了，但是过期存储的位置却失败了，那么我们将没有任何办法删除这部分数据了。
	ns.expireAtData[expireAtKey] = append(ns.expireAtData[expireAtKey], addr)
	ns.expireLocker.Unlock()
	ns.dataLocker.Lock()
	_, ok = ns.data[serviceName]
	if !ok {
		ns.data[serviceName] = make(map[string]*Address, 0)
	}
	ns.data[serviceName][address] = addr
	ns.dataLocker.Unlock()
}

// 删除注册信息
func Delete(serviceName, address string) {
	ns := serviceNameData
	ns.dataLocker.Lock()
	ns.deleteNotLock(serviceName, address)
	ns.dataLocker.Unlock()
}

func (ns nameStore) deleteNotLock(serviceName, address string) {
	_, ok := ns.data[serviceName]
	if ok {
		delete(ns.data[serviceName], address)
	}
}

// 修改注册信息有效时间
func Keepalive(serviceName, address string) {
	ns := serviceNameData
	_, ok := ns.data[serviceName]
	if !ok { // 服务存在，却没有找到内容？重新注册一个！
		Register(serviceName, address)
	}
	_, ok = ns.data[serviceName][address]
	if !ok { // 服务存在，却没有找到内容？重新注册一个！
		Register(serviceName, address)
	}
	ns.dataLocker.Lock()
	addr := ns.data[serviceName][address]
	addr.expireAt = time.Now().Add(expireTs).Unix()
	ns.data[serviceName][address] = addr
	ns.dataLocker.Unlock()
	ns.expireLocker.Lock()
	expireAtKey := addr.expireAt
	_, ok = ns.expireAtData[expireAtKey]
	if !ok {
		ns.expireAtData[expireAtKey] = make([]*Address, 0)
	}
	ns.expireAtData[expireAtKey] = append(ns.expireAtData[expireAtKey], addr)
	ns.expireLocker.Unlock()
}

// 根据服务名称获取地址信息
func GetByServiceName(serviceName string) []string {
	ns := serviceNameData
	ns.dataLocker.RLock()
	defer ns.dataLocker.RUnlock()
	mp, ok := ns.data[serviceName]
	if !ok {
		return []string{}
	}
	if len(mp) == 0 {
		return []string{}
	}
	list := make([]string, 0)
	for _, addr := range mp {
		if addr.expireAt <= time.Now().Unix() {
			continue
		}
		list = append(list, addr.addr)
	}
	return list
}

func (ns nameStore) clearData() {
	timeTicker := time.NewTicker(3 * time.Second) // 每3s触发一次
	defer timeTicker.Stop()
	for {
		select {
		case <-timeTicker.C:
			list := make([]*Address, 0)
			ns.expireLocker.Lock()
			for key, items := range ns.expireAtData {
				if key <= time.Now().Unix()-10 { // 过期数据多保留十秒
					list = append(list, items...)
					delete(ns.expireAtData, key)
				}
			}
			ns.expireLocker.Unlock()
			ns.dataLocker.Lock()
			for _, item := range list {
				if item.expireAt <= time.Now().Unix()-10 {
					ns.deleteNotLock(item.serviceName, item.addr)
				}
			}
			ns.dataLocker.Unlock()
		}
	}
}

func GetAllData() *nameStore {
	return serviceNameData
}

main.go

package server

import (
	"context"
	"fmt"
	"grpc/name"
	"io"
	"log"
)

type NameServer struct {
	name.UnimplementedNameServer
}

func (NameServer) Register(ctx context.Context, in *name.NameRequest) (*name.NameResponse, error) {
	for _, address := range in.Address {
		Register(in.ServiceName, address)
	}
	fmt.Println(GetByServiceName(in.ServiceName))
	return &name.NameResponse{
		ServiceName: in.ServiceName,
	}, nil
}
func (NameServer) Delete(ctx context.Context, in *name.NameRequest) (*name.NameResponse, error) {
	for _, address := range in.Address {
		Delete(in.ServiceName, address)
	}
	fmt.Println(GetByServiceName(in.ServiceName))
	return &name.NameResponse{
		ServiceName: in.ServiceName,
	}, nil
}
func (NameServer) Keepalive(stream name.Name_KeepaliveServer) error {
	for {
		req, err := stream.Recv()
		if err == io.EOF {
			break
		}
		if err != nil {
			return err
		}
		fmt.Println(req)
		for _, address := range req.Address {
			Keepalive(req.ServiceName, address)
		}
	}
	return stream.SendAndClose(&name.NameResponse{})
}
func (NameServer) GetAddress(ctx context.Context, in *name.NameRequest) (*name.NameResponse, error) {
	addresss := GetByServiceName(in.ServiceName)
	log.Println(addresss)
	return &name.NameResponse{
		ServiceName: in.ServiceName,
		Address:     addresss,
	}, nil
}

服务器注册服务

name.go

package server

import (
	"context"
	"grpc/name"
	"log"
	"time"

	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials/insecure"
)

type NameServer struct {
	conn *grpc.ClientConn
}

func NewNameServer(addr string) *NameServer {
	defer func() {
		if err := recover(); err != nil {
			log.Println(err)
		}
	}()
	conn, err := grpc.Dial(addr, grpc.WithTransportCredentials(insecure.NewCredentials()))
	if err != nil {
		log.Println(err)
	}
	return &NameServer{
		conn: conn,
	}
}

func (ns *NameServer) Close() {
	defer func() { // 这一段的目的是，确保当名称服务器没有启动的时候，我们也能正确启动我们的echo服务器
		if err := recover(); err != nil {
			log.Println(err)
		}
	}()
	ns.conn.Close()
}

func (ns *NameServer) RegisterName(serviceName, address string) {
	defer func() {
		if err := recover(); err != nil {
			log.Println(err)
		}
	}()
	client := name.NewNameClient(ns.conn) // 此时服务器作为名称解析服务器的客户端
	in := &name.NameRequest{
		ServiceName: serviceName,
		Address:     []string{address},
	}
	_, err := client.Register(context.Background(), in)
	if err != nil {
		log.Println(err)
	}
}

func (ns *NameServer) DeleteName(serviceName, address string) {
	defer func() {
		if err := recover(); err != nil {
			log.Println(err)
		}
	}()
	client := name.NewNameClient(ns.conn) // 此时服务器作为名称解析服务器的客户端
	in := &name.NameRequest{
		ServiceName: serviceName,
		Address:     []string{address},
	}
	_, err := client.Delete(context.Background(), in)
	if err != nil {
		log.Println(err)
	}
}

func (ns *NameServer) KeepaliveName(serviceName, address string) {
	defer func() {
		if err := recover(); err != nil {
			log.Println(err)
		}
	}()
	client := name.NewNameClient(ns.conn) // 此时服务器作为名称解析服务器的客户端
	in := &name.NameRequest{
		ServiceName: serviceName,
		Address:     []string{address},
	}
	stream, err := client.Keepalive(context.Background())
	if err != nil {
		log.Println(err)
	}
	for {
		stream.Send(in)
		time.Sleep(time.Second)
	}
}

main.go

package main

import (
	"context"
	"flag"
	"fmt"
	"grpc/echo"
	"grpc/echo-server/server"
	"log"
	"net"
	"os"
	"os/signal"

	"google.golang.org/grpc"
)

var (
	port = flag.Int("port", 50001, "")
)

func getOptions() (opts []grpc.ServerOption) {
	opts = make([]grpc.ServerOption, 0)
	// opts = append(opts, server.GetTLsOpt()) // 单向TLS
	opts = append(opts, server.GetMTlsOpt())
	// 附加一元拦截器
	opts = append(opts, grpc.UnaryInterceptor(server.UnaryInterceptor))
	// 附加流式请求拦截器
	opts = append(opts, grpc.StreamInterceptor(server.StreamInterceptor))
	// 附加链式一元拦截器
	// opts = append(opts, grpc.ChainUnaryInterceptor())
	// 附加链式流式请求拦截器
	// opts = append(opts, grpc.ChainStreamInterceptor())
	opts = append(opts, server.GetKeepaliveOpt()...) // keepalive的配置
	return opts
}

func main() {
	flag.Parse()
	lis, err := net.Listen("tcp", fmt.Sprintf(":%d", *port))
	if err != nil {
		log.Fatal(err)
	}
	// 在新建服务时传入opts
	s := grpc.NewServer(getOptions()...)
	echo.RegisterEchoServer(s, &server.EchoServer{})
	// 启动echo服务器
	go func() {
		log.Printf("server listening at :%v\n", lis.Addr())
		if err := s.Serve(lis); err != nil {
			log.Fatal(err)
		}
	}()
	// 连接name服务器，注册服务，并保活
	nameServer := server.NewNameServer("localhost:60001")
	serviceName := "myecho"
	addr := fmt.Sprintf("localhost:%d", *port)
	go func() {
		nameServer.RegisterName(serviceName, addr)
		nameServer.KeepaliveName(serviceName, addr)
	}()
	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, os.Kill)
	defer stop()
	<-ctx.Done()
	nameServer.DeleteName(serviceName, addr)
	nameServer.Close()
}

客户端查询服务

name_resolver.go

package client

import (
	"context"
	"grpc/name"
	"log"

	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials/insecure"
	"google.golang.org/grpc/resolver"
)

const (
	MyScheme      = "myscheme"
	MyServiceName = "myecho"
)

// 模拟的地址列表, 使用名称解析服务器就不再需要了
// var addrs = []string{"localhost:50001", "localhost:50002", "localhost:50003"}

func GetNameResolver(ns *NameServer) grpc.DialOption {
	nameServer = ns
	return grpc.WithResolvers(&MyResolverBuilder{})
}

type MyResolverBuilder struct {
}

func (*MyResolverBuilder) Build(target resolver.Target, cc resolver.ClientConn, opts resolver.BuildOptions) (resolver.Resolver, error) {
	r := &MyResolver{
		target: target,
		cc:     cc,
		// addressStore: map[string][]string{MyServiceName: addrs},
		addressStore: map[string][]string{MyServiceName: nameServer.getAddressByServiceName(MyServiceName)},
	}
	r.start()
	return r, nil
}

func (r *MyResolver) start() {
	log.Println("Resolver start")
	addrStrs := r.addressStore[r.target.Endpoint]
	addrs := make([]resolver.Address, len(addrStrs))
	for i, s := range addrStrs {
		addrs[i] = resolver.Address{
			Addr: s,
		}
	}
	r.cc.UpdateState(resolver.State{Addresses: addrs})
}

func (*MyResolverBuilder) Scheme() string {
	return MyScheme
}

type MyResolver struct {
	target       resolver.Target
	cc           resolver.ClientConn
	addressStore map[string][]string
}

func (r *MyResolver) ResolveNow(o resolver.ResolveNowOptions) {
	// 这个函数什么时候调用？在服务中断的时候，我们需要重新进行服务解析，查找下一个地址，这时候会调用这个函数
	log.Println("ResolveNow")
	log.Println(r.cc)
	// 演示：起3个服务，终止50001，这里我们重新返回一个列表，这样就能正确跳转到50002了
	// r.addressStore = map[string][]string{MyServiceName: []string{"localhost:50002", "localhost:50003"}}
	// 使用名称解析服务器
	r.addressStore = map[string][]string{MyServiceName: nameServer.getAddressByServiceName(MyServiceName)}
	r.start()
	log.Println(r.cc)
}
func (r *MyResolver) Close() {
	// 关闭名称解析服务器的连接
	nameServer.Close()
}

/* 连接自实现的名称服务器 */

var nameServer *NameServer

type NameServer struct {
	conn *grpc.ClientConn
}

func NewNameServer(addr string) *NameServer {
	defer func() {
		if err := recover(); err != nil {
			log.Println(err)
		}
	}()
	conn, err := grpc.Dial(addr, grpc.WithTransportCredentials(insecure.NewCredentials()))
	if err != nil {
		log.Println(err)
	}
	return &NameServer{
		conn: conn,
	}
}

func (ns *NameServer) Close() {
	defer func() { // 这一段的目的是，确保当名称服务器没有启动的时候，我们也能正确启动我们的echo服务器
		if err := recover(); err != nil {
			log.Println(err)
		}
	}()
	ns.conn.Close()
}

func (ns *NameServer) getAddressByServiceName(serviceName string) []string {
	defer func() {
		if err := recover(); err != nil {
			log.Println(err)
		}
	}()
	client := name.NewNameClient(ns.conn)
	in := &name.NameRequest{
		ServiceName: serviceName,
	}
	res, err := client.GetAddress(context.Background(), in)
	if err != nil {
		log.Println(err)
		return []string{}
	}
	log.Println(res.Address)
	return res.Address
}

main.go

package main

import (
	"flag"
	"fmt"
	"grpc/echo"
	"grpc/echo-client/client"
	"log"

	"google.golang.org/grpc"
)

var (
	addr = flag.String("addr", "localhost:50051", "")
)

func getOptions() (opts []grpc.DialOption) {
	opts = make([]grpc.DialOption, 0)
	// opts = append(opts, client.GetTLsOpt())
	opts = append(opts, client.GetMTlsOpt())
	opts = append(opts, grpc.WithUnaryInterceptor(client.UnaryInterceptor))
	opts = append(opts, grpc.WithStreamInterceptor(client.StreamInterceptor))
	// 在main方法中添加oauth内容，也可以参考client/interceptor中的方法，在拦截器中添加oauth
	// opts = append(opts, client.GetOAuth(client.FetchToken()))
	// 添加保活配置
	opts = append(opts, client.GetKeepaliveOpt())
	// 添加名称解析器
	opts = append(opts, client.GetNameResolver(client.NewNameServer("localhost:60001")))
	return opts
}

func main() {
	flag.Parse()
	// conn, err := grpc.Dial(*addr, grpc.WithTransportCredentials(insecure.NewCredentials()))
	// 根据地址访问
	// conn, err := grpc.Dial(*addr, getOptions()...)
	// 根据协议 + 服务名，通过名称服务解析，访问服务器
	conn, err := grpc.Dial(fmt.Sprintf("%s:///%s", client.MyScheme, client.MyServiceName), getOptions()...)
	if err != nil {
		log.Fatal(err)
	}
	defer conn.Close()
	c := echo.NewEchoClient(conn)
	client.CallServerStream(c)
	client.CallUnary(c)
	// client.CallClientStream(c)
	// client.CallBidirectional(c)

}

客户端Client连接池实现

客户端

grpc_client_pool.go

package client_pool

import (
	"log"
	"sync"

	"google.golang.org/grpc"
	"google.golang.org/grpc/connectivity"
)

type ClientPool interface {
	Get() *grpc.ClientConn
	Put(conn *grpc.ClientConn)
}

type clientpool struct {
	pool sync.Pool
}

func GetPool(target string, opts ...grpc.DialOption) (ClientPool, error) {
	return &clientpool{
		pool: sync.Pool{
			New: func() any {
				conn, err := grpc.Dial(target, opts...)
				if err != nil {
					log.Println(err)
					return nil
				}
				return conn
			},
		},
	}, nil
}

func (c *clientpool) Get() *grpc.ClientConn {
	conn := c.pool.Get().(*grpc.ClientConn)
	if conn.GetState() == connectivity.Shutdown || conn.GetState() == connectivity.TransientFailure {
		conn.Close()
		conn = c.pool.New().(*grpc.ClientConn)
	}
	return conn
}

func (c *clientpool) Put(conn *grpc.ClientConn) {
	if conn.GetState() == connectivity.Shutdown || conn.GetState() == connectivity.TransientFailure {
		conn.Close()
		return
	}
	c.pool.Put(conn)
}

main.go

package main

import (
	"flag"
	"fmt"
	"grpc/echo"
	"grpc/echo-client/client"
	client_pool "grpc/echo-client/client-pool"
	"log"

	"google.golang.org/grpc"
)

var (
	addr = flag.String("addr", "localhost:50051", "")
)

func getOptions() (opts []grpc.DialOption) {
	opts = make([]grpc.DialOption, 0)
	// opts = append(opts, client.GetTLsOpt())
	opts = append(opts, client.GetMTlsOpt())
	opts = append(opts, grpc.WithUnaryInterceptor(client.UnaryInterceptor))
	opts = append(opts, grpc.WithStreamInterceptor(client.StreamInterceptor))
	// 在main方法中添加oauth内容，也可以参考client/interceptor中的方法，在拦截器中添加oauth
	// opts = append(opts, client.GetOAuth(client.FetchToken()))
	// 添加保活配置
	opts = append(opts, client.GetKeepaliveOpt())
	// 添加名称解析器
	opts = append(opts, client.GetNameResolver(client.NewNameServer("localhost:60001")))
	return opts
}

func main() {
	flag.Parse()
	// // conn, err := grpc.Dial(*addr, grpc.WithTransportCredentials(insecure.NewCredentials()))
	// // 根据地址访问
	// // conn, err := grpc.Dial(*addr, getOptions()...)
	// // 根据协议 + 服务名，通过名称服务解析，访问服务器
	// conn, err := grpc.Dial(fmt.Sprintf("%s:///%s", client.MyScheme, client.MyServiceName), getOptions()...)
	// if err != nil {
	// 	log.Fatal(err)
	// }
	// defer conn.Close()

	// 通过连接池连接
	pool, err := client_pool.GetPool(fmt.Sprintf("%s:///%s", client.MyScheme, client.MyServiceName), getOptions()...)
	if err != nil {
		log.Fatal(err)
	}
	conn := pool.Get()
	defer pool.Put(conn) // 使用完不再close而是放回连接池

	c := echo.NewEchoClient(conn)
	client.CallServerStream(c)
	client.CallUnary(c)
	// client.CallClientStream(c)
	// client.CallBidirectional(c)

}

多路复用与健康检查

下载健康检查工具

mkdir health_probe
cd ./health_probe
# 启用了双向tls
./grpc_health_probe-linux-amd64 -addr=localhost:50021 -tls -tls-ca-cert ../x509/ca_cert.pem -tls-client-cert ../x509/client_cert.pem -tls-client-key ../x509/client_key.pem -tls-server-name echo.test.example.com

多路复用方式启动健康检查服务

main.go

package main

import (
	"context"
	"flag"
	"fmt"
	"grpc/echo"
	"grpc/echo-server/server"
	"log"
	"net"
	"os"
	"os/signal"

	"google.golang.org/grpc"
	"google.golang.org/grpc/health"
	"google.golang.org/grpc/health/grpc_health_v1"
)

var (
	port = flag.Int("port", 50001, "")
)

func getOptions() (opts []grpc.ServerOption) {
	opts = make([]grpc.ServerOption, 0)
	// opts = append(opts, server.GetTLsOpt()) // 单向TLS
	opts = append(opts, server.GetMTlsOpt())
	// 附加一元拦截器
	opts = append(opts, grpc.UnaryInterceptor(server.UnaryInterceptor))
	// 附加流式请求拦截器
	opts = append(opts, grpc.StreamInterceptor(server.StreamInterceptor))
	// 附加链式一元拦截器
	// opts = append(opts, grpc.ChainUnaryInterceptor())
	// 附加链式流式请求拦截器
	// opts = append(opts, grpc.ChainStreamInterceptor())
	opts = append(opts, server.GetKeepaliveOpt()...) // keepalive的配置
	return opts
}

func main() {
	flag.Parse()
	lis, err := net.Listen("tcp", fmt.Sprintf(":%d", *port))
	if err != nil {
		log.Fatal(err)
	}
	// 在新建服务时传入opts
	s := grpc.NewServer(getOptions()...)
	echo.RegisterEchoServer(s, &server.EchoServer{})
	// 健康检查服务
	// 在同一个grpc server上注册了多个自己实现的server，是多路复用
	// 为什么要启动多路复用，而不是新写一个s1:=grpc.NewServer(getOptions()...)，在s1上注册健康检查server呢？
	// 因为服务启动Serve需要使用一个新的协程，假设仅仅健康检查的服务启动了，那么也可以通过这个检查，那我们就无从得知待检查的服务是否正常了
	grpc_health_v1.RegisterHealthServer(s, health.NewServer())
	// 启动echo服务器
	go func() {
		log.Printf("server listening at :%v\n", lis.Addr())
		if err := s.Serve(lis); err != nil {
			log.Fatal(err)
		}
	}()
	// 连接name服务器，注册服务，并保活
	nameServer := server.NewNameServer("localhost:60001")
	serviceName := "myecho"
	addr := fmt.Sprintf("localhost:%d", *port)
	go func() {
		nameServer.RegisterName(serviceName, addr)
		nameServer.KeepaliveName(serviceName, addr)
	}()
	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, os.Kill)
	defer stop()
	<-ctx.Done()
	nameServer.DeleteName(serviceName, addr)
	nameServer.Close()
}

在服务端拦截器位置修改代码

interceptor.go

func UnaryInterceptor(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (resp interface{}, err error) {
	fmt.Println("server UnaryInterceptor")
	fmt.Println(info)
	if info.FullMethod != "/grpc.health.v1.Health/Check" { // 健康检查不验证token
		err = oauth2Valid(ctx) // 添加身份校验
		if err != nil {
			return nil, err
		}
	}
	// 所谓的拦截器就是中间件，所以最后一定要调用handler
	return handler(ctx, req)
}

通过健康检查

1
2

./grpc_health_probe-linux-amd64 -addr=localhost:50021 -tls -tls-ca-cert ../x509/ca_cert.pem -tls-client-cert ../x509/client_cert.pem -tls-client-key ../x509/client_key.pem -tls-server-name echo.test.example.com
# status: SERVING

2024-12-06 该篇文章被 Cleofwine 归为分类: Go云原生

高性能RPC框架gRPC

使用Protobuf定义服务

一元请求与流式请求

服务端

客户端

header与trailer元数据传输

客户端元数据

服务端元数据

客户端设置元数据和获取元数据

一元请求

流式调用

服务端设置元数据和获取元数据

一元请求

流式调用

TLS与双向TLS确保通讯安全

生成证书

服务端

客户端

OAuth2认证与拦截器

服务端

客户端

Keepalive连接保活配置

服务端

客户端

Resolver服务名称解析器

客户端添加简单名称解析器

自定义实现名称解析服务器

实现名称解析服务器

定义服务

实现名称服务器

服务器注册服务

客户端查询服务

客户端Client连接池实现

客户端

多路复用与健康检查