14-RPC-自研微服务框架

RPC

RPC 框架是分布式领域核心组件，也是微服务的基础。

RPC （Remote Procedure Call）全称是远程过程调用，相对于本地方法调用，在同一内存空间可以直接通过方法栈实现调用，远程调用则跨了不同的服务终端，并不能直接调用。

RPC框架 要解决的就是远程方法调用的问题，并且实现调用远程服务像调用本地服务一样简单，框架内部封装实现了网络调用的细节。

在这里插入图片描述

1. 通信协议选择

根据不同的需求来选择通信协议，UDP是不可靠传输，一般来说很少做为RPC框架的选择。

TCP和HTTP是最佳选择。

HTTP虽然有很多无用的头部信息，传输效率上会比较低，但是HTTP通用性更强，跨语言，跨平台，更易移植。

TCP可靠传输，需要自定义协议，传输效率更高，但是通用性不强。

1.1 HTTP/1.0和HTTP/1.1的区别

HTTP1.0最早在网页中使用是在1996年，那个时候只是使用一些较为简单的网页上和网络请求上，而HTTP1.1则在1999年才开始广泛应用于现在的各大浏览器网络请求中，同时HTTP1.1也是当前使用最为广泛的HTTP协议。 主要区别主要体现在：

  1. 缓存处理，在HTTP1.0中主要使用header里的If-Modified-Since,Expires来做为缓存判断的标准，HTTP1.1则引入了更多的缓存控制策略例如Entity tag，If-Unmodified-Since, If-Match, If-None-Match等更多可供选择的缓存头来控制缓存策略。
  2. 带宽优化及网络连接的使用，HTTP1.0中，存在一些浪费带宽的现象，例如客户端只是需要某个对象的一部分，而服务器却将整个对象送过来了，并且不支持断点续传功能，HTTP1.1则在请求头引入了range头域，它允许只请求资源的某个部分，即返回码是206（Partial Content），这样就方便了开发者自由的选择以便于充分利用带宽和连接。
  3. 错误通知的管理，在HTTP1.1中新增了24个错误状态响应码，如409（Conflict）表示请求的资源与资源的当前状态发生冲突；410（Gone）表示服务器上的某个资源被永久性的删除。
  4. Host头处理，在HTTP1.0中认为每台服务器都绑定一个唯一的IP地址，因此，请求消息中的URL并没有传递主机名（hostname）。但随着虚拟主机技术的发展，在一台物理服务器上可以存在多个虚拟主机（Multi-homed Web Servers），并且它们共享一个IP地址。HTTP1.1的请求消息和响应消息都应支持Host头域，且请求消息中如果没有Host头域会报告一个错误（400 Bad Request）。
  5. 长连接，HTTP 1.1支持长连接（PersistentConnection）和请求的流水线（Pipelining）处理，在一个TCP连接上可以传送多个HTTP请求和响应，减少了建立和关闭连接的消耗和延迟，在HTTP1.1中默认开启Connection： keep-alive，一定程度上弥补了HTTP1.0每次请求都要创建连接的缺点。

1.2 HTTP/1.1和HTTP/2的区别

  • 新的二进制格式（Binary Format），HTTP1.x的解析是基于文本。基于文本协议的格式解析存在天然缺陷，文本的表现形式有多样性，要做到健壮性考虑的场景必然很多，二进制则不同，只认0和1的组合。基于这种考虑HTTP2.0的协议解析决定采用二进制格式，实现方便且健壮。
  • 多路复用（MultiPlexing），即连接共享，即每一个request都是是用作连接共享机制的。一个request对应一个id，这样一个连接上可以有多个request，每个连接的request可以随机的混杂在一起，接收方可以根据request的 id将request再归属到各自不同的服务端请求里面。
  • header压缩，如上文中所言，对前面提到过HTTP1.x的header带有大量信息，而且每次都要重复发送，HTTP2.0使用encoder来减少需要传输的header大小，通讯双方各自cache一份header fields表，既避免了重复header的传输，又减小了需要传输的大小。
  • 服务端推送（server push）HTTP2.0也具有server push功能。

grpc采用了http2协议，由于http的通用性，所以现在的几乎所有的rpc框架都支持grpc

2. 序列化协议

数据在网络中传输，必须是二进制的，所以我们需要先将传输的对象进行序列化之后，才能传输。

接收方通过反序列化将数据解析出来。

序列化协议有XML、 JSON、Protobuf、Thrift 等，Golang 原生支持的 Gob 协议。

3. 编解码

如果使用TCP，我们需要定义数据传输的格式，防止在传输过程中出现的粘包，拆包等问题。

在这里插入图片描述

假设客户端分别发送了两个数据包D1和D2给服务端，由于服务端一次读取到字节数是不确定的，故可能存在以下四种情况：

  1. 服务端分两次读取到了两个独立的数据包，分别是D1和D2，没有粘包和拆包
  2. 服务端一次接受到了两个数据包，D1和D2粘合在一起，称之为TCP粘包
  3. 服务端分两次读取到了数据包，第一次读取到了完整的D1包和D2包的部分内容，第二次读取到了D2包的剩余内容，这称之为TCP拆包
  4. 服务端分两次读取到了数据包，第一次读取到了D1包的部分内容D1_1，第二次读取到了D1包的剩余部分内容D1_2和完整的D2包。

特别要注意的是，如果TCP的接受滑窗非常小，而数据包D1和D2比较大，很有可能会发生第五种情况，即服务端分多次才能将D1和D2包完全接收，期间发生多次拆包。

自定义格式可以使用定长的头和不定长的体，标识数据长度即可。

1字节1字节4字节1字节1字节1字节8字节不定
魔法数（Magic Number）版本（Version）消息长度（full length）消息类型（messageType）压缩类型（compress）序列化类型（serialize）请求id（requestId）请求体（body）
  • magic number : 通信双方协商的一个暗号 魔数的作用是用于服务端在接收数据时先解析出魔数做正确性对比。如果和协议中的魔数不匹配，则认为是非法数据
  • version : 不同版本的协议对应的解析方法可能是不同的，应对业务变化需求
  • full length： 记录了整个消息的长度
  • messageType：普通请求、普通响应、心跳等，根据消息类型做出不同的解析
  • compress： 序列化的字节流，还可以进行压缩，使得体积更小，在网络传输更快，不一定要使用
  • serialize：序列化方式，比如json，protostuff，glob等
  • request id：每个请求分配好请求Id，这样响应数据的时候，才能对的上
  • body：具体的数据

4. 实现

4.1 http方式

package rpc

import (
	"bufio"
	"bytes"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net/http"
	"net/url"
	"strings"
	"time"
)

type MsHttpClient struct {
	client http.Client
}

// NewHttpClient Transport请求分发，协程安全，支持连接池s
func NewHttpClient() *MsHttpClient {
	client := http.Client{
		Timeout: time.Duration(3) * time.Second,
		Transport: &http.Transport{
			MaxIdleConnsPerHost:   5,
			MaxConnsPerHost:       100,
			IdleConnTimeout:       90 * time.Second,
			TLSHandshakeTimeout:   10 * time.Second,
			ExpectContinueTimeout: 1 * time.Second,
		},
	}
	return &MsHttpClient{client: client}
}

func (c *MsHttpClient) GetRequest(method string, url string, args map[string]any) (*http.Request, error) {
	if args != nil && len(args) > 0 {
		url = url + "?" + c.toValues(args)
	}
	req, err := http.NewRequest(method, url, nil)
	if err != nil {
		return nil, err
	}
	return req, nil
}

func (c *MsHttpClient) FormRequest(method string, url string, args map[string]any) (*http.Request, error) {
	req, err := http.NewRequest(method, url, strings.NewReader(c.toValues(args)))
	if err != nil {
		return nil, err
	}
	return req, nil
}

func (c *MsHttpClient) JsonRequest(method string, url string, args map[string]any) (*http.Request, error) {
	jsonStr, _ := json.Marshal(args)
	req, err := http.NewRequest(method, url, bytes.NewReader(jsonStr))
	if err != nil {
		return nil, err
	}
	return req, nil
}

func (c *MsHttpClient) Get(url string, args map[string]any) ([]byte, error) {
	if args != nil && len(args) > 0 {
		url = url + "?" + c.toValues(args)
	}
	req, err := http.NewRequest("GET", url, nil)
	if err != nil {
		return nil, err
	}
	return c.handleResponse(req)
}

func (c *MsHttpClient) PostForm(url string, args map[string]any) ([]byte, error) {
	req, err := http.NewRequest("POST", url, strings.NewReader(c.toValues(args)))
	if err != nil {
		return nil, err
	}
	return c.handleResponse(req)
}

func (c *MsHttpClient) PostJson(url string, args map[string]any) ([]byte, error) {
	jsonStr, _ := json.Marshal(args)
	req, err := http.NewRequest("POST", url, bytes.NewReader(jsonStr))
	if err != nil {
		return nil, err
	}
	return c.handleResponse(req)
}

func (c *MsHttpClient) Response(req *http.Request) ([]byte, error) {
	return c.handleResponse(req)
}
func (c *MsHttpClient) handleResponse(req *http.Request) ([]byte, error) {
	var err error
	response, err := c.client.Do(req)
	if err != nil {
		return nil, err
	}
	if response.StatusCode != 200 {
		return nil, errors.New(response.Status)
	}
	buffLen := 79
	buff := make([]byte, buffLen)
	body := make([]byte, 0)
	reader := bufio.NewReader(response.Body)
	for {
		n, err := reader.Read(buff)
		if err == io.EOF || n == 0 {
			break
		}
		body = append(body, buff[:n]...)
		if n < buffLen {
			break
		}
	}
	defer response.Body.Close()
	if err != nil {
		return nil, err
	}
	return body, nil
}

func (c *MsHttpClient) toValues(args map[string]any) string {
	if args != nil && len(args) > 0 {
		params := url.Values{}
		for k, v := range args {
			params.Set(k, fmt.Sprintf("%v", v))
		}
		return params.Encode()
	}
	return ""
}

ordercenter：

package main

import (
	"encoding/json"
	"fmt"
	"github.com/mszlu521/msgo"
	"github.com/mszlu521/msgo/rpc"
	"net/http"
)

type Result struct {
	Code int    `json:"code"`
	Msg  string `json:"msg"`
	Data any    `json:"data"`
}

type Goods struct {
	Id   int64  `json:"id"`
	Name string `json:"name"`
}

func main() {

	engine := msgo.Default()
	client := rpc.NewHttpClient()
	g := engine.Group("order")
	g.Get("/find", func(ctx *msgo.Context) {
		//查询商品
		bytes, err := client.Get("http://localhost:9002/goods/find", nil)
		if err != nil {
			ctx.Logger.Error(err)
		}
		fmt.Println(string(bytes))
		v := &Result{}
		json.Unmarshal(bytes, v)
		ctx.JSON(http.StatusOK, v)
	})
	engine.Run(":9003")
}

goodsCenter：

package main

import (
	"github.com/mszlu521/msgo"
	"net/http"
)

type Result struct {
	Code int    `json:"code"`
	Msg  string `json:"msg"`
	Data any    `json:"data"`
}

type Goods struct {
	Id   int64  `json:"id"`
	Name string `json:"name"`
}

func main() {

	engine := msgo.Default()
	g := engine.Group("goods")
	g.Get("/find", func(ctx *msgo.Context) {
		//查询商品
		goods := Goods{Id: 1000, Name: "商品中心9001商品"}
		ctx.JSON(http.StatusOK, &Result{Code: 200, Msg: "success", Data: goods})
	})
	engine.Run(":9002")
}

4.2 改造http方式

config:

package rpc

import "strconv"

type Config struct {
	Protocol string
	Host     string
	Port     int
	Ssl      bool
}

func (c Config) Url() string {
	switch c.Protocol {
	case HTTP, HTTP2:
		prefix := "http://"
		if c.Ssl {
			prefix = "https://"
		}
		return prefix + c.Host + ":" + strconv.FormatInt(int64(c.Port), 10)
	}
	return ""
}

const (
	HTTP  = "HTTP"
	HTTP2 = "HTTP2"
	TCP   = "TCP"
)

const (
	GET      = "GET"
	POSTForm = "POST_FORM"
	POSTJson = "POST_JSON"
)

rpc.go:

package rpc

type MsService interface {
	Env() Config
}



func (c *MsHttpClient) Use(name string, s MsService) {
	if c.serviceMap == nil {
		c.serviceMap = make(map[string]MsService)
	}
	c.serviceMap[name] = s
}

func (c *MsHttpClient) Do(name string, method string) MsService {
	s, ok := c.serviceMap[name]
	if !ok {
		panic(errors.New(name + " not exist, please action"))
	}
	t := reflect.TypeOf(s)
	v := reflect.ValueOf(s)
	if t.Kind() != reflect.Pointer {
		panic(errors.New("service must be pointer"))
	}
	tVar := t.Elem()
	vVar := v.Elem()
	findIndex := -1
	for i := 0; i < tVar.NumField(); i++ {
		field := tVar.Field(i)
		name := field.Name
		if method == name {
			findIndex = i
		}
	}
	if findIndex == -1 {
		panic(errors.New(method + " not exist"))
	}
	requestPath := tVar.Field(findIndex).Tag.Get("msrpc")
	if requestPath == "" {
		panic(errors.New("msrpc tag not exist"))
	}
	split := strings.Split(requestPath, ",")
	mt := split[0]
	path := split[1]
	co := s.Env()
	prefix := co.Url()
	f := func(args map[string]any) ([]byte, error) {
		if mt == GET {
			return c.Get(prefix+path, args)
		}
		if mt == POSTForm {
			return c.PostForm(prefix+path, args)
		}
		if mt == POSTJson {
			return c.PostJson(prefix+path, args)
		}
		return nil, nil
	}
	value := reflect.ValueOf(f)
	vVar.Field(findIndex).Set(value)
	return s
}

goods:

package service

import (
	"github.com/mszlu521/msgo/rpc"
)

type Goods struct {
	Id   int64  `json:"id"`
	Name string `json:"name"`
}

type GoodsService struct {
	Find func(args map[string]any) ([]byte, error) `msrpc:"GET,/goods/find"`
}

func (r *GoodsService) Env() rpc.Config {
	c := rpc.Config{
		Host:     "localhost",
		Port:     9002,
		Protocol: rpc.HTTP,
	}
	return c
}


package main

import (
	"encoding/json"
	"fmt"
	"github.com/mszlu521/msgo"
	"github.com/mszlu521/msgo/rpc"
	"github.com/mszlu521/ordercenter/model"
	"github.com/mszlu521/ordercenter/service"
	"net/http"
)

func main() {
	engine := msgo.Default()
	client := rpc.NewHttpClient()
	g := engine.Group("order")
	goodsService := &service.GoodsService{}
	client.Use("goodsService", goodsService)
	g.Get("/find", func(ctx *msgo.Context) {
		//查询商品
		v := &model.Result{}
		bytes, err := client.Do("goodsService", "Find").(*service.GoodsService).Find(nil)
		if err != nil {
			ctx.Logger.Error(err)
		}
		fmt.Println(string(bytes))
		json.Unmarshal(bytes, v)
		ctx.JSON(http.StatusOK, v)
	})
	engine.Run(":9003")
}

通过上述改造，我们可以比较轻易的使用框架，来实现http方式的rpc调用

记住：框架的目的是易用，但同时需要遵守规则，所以定义规则也是框架的一部分

4.3 http2（grpc）方式

有关grpc的使用可以先去看教程，教程地址

go get google.golang.org/grpc

protoc  --go_out=./ --go-grpc_out=./  .\api\goods.proto

goodscenter服务端：

syntax = "proto3";

//import "google/protobuf/any.proto";

option go_package="/api";

package api;

service GoodsApi {
  rpc Find(GoodsRequest) returns (GoodsResponse);
}

message GoodsRequest {

}

message GoodsResponse {
  int64 Code = 1;
  string Msg = 2;
  Goods Data = 3;
}

message Goods {
  int64 Id = 1;
  string Name = 2;
}

package service

import (
	"context"
	"github.com/mszlu521/goodscenter/api"
)

type GoodsApiService struct {
}

func (GoodsApiService) Find(context.Context, *api.GoodsRequest) (*api.GoodsResponse, error) {
	goods := &api.Goods{Id: 1000, Name: "商品中心9002商品,grpc提供"}
	res := &api.GoodsResponse{
		Code: 200,
		Msg:  "success",
		Data: goods,
	}
	return res, nil
}
func (GoodsApiService) mustEmbedUnimplementedGoodsApiServer() {}

grpc服务端：

listen, _ := net.Listen("tcp", ":9111")
	server := grpc.NewServer()
	api.RegisterGoodsApiServer(server, &api.GoodsApiService{})
	err := server.Serve(listen)
	log.Println(err)

grpc客户端:

g.Get("/findGrpc", func(ctx *msgo.Context) {
		//查询商品
		var serviceHost = "127.0.0.1:9111"
		conn, err := grpc.Dial(serviceHost, grpc.WithTransportCredentials(insecure.NewCredentials()))
		if err != nil {
			fmt.Println(err)
		}
		defer conn.Close()

		client := api.NewGoodsApiClient(conn)
		rsp, err := client.Find(context.TODO(), &api.GoodsRequest{})

		if err != nil {
			fmt.Println(err)
		}
		ctx.JSON(http.StatusOK, rsp)
	})
4.3.1 形成框架工具

服务端：

package rpc

import (
	"google.golang.org/grpc"
	"net"
)

type MsGrpcServer struct {
	listen     net.Listener
	grpcServer *grpc.Server
	registers  []func(grpcServer *grpc.Server)
	ops        []grpc.ServerOption
}

func NewGrpcServer(address string, ops ...MsGrpcOption) (*MsGrpcServer, error) {
	listen, err := net.Listen("tcp", address)
	if err != nil {
		return nil, err
	}
	ms := &MsGrpcServer{
		listen: listen,
	}
	for _, op := range ops {
		op.Apply(ms)
	}
	s := grpc.NewServer(ms.ops...)
	ms.grpcServer = s
	return ms, nil
}

func (s *MsGrpcServer) Run() error {
	for _, register := range s.registers {
		register(s.grpcServer)
	}
	return s.grpcServer.Serve(s.listen)
}

func (s *MsGrpcServer) Register(register func(grpServer *grpc.Server)) {
	s.registers = append(s.registers, register)
}

type MsGrpcOption interface {
	Apply(s *MsGrpcServer)
}

type DefaultGrpcOption struct {
	f func(s *MsGrpcServer)
}

func (d DefaultGrpcOption) Apply(s *MsGrpcServer) {
	d.f(s)
}

func WithGrpcOptions(options ...grpc.ServerOption) MsGrpcOption {
	return DefaultGrpcOption{f: func(s *MsGrpcServer) {
		s.ops = append(s.ops, options...)
	}}
}


   grpcServer, _ := rpc.NewGrpcServer(":9111")
	grpcServer.Register(func(grpServer *grpc.Server) {
		api.RegisterGoodsApiServer(grpServer, &api.GoodsApiService{})
	})
	err := grpcServer.Run()


type MsGrpcClient struct {
	Conn *grpc.ClientConn
}

func NewGrpcClient(config *MsGrpcClientConfig) (*MsGrpcClient, error) {
	var ctx = context.Background()
	var dialOptions = config.dialOptions

	if config.Block {
		//阻塞
		if config.DialTimeout > time.Duration(0) {
			var cancel context.CancelFunc
			ctx, cancel = context.WithTimeout(ctx, config.DialTimeout)
			defer cancel()
		}
		dialOptions = append(dialOptions, grpc.WithBlock())
	}
	if config.KeepAlive != nil {
		dialOptions = append(dialOptions, grpc.WithKeepaliveParams(*config.KeepAlive))
	}
	conn, err := grpc.DialContext(ctx, config.Address, dialOptions...)
	if err != nil {
		return nil, err
	}
	return &MsGrpcClient{
		Conn: conn,
	}, nil
}

type MsGrpcClientConfig struct {
	Address     string
	Block       bool
	DialTimeout time.Duration
	ReadTimeout time.Duration
	Direct      bool
	KeepAlive   *keepalive.ClientParameters
	dialOptions []grpc.DialOption
}

func DefaultGrpcClientConfig() *MsGrpcClientConfig {
	return &MsGrpcClientConfig{
		dialOptions: []grpc.DialOption{
			grpc.WithTransportCredentials(insecure.NewCredentials()),
		},
		DialTimeout: time.Second * 3,
		ReadTimeout: time.Second * 2,
		Block:       true,
	}
}

4.4 TCP方式

tcp方式就需要实现序列化，编解码等操作了

序列化协议支持两种：

Protobuf 和 go的Gob协议。

4.4.1 server端

type Serializer interface {
	Serialize(i interface{}) ([]byte, error)
	Deserialize(data []byte, i interface{}) error
}
type GobSerializer struct{}

func (c GobSerializer) Serialize(data any) ([]byte, error) {
	var buffer bytes.Buffer
	encoder := gob.NewEncoder(&buffer)
	if err := encoder.Encode(data); err != nil {
		return nil, err
	}
	return buffer.Bytes(), nil
}

func (c GobSerializer) Deserialize(data []byte, target any) error {
	buffer := bytes.NewBuffer(data)
	decoder := gob.NewDecoder(buffer)
	return decoder.Decode(target)
}

type MsRpcMessage struct {
	//头
	Header *Header
	//消息体
	Data any
}

const mn byte = 0x1d
const version = 0x01

type CompressType byte

const (
	Gzip CompressType = iota
)

type SerializeType byte

const (
	Gob SerializeType = iota
	ProtoBuff
)

type MessageType byte

const (
	msgRequest MessageType = iota
	msgResponse
	msgPing
	msgPong
)

type Header struct {
	MagicNumber   byte
	Version       byte
	FullLength    int32
	MessageType   MessageType
	CompressType  CompressType
	SerializeType SerializeType
	RequestId     int64
}

type MsRpcRequest struct {
	RequestId   int64
	ServiceName string
	MethodName  string
	Args        []any
}

type MsRpcResponse struct {
	RequestId     int64
	Code          int16
	Msg           string
	CompressType  CompressType
	SerializeType SerializeType
	Data          any
}

type MsRpcServer interface {
	Register(name string, service interface{})
	Run()
	Stop()
}

type MsTcpServer struct {
	listener   net.Listener
	Host       string
	Port       int
	Network    string
	serviceMap map[string]interface{}
}

type MsTcpConn struct {
	s       *MsTcpServer
	conn    net.Conn
	rspChan chan *MsRpcResponse
}

func (c *MsTcpConn) writeHandle() {
	ctx := context.Background()
	_, cancel := context.WithTimeout(ctx, time.Duration(3)*time.Second)
	defer cancel()
	select {
	case rsp := <-c.rspChan:
		//编码数据
		err := c.Send(c.conn, rsp)
		if err != nil {
			log.Println(err)
		}
		return
	case <-ctx.Done():
		log.Println("超时了")
		return
	}
}

func (c *MsTcpConn) Send(conn net.Conn, rsp *MsRpcResponse) error {
	headers := make([]byte, 17)
	//magic number
	headers[0] = mn
	//version
	headers[1] = version
	//full length
	//消息类型
	headers[6] = byte(msgResponse)
	//压缩类型
	headers[7] = byte(rsp.CompressType)
	//序列化
	headers[8] = byte(rsp.SerializeType)
	//请求id
	binary.BigEndian.PutUint64(headers[9:], uint64(rsp.RequestId))

	serializer, err := loadSerialize(rsp.SerializeType)
	if err != nil {
		return err
	}
	body, err := serializer.Serialize(rsp)
	if err != nil {
		return err
	}
	body, err = compress(body, rsp.CompressType)
	if err != nil {
		return err
	}
	fullLen := 17 + len(body)
	binary.BigEndian.PutUint32(headers[2:6], uint32(fullLen))
	_, err = conn.Write(headers[:])
	if err != nil {
		return err
	}
	err = binary.Write(c.conn, binary.BigEndian, body[:])
	if err != nil {
		return err
	}
	log.Println("发送数据成功")
	return nil
}

func NewTcpServer(host string, port int) *MsTcpServer {
	return &MsTcpServer{
		Host:    host,
		Port:    port,
		Network: "tcp",
	}
}
func (s *MsTcpServer) Register(name string, service interface{}) {
	if s.serviceMap == nil {
		s.serviceMap = make(map[string]interface{})
	}
	v := reflect.ValueOf(service)
	if v.Kind() != reflect.Pointer {
		panic(errors.New("service not pointer"))
	}
	s.serviceMap[name] = service
}
func (s *MsTcpServer) Run() {
	addr := fmt.Sprintf("%s:%d", s.Host, s.Port)
	listen, err := net.Listen(s.Network, addr)
	if err != nil {
		panic(err)
	}
	s.listener = listen
	for {
		conn, err := s.listener.Accept()
		if err != nil {
			log.Println(err)
			continue
		}
		msConn := &MsTcpConn{conn: conn, rspChan: make(chan *MsRpcResponse, 1), s: s}
		go s.readHandle(msConn)
		go msConn.writeHandle()
	}
}

func (s *MsTcpServer) readHandle(msConn *MsTcpConn) {
	defer func() {
		if err := recover(); err != nil {
			log.Println(err)
			msConn.conn.Close()
		}
	}()
	msg := s.decodeFrame(msConn.conn)
	if msg == nil {
		msConn.rspChan <- nil
		return
	}
	//根据请求
	if msg.Header.MessageType == msgRequest {
		req := msg.Data.(*MsRpcRequest)
		//查找注册的服务匹配后进行调用，调用完发送到一个channel当中
		service, ok := s.serviceMap[req.ServiceName]
		rsp := &MsRpcResponse{RequestId: req.RequestId, CompressType: msg.Header.CompressType, SerializeType: msg.Header.SerializeType}
		if !ok {
			rsp.Code = 500
			rsp.Msg = "no service found"
			msConn.rspChan <- rsp
			return
		}
		v := reflect.ValueOf(service)
		reflectMethod := v.MethodByName(req.MethodName)
		args := make([]reflect.Value, len(req.Args))
		for i := range req.Args {
			args[i] = reflect.ValueOf(req.Args[i])
		}
		result := reflectMethod.Call(args)
		if len(result) == 0 {
			//无返回结果
			rsp.Code = 200
			msConn.rspChan <- rsp
			return
		}
		resArgs := make([]interface{}, len(result))
		for i := 0; i < len(result); i++ {
			resArgs[i] = result[i].Interface()
		}
		var err error
		if _, ok := result[len(result)-1].Interface().(error); ok {
			err = result[len(result)-1].Interface().(error)
		}

		if err != nil {
			rsp.Code = 500
			rsp.Msg = err.Error()
		}
		rsp.Code = 200
		rsp.Data = resArgs[0]
		msConn.rspChan <- rsp
		log.Println("接收数据成功")
		return
	}
}

func (s *MsTcpServer) Close() {
	if s.listener != nil {
		s.listener.Close()
	}
}

func (*MsTcpServer) decodeFrame(conn net.Conn) *MsRpcMessage {
	//读取数据 先读取header部分
	//1+1+4+1+1+1+8 = 17字节
	headers := make([]byte, 17)
	_, err := io.ReadFull(conn, headers)
	if err != nil {
		log.Println(err)
		return nil
	}
	//magic number
	magicNumber := headers[0]
	if magicNumber != mn {
		log.Println("magic number not valid : ", magicNumber)
		return nil
	}
	//version
	version := headers[1]
	//
	fullLength := headers[2:6]
	//
	mt := headers[6]
	messageType := MessageType(mt)
	//压缩类型
	compressType := headers[7]
	//序列化类型
	serializeType := headers[8]
	//请求id
	requestId := headers[9:]

	//将body解析出来，包装成request 根据请求内容查找对应的服务，完成调用
	//网络调用 大端
	fl := int32(binary.BigEndian.Uint32(fullLength))
	bodyLen := fl - 17
	body := make([]byte, bodyLen)
	_, err = io.ReadFull(conn, body)
	log.Println("读完了")
	if err != nil {
		log.Println(err)
		return nil
	}
	//先解压
	body, err = unCompress(body, CompressType(compressType))
	if err != nil {
		log.Println(err)
		return nil
	}
	//反序列化
	serializer, err := loadSerialize(SerializeType(serializeType))
	if err != nil {
		log.Println(err)
		return nil
	}
	header := &Header{}
	header.MagicNumber = magicNumber
	header.FullLength = fl
	header.CompressType = CompressType(compressType)
	header.Version = version
	header.SerializeType = SerializeType(serializeType)
	header.RequestId = int64(binary.BigEndian.Uint64(requestId))
	header.MessageType = messageType

	if messageType == msgRequest {
		msg := &MsRpcMessage{}
		msg.Header = header
		req := &MsRpcRequest{}
		err := serializer.Deserialize(body, req)
		if err != nil {
			log.Println(err)
			return nil
		}
		msg.Data = req
		return msg
	}
	if messageType == msgResponse {
		msg := &MsRpcMessage{}
		msg.Header = header
		rsp := &MsRpcResponse{}
		err := serializer.Deserialize(body, rsp)
		if err != nil {
			log.Println(err)
			return nil
		}
		msg.Data = rsp
		return msg
	}
	return nil
}

func loadSerialize(serializeType SerializeType) (Serializer, error) {
	switch serializeType {
	case Gob:
		//gob
		s := &GobSerializer{}
		return s, nil
	}
	return nil, errors.New("no serializeType")
}

func compress(body []byte, compressType CompressType) ([]byte, error) {
	switch compressType {
	case Gzip:
		//return body, nil
		//gzip
		//创建一个新的 byte 输出流
		var buf bytes.Buffer
		w := gzip.NewWriter(&buf)

		_, err := w.Write(body)
		if err != nil {
			return nil, err
		}
		if err := w.Close(); err != nil {
			return nil, err
		}
		return buf.Bytes(), nil
	}
	return nil, errors.New("no compressType")
}

func unCompress(body []byte, compressType CompressType) ([]byte, error) {
	switch compressType {
	case Gzip:
		//return body, nil
		//gzip
		reader, err := gzip.NewReader(bytes.NewReader(body))
		defer reader.Close()
		if err != nil {
			return nil, err
		}
		buf := new(bytes.Buffer)
		// 从 Reader 中读取出数据
		if _, err := buf.ReadFrom(reader); err != nil {
			return nil, err
		}
		return buf.Bytes(), nil
	}
	return nil, errors.New("no compressType")
}

tcpServer := rpc.NewTcpServer("localhost", 9112)
	gob.Register(&model.Result{})
	gob.Register(&model.Goods{})
	tcpServer.Register("goods", &service.GoodsRpcService{})
	go tcpServer.Run()
	go engine.Run(":9002")
	quit := make(chan os.Signal)
	signal.Notify(quit, syscall.SIGINT, syscall.SIGTERM, syscall.SIGHUP, syscall.SIGQUIT)
	<-quit
	tcpServer.Close()

package service

import (
	"github.com/mszlu521/goodscenter/model"
)

type GoodsRpcService struct {
}

func (*GoodsRpcService) Find(id int64) *model.Result {
	goods := model.Goods{Id: 1000, Name: "商品中心9002商品"}
	return &model.Result{Code: 200, Msg: "success", Data: goods}
}
4.4.2 client端

type MsRpcClient interface {
	Connect() error
	Invoke(context context.Context, serviceName string, methodName string, args []any) (any, error)
	Close() error
}

type MsTcpClient struct {
	conn   net.Conn
	option TcpClientOption
}

type TcpClientOption struct {
	Retries           int
	ConnectionTimeout time.Duration
	SerializeType     SerializeType
	CompressType      CompressType
	Host              string
	Port              int
}

var DefaultOption = TcpClientOption{
	Host:              "127.0.0.1",
	Port:              9112,
	Retries:           3,
	ConnectionTimeout: 5 * time.Second,
	SerializeType:     Gob,
	CompressType:      Gzip,
}

func NewTcpClient(option TcpClientOption) *MsTcpClient {
	return &MsTcpClient{option: option}
}

func (c *MsTcpClient) Connect() error {
	addr := fmt.Sprintf("%s:%d", c.option.Host, c.option.Port)
	conn, err := net.DialTimeout("tcp", addr, c.option.ConnectionTimeout)
	if err != nil {
		return err
	}
	c.conn = conn
	return nil
}

var reqId int64

func (c *MsTcpClient) Invoke(ctx context.Context, serviceName string, methodName string, args []any) (any, error) {
	var cancel context.CancelFunc
	ctx, cancel = context.WithTimeout(ctx, c.option.ConnectionTimeout)
	defer cancel()

	req := &MsRpcRequest{}
	req.RequestId = atomic.AddInt64(&reqId, 1)
	req.ServiceName = serviceName
	req.MethodName = methodName
	req.args = args

	headers := make([]byte, 17)
	//magic number
	headers[0] = mn
	//version
	headers[1] = version
	//full length
	//消息类型
	headers[6] = byte(msgRequest)
	//压缩类型
	headers[7] = byte(c.option.CompressType)
	//序列化
	headers[8] = byte(c.option.SerializeType)
	//请求id
	binary.BigEndian.PutUint64(headers[9:], uint64(req.RequestId))

	serializer, err := loadSerialize(c.option.SerializeType)
	if err != nil {
		return nil, err
	}
	body, err := serializer.Serialize(req)
	if err != nil {
		return nil, err
	}
	body, err = compress(body, c.option.CompressType)
	if err != nil {
		return nil, err
	}
	fullLen := 17 + len(body)
	binary.BigEndian.PutUint32(headers[2:6], uint32(fullLen))
	_, err = c.conn.Write(headers[:])
	if err != nil {
		return nil, err
	}
	err = binary.Write(c.conn, binary.BigEndian, body[:])
	if err != nil {
		return nil, err
	}
	rspChan := make(chan *MsRpcResponse)
	go c.readHandle(rspChan)
	rsp := <-rspChan
	return rsp, nil
}

func (c *MsTcpClient) Close() error {
	if c.conn != nil {
		return c.conn.Close()
	}
	return nil
}

func (c *MsTcpClient) readHandle(rspChan chan *MsRpcResponse) {
	defer func() {
		if err := recover(); err != nil {
			log.Println(err)
			c.conn.Close()
		}
	}()
	for {
		msg := c.decodeFrame(c.conn)
		if msg == nil {
			log.Println("未解析出任何数据")
			rspChan <- nil
			return
		}
		//根据请求
		if msg.Header.MessageType == msgResponse {
			rsp := msg.Data.(*MsRpcResponse)
			rspChan <- rsp
			return
		}
	}
}

func (*MsTcpClient) decodeFrame(conn net.Conn) *MsRpcMessage {
	//读取数据 先读取header部分
	//1+1+4+1+1+1+8 = 17字节
	headers := make([]byte, 17)
	_, err := io.ReadFull(conn, headers)
	if err != nil {
		log.Println(err)
		return nil
	}
	//magic number
	magicNumber := headers[0]
	if magicNumber != mn {
		log.Println("magic number not valid : ", magicNumber)
		return nil
	}
	//version
	version := headers[1]
	//
	fullLength := headers[2:6]
	//
	mt := headers[6]
	messageType := MessageType(mt)
	//压缩类型
	compressType := headers[7]
	//序列化类型
	serializeType := headers[8]
	//请求id
	requestId := headers[9:]

	//将body解析出来，包装成request 根据请求内容查找对应的服务，完成调用
	//网络调用 大端
	fl := int32(binary.BigEndian.Uint32(fullLength))
	bodyLen := fl - 17
	body := make([]byte, bodyLen)
	_, err = io.ReadFull(conn, body)
	log.Println("读完了")
	if err != nil {
		log.Println(err)
		return nil
	}
	//先解压
	body, err = unCompress(body, CompressType(compressType))
	if err != nil {
		log.Println(err)
		return nil
	}
	//反序列化
	serializer, err := loadSerialize(SerializeType(serializeType))
	if err != nil {
		log.Println(err)
		return nil
	}
	header := &Header{}
	header.MagicNumber = magicNumber
	header.FullLength = fl
	header.CompressType = CompressType(compressType)
	header.Version = version
	header.SerializeType = SerializeType(serializeType)
	header.RequestId = int64(binary.BigEndian.Uint64(requestId))
	header.MessageType = messageType

	if messageType == msgRequest {
		msg := &MsRpcMessage{}
		msg.Header = header
		req := &MsRpcRequest{}
		err := serializer.Deserialize(body, req)
		if err != nil {
			log.Println(err)
			return nil
		}
		msg.Data = req
		return msg
	}
	if messageType == msgResponse {
		msg := &MsRpcMessage{}
		msg.Header = header
		rsp := &MsRpcResponse{}
		err := serializer.Deserialize(body, rsp)
		if err != nil {
			log.Println(err)
			return nil
		}
		msg.Data = rsp
		return msg
	}
	return nil
}

type MsTcpClientProxy struct {
	client *MsTcpClient
	option TcpClientOption
}

func NewMsTcpClientProxy(option TcpClientOption) *MsTcpClientProxy {
	return &MsTcpClientProxy{option: option}
}

func (p *MsTcpClientProxy) Call(ctx context.Context, serviceName string, methodName string, args []any) (any, error) {
	client := NewTcpClient(p.option)
	p.client = client
	err := client.Connect()
	if err != nil {
		return nil, err
	}
	for i := 0; i < p.option.Retries; i++ {
		result, err := client.Invoke(ctx, serviceName, methodName, args)
		if err != nil {
			if i >= p.option.Retries-1 {
				log.Println(errors.New("already retry all time"))
				client.Close()
				return nil, err
			}
			continue
		}
		client.Close()
		return result, nil
	}
	return nil, errors.New("retry time is 0")
}


g.Get("/findTcp", func(ctx *msgo.Context) {
		//查询商品
		gob.Register(&model.Result{})
		gob.Register(&model.Goods{})
		args := make([]any, 1)
		args[0] = 1
		result, err := proxy.Call(context.Background(), "goods", "Find", args)
		if err != nil {
			panic(err)
		}
		ctx.JSON(http.StatusOK, result)
	})
4.4.3 protobuf序列化支持

type ProtobufSerializer struct{}

func (c ProtobufSerializer) Serialize(data any) ([]byte, error) {
	marshal, err := proto.Marshal(data.(proto.Message))
	if err != nil {
		return nil, err
	}
	return marshal, nil
}

func (c ProtobufSerializer) Deserialize(data []byte, target any) error {
	message := target.(proto.Message)
	return proto.Unmarshal(data, message)
}

protoc  --go_out=./ --go-grpc_out=./  .\rpc\tcp.proto 

syntax = "proto3";

import "google/protobuf/struct.proto";

option go_package="/rpc";

package rpc;

message Request {
  int64 RequestId = 1;
  string ServiceName = 2;
  string MethodName = 3;
  repeated google.protobuf.Value Args = 4;
}

message Response {
  int64 RequestId = 1;
  int32 Code = 2;
  string Msg = 3;
  int32 CompressType = 4;
  int32 SerializeType = 5;
  google.protobuf.Value Data = 6;
}

package rpc

import (
	"bytes"
	"compress/gzip"
	"context"
	"encoding/binary"
	"encoding/gob"
	"encoding/json"
	"errors"
	"fmt"
	"google.golang.org/protobuf/proto"
	"google.golang.org/protobuf/types/known/structpb"
	"io"
	"log"
	"net"
	"reflect"
	"sync/atomic"
	"time"
)

type Serializer interface {
	Serialize(i interface{}) ([]byte, error)
	Deserialize(data []byte, i interface{}) error
}
type GobSerializer struct{}

func (c GobSerializer) Serialize(data any) ([]byte, error) {
	var buffer bytes.Buffer
	encoder := gob.NewEncoder(&buffer)
	if err := encoder.Encode(data); err != nil {
		return nil, err
	}
	return buffer.Bytes(), nil
}

func (c GobSerializer) Deserialize(data []byte, target any) error {
	buffer := bytes.NewBuffer(data)
	decoder := gob.NewDecoder(buffer)
	return decoder.Decode(target)
}

type ProtobufSerializer struct{}

func (c ProtobufSerializer) Serialize(data any) ([]byte, error) {
	marshal, err := proto.Marshal(data.(proto.Message))
	if err != nil {
		return nil, err
	}
	return marshal, nil
}

func (c ProtobufSerializer) Deserialize(data []byte, target any) error {
	message := target.(proto.Message)
	return proto.Unmarshal(data, message)
}

type MsRpcMessage struct {
	//头
	Header *Header
	//消息体
	Data any
}

const mn byte = 0x1d
const version = 0x01

type CompressType byte

const (
	Gzip CompressType = iota
)

type SerializeType byte

const (
	Gob SerializeType = iota
	ProtoBuff
)

type MessageType byte

const (
	msgRequest MessageType = iota
	msgResponse
	msgPing
	msgPong
)

type Header struct {
	MagicNumber   byte
	Version       byte
	FullLength    int32
	MessageType   MessageType
	CompressType  CompressType
	SerializeType SerializeType
	RequestId     int64
}

type MsRpcRequest struct {
	RequestId   int64
	ServiceName string
	MethodName  string
	Args        []any
}

type MsRpcResponse struct {
	RequestId     int64
	Code          int16
	Msg           string
	CompressType  CompressType
	SerializeType SerializeType
	Data          any
}

type MsRpcServer interface {
	Register(name string, service interface{})
	Run()
	Stop()
}

type MsTcpServer struct {
	listener   net.Listener
	Host       string
	Port       int
	Network    string
	serviceMap map[string]interface{}
}

type MsTcpConn struct {
	s       *MsTcpServer
	conn    net.Conn
	rspChan chan *MsRpcResponse
}

func (c *MsTcpConn) writeHandle() {
	ctx := context.Background()
	_, cancel := context.WithTimeout(ctx, time.Duration(3)*time.Second)
	defer cancel()
	select {
	case rsp := <-c.rspChan:
		//编码数据
		err := c.Send(c.conn, rsp)
		if err != nil {
			log.Println(err)
		}
		return
	case <-ctx.Done():
		log.Println("超时了")
		return
	}
}

func (c *MsTcpConn) Send(conn net.Conn, rsp *MsRpcResponse) error {
	headers := make([]byte, 17)
	//magic number
	headers[0] = mn
	//version
	headers[1] = version
	//full length
	//消息类型
	headers[6] = byte(msgResponse)
	//压缩类型
	headers[7] = byte(rsp.CompressType)
	//序列化
	headers[8] = byte(rsp.SerializeType)
	//请求id
	binary.BigEndian.PutUint64(headers[9:], uint64(rsp.RequestId))

	serializer, err := loadSerialize(SerializeType(rsp.SerializeType))
	if err != nil {
		return err
	}
	var body []byte
	if ProtoBuff == rsp.SerializeType {
		pRsp := &Response{}
		pRsp.SerializeType = int32(rsp.SerializeType)
		pRsp.CompressType = int32(rsp.CompressType)
		pRsp.Code = int32(rsp.Code)
		pRsp.Msg = rsp.Msg
		pRsp.RequestId = rsp.RequestId
		//value, err := structpb.
		//	log.Println(err)
		m := make(map[string]any)
		marshal, _ := json.Marshal(rsp.Data)
		_ = json.Unmarshal(marshal, &m)
		value, err := structpb.NewStruct(m)
		log.Println(err)
		pRsp.Data = structpb.NewStructValue(value)
		body, err = serializer.Serialize(pRsp)
	} else {
		body, err = serializer.Serialize(rsp)
	}
	if err != nil {
		return err
	}
	body, err = compress(body, CompressType(rsp.CompressType))
	if err != nil {
		return err
	}
	fullLen := 17 + len(body)
	binary.BigEndian.PutUint32(headers[2:6], uint32(fullLen))
	_, err = conn.Write(headers[:])
	if err != nil {
		return err
	}
	err = binary.Write(c.conn, binary.BigEndian, body[:])
	if err != nil {
		return err
	}
	log.Println("发送数据成功")
	return nil
}

func NewTcpServer(host string, port int) *MsTcpServer {
	return &MsTcpServer{
		Host:    host,
		Port:    port,
		Network: "tcp",
	}
}
func (s *MsTcpServer) Register(name string, service interface{}) {
	if s.serviceMap == nil {
		s.serviceMap = make(map[string]interface{})
	}
	v := reflect.ValueOf(service)
	if v.Kind() != reflect.Pointer {
		panic(errors.New("service not pointer"))
	}
	s.serviceMap[name] = service
}
func (s *MsTcpServer) Run() {
	addr := fmt.Sprintf("%s:%d", s.Host, s.Port)
	listen, err := net.Listen(s.Network, addr)
	if err != nil {
		panic(err)
	}
	s.listener = listen
	for {
		conn, err := s.listener.Accept()
		if err != nil {
			log.Println(err)
			continue
		}
		msConn := &MsTcpConn{conn: conn, rspChan: make(chan *MsRpcResponse, 1), s: s}
		go s.readHandle(msConn)
		go msConn.writeHandle()
	}
}

func (s *MsTcpServer) readHandle(msConn *MsTcpConn) {
	defer func() {
		if err := recover(); err != nil {
			log.Println(err)
			msConn.conn.Close()
		}
	}()
	msg := s.decodeFrame(msConn.conn)
	if msg == nil {
		msConn.rspChan <- nil
		return
	}
	//根据请求
	if msg.Header.MessageType == msgRequest {
		req := msg.Data.(*Request)
		//查找注册的服务匹配后进行调用，调用完发送到一个channel当中
		service, ok := s.serviceMap[req.ServiceName]
		rsp := &MsRpcResponse{RequestId: req.RequestId, CompressType: msg.Header.CompressType, SerializeType: msg.Header.SerializeType}
		if !ok {
			rsp.Code = 500
			rsp.Msg = "no service found"
			msConn.rspChan <- rsp
			return
		}
		v := reflect.ValueOf(service)
		reflectMethod := v.MethodByName(req.MethodName)
		args := make([]reflect.Value, len(req.Args))
		for i := range req.Args {
			of := reflect.ValueOf(req.Args[i].AsInterface())
			of = of.Convert(reflectMethod.Type().In(i))
			args[i] = of
		}
		result := reflectMethod.Call(args)
		if len(result) == 0 {
			//无返回结果
			rsp.Code = 200
			msConn.rspChan <- rsp
			return
		}
		resArgs := make([]interface{}, len(result))
		for i := 0; i < len(result); i++ {
			resArgs[i] = result[i].Interface()
		}
		var err error
		if _, ok := result[len(result)-1].Interface().(error); ok {
			err = result[len(result)-1].Interface().(error)
		}

		if err != nil {
			rsp.Code = 500
			rsp.Msg = err.Error()
		}
		rsp.Code = 200
		rsp.Data = resArgs[0]
		msConn.rspChan <- rsp
		log.Println("接收数据成功")
		return
	}
}

func (s *MsTcpServer) Close() {
	if s.listener != nil {
		s.listener.Close()
	}
}

func (*MsTcpServer) decodeFrame(conn net.Conn) *MsRpcMessage {
	//读取数据 先读取header部分
	//1+1+4+1+1+1+8 = 17字节
	headers := make([]byte, 17)
	_, err := io.ReadFull(conn, headers)
	if err != nil {
		log.Println(err)
		return nil
	}
	//magic number
	magicNumber := headers[0]
	if magicNumber != mn {
		log.Println("magic number not valid : ", magicNumber)
		return nil
	}
	//version
	version := headers[1]
	//
	fullLength := headers[2:6]
	//
	mt := headers[6]
	messageType := MessageType(mt)
	//压缩类型
	compressType := headers[7]
	//序列化类型
	serializeType := headers[8]
	//请求id
	requestId := headers[9:]

	//将body解析出来，包装成request 根据请求内容查找对应的服务，完成调用
	//网络调用 大端
	fl := int32(binary.BigEndian.Uint32(fullLength))
	bodyLen := fl - 17
	body := make([]byte, bodyLen)
	_, err = io.ReadFull(conn, body)
	log.Println("读完了")
	if err != nil {
		log.Println(err)
		return nil
	}
	//先解压
	body, err = unCompress(body, CompressType(compressType))
	if err != nil {
		log.Println(err)
		return nil
	}
	//反序列化
	serializer, err := loadSerialize(SerializeType(serializeType))
	if err != nil {
		log.Println(err)
		return nil
	}
	header := &Header{}
	header.MagicNumber = magicNumber
	header.FullLength = fl
	header.CompressType = CompressType(compressType)
	header.Version = version
	header.SerializeType = SerializeType(serializeType)
	header.RequestId = int64(binary.BigEndian.Uint64(requestId))
	header.MessageType = messageType

	if messageType == msgRequest {
		msg := &MsRpcMessage{}
		msg.Header = header
		if ProtoBuff == SerializeType(serializeType) {
			req := &Request{}
			err := serializer.Deserialize(body, req)
			if err != nil {
				log.Println(err)
				return nil
			}
			msg.Data = req
		} else {
			req := &MsRpcRequest{}
			err := serializer.Deserialize(body, req)
			if err != nil {
				log.Println(err)
				return nil
			}
			msg.Data = req
		}
		return msg
	}
	if messageType == msgResponse {
		msg := &MsRpcMessage{}
		msg.Header = header
		if ProtoBuff == SerializeType(serializeType) {
			rsp := &Response{}
			err := serializer.Deserialize(body, rsp)
			if err != nil {
				log.Println(err)
				return nil
			}
			msg.Data = rsp
		} else {
			rsp := &MsRpcResponse{}
			err := serializer.Deserialize(body, rsp)
			if err != nil {
				log.Println(err)
				return nil
			}
			msg.Data = rsp
		}
		return msg
	}
	return nil
}

func loadSerialize(serializeType SerializeType) (Serializer, error) {
	switch serializeType {
	case Gob:
		//gob
		s := &GobSerializer{}
		return s, nil
	case ProtoBuff:
		s := &ProtobufSerializer{}
		return s, nil
	}
	return nil, errors.New("no serializeType")
}

func compress(body []byte, compressType CompressType) ([]byte, error) {
	switch compressType {
	case Gzip:
		//return body, nil
		//gzip
		//创建一个新的 byte 输出流
		var buf bytes.Buffer
		w := gzip.NewWriter(&buf)

		_, err := w.Write(body)
		if err != nil {
			return nil, err
		}
		if err := w.Close(); err != nil {
			return nil, err
		}
		return buf.Bytes(), nil
	}
	return nil, errors.New("no compressType")
}

func unCompress(body []byte, compressType CompressType) ([]byte, error) {
	switch compressType {
	case Gzip:
		//return body, nil
		//gzip
		reader, err := gzip.NewReader(bytes.NewReader(body))
		defer reader.Close()
		if err != nil {
			return nil, err
		}
		buf := new(bytes.Buffer)
		// 从 Reader 中读取出数据
		if _, err := buf.ReadFrom(reader); err != nil {
			return nil, err
		}
		return buf.Bytes(), nil
	}
	return nil, errors.New("no compressType")
}

type MsRpcClient interface {
	Connect() error
	Invoke(context context.Context, serviceName string, methodName string, args []any) (any, error)
	Close() error
}

type MsTcpClient struct {
	conn   net.Conn
	option TcpClientOption
}

type TcpClientOption struct {
	Retries           int
	ConnectionTimeout time.Duration
	SerializeType     SerializeType
	CompressType      CompressType
	Host              string
	Port              int
}

var DefaultOption = TcpClientOption{
	Host:              "127.0.0.1",
	Port:              9112,
	Retries:           3,
	ConnectionTimeout: 5 * time.Second,
	SerializeType:     Gob,
	CompressType:      Gzip,
}

func NewTcpClient(option TcpClientOption) *MsTcpClient {
	return &MsTcpClient{option: option}
}

func (c *MsTcpClient) Connect() error {
	addr := fmt.Sprintf("%s:%d", c.option.Host, c.option.Port)
	conn, err := net.DialTimeout("tcp", addr, c.option.ConnectionTimeout)
	if err != nil {
		return err
	}
	c.conn = conn
	return nil
}

var reqId int64

func (c *MsTcpClient) Invoke(ctx context.Context, serviceName string, methodName string, args []any) (any, error) {
	var cancel context.CancelFunc
	ctx, cancel = context.WithTimeout(ctx, c.option.ConnectionTimeout)
	defer cancel()

	req := &MsRpcRequest{}
	req.RequestId = atomic.AddInt64(&reqId, 1)
	req.ServiceName = serviceName
	req.MethodName = methodName
	req.Args = args

	headers := make([]byte, 17)
	//magic number
	headers[0] = mn
	//version
	headers[1] = version
	//full length
	//消息类型
	headers[6] = byte(msgRequest)
	//压缩类型
	headers[7] = byte(c.option.CompressType)
	//序列化
	headers[8] = byte(c.option.SerializeType)
	//请求id
	binary.BigEndian.PutUint64(headers[9:], uint64(req.RequestId))

	serializer, err := loadSerialize(c.option.SerializeType)
	if err != nil {
		return nil, err
	}
	var body []byte
	if ProtoBuff == c.option.SerializeType {
		pReq := &Request{}
		pReq.RequestId = atomic.AddInt64(&reqId, 1)
		pReq.ServiceName = serviceName
		pReq.MethodName = methodName
		list, err := structpb.NewList(args)
		log.Println(err)
		pReq.Args = list.Values
		body, err = serializer.Serialize(pReq)
	} else {
		body, err = serializer.Serialize(req)
	}
	fmt.Println(body)
	if err != nil {
		return nil, err
	}
	log.Println(body)
	body, err = compress(body, c.option.CompressType)
	if err != nil {
		return nil, err
	}
	fullLen := 17 + len(body)
	binary.BigEndian.PutUint32(headers[2:6], uint32(fullLen))
	_, err = c.conn.Write(headers[:])
	if err != nil {
		return nil, err
	}
	log.Println(body)
	log.Println("len:", len(body))
	err = binary.Write(c.conn, binary.BigEndian, body[:])
	if err != nil {
		return nil, err
	}
	rspChan := make(chan *MsRpcResponse)
	go c.readHandle(rspChan)
	rsp := <-rspChan
	return rsp, nil
}

func (c *MsTcpClient) Close() error {
	if c.conn != nil {
		return c.conn.Close()
	}
	return nil
}

func (c *MsTcpClient) readHandle(rspChan chan *MsRpcResponse) {
	defer func() {
		if err := recover(); err != nil {
			log.Println(err)
			c.conn.Close()
		}
	}()
	for {
		msg := c.decodeFrame(c.conn)
		if msg == nil {
			log.Println("未解析出任何数据")
			rspChan <- nil
			return
		}
		//根据请求
		if msg.Header.MessageType == msgResponse {
			if msg.Header.SerializeType == ProtoBuff {
				rsp := msg.Data.(*Response)
				asInterface := rsp.Data.AsInterface()
				marshal, _ := json.Marshal(asInterface)
				rsp1 := &MsRpcResponse{}
				json.Unmarshal(marshal, rsp1)
				rspChan <- rsp1
			} else {
				rsp := msg.Data.(*MsRpcResponse)
				rspChan <- rsp
			}
			return
		}
	}
}

func (*MsTcpClient) decodeFrame(conn net.Conn) *MsRpcMessage {
	//读取数据 先读取header部分
	//1+1+4+1+1+1+8 = 17字节
	headers := make([]byte, 17)
	_, err := io.ReadFull(conn, headers)
	if err != nil {
		log.Println(err)
		return nil
	}
	//magic number
	magicNumber := headers[0]
	if magicNumber != mn {
		log.Println("magic number not valid : ", magicNumber)
		return nil
	}
	//version
	version := headers[1]
	//
	fullLength := headers[2:6]
	//
	mt := headers[6]
	messageType := MessageType(mt)
	//压缩类型
	compressType := headers[7]
	//序列化类型
	serializeType := headers[8]
	//请求id
	requestId := headers[9:]

	//将body解析出来，包装成request 根据请求内容查找对应的服务，完成调用
	//网络调用 大端
	fl := int32(binary.BigEndian.Uint32(fullLength))
	bodyLen := fl - 17
	body := make([]byte, bodyLen)
	_, err = io.ReadFull(conn, body)
	log.Println("读完了")
	if err != nil {
		log.Println(err)
		return nil
	}
	//先解压
	body, err = unCompress(body, CompressType(compressType))
	if err != nil {
		log.Println(err)
		return nil
	}
	//反序列化
	serializer, err := loadSerialize(SerializeType(serializeType))
	if err != nil {
		log.Println(err)
		return nil
	}
	header := &Header{}
	header.MagicNumber = magicNumber
	header.FullLength = fl
	header.CompressType = CompressType(compressType)
	header.Version = version
	header.SerializeType = SerializeType(serializeType)
	header.RequestId = int64(binary.BigEndian.Uint64(requestId))
	header.MessageType = messageType

	if messageType == msgRequest {
		msg := &MsRpcMessage{}
		msg.Header = header
		if ProtoBuff == SerializeType(serializeType) {
			req := &Request{}
			err := serializer.Deserialize(body, req)
			if err != nil {
				log.Println(err)
				return nil
			}
			msg.Data = req
		} else {
			req := &MsRpcRequest{}
			err := serializer.Deserialize(body, req)
			if err != nil {
				log.Println(err)
				return nil
			}
			msg.Data = req
		}
		return msg
	}
	if messageType == msgResponse {
		msg := &MsRpcMessage{}
		msg.Header = header
		if ProtoBuff == SerializeType(serializeType) {
			rsp := &Response{}
			err := serializer.Deserialize(body, rsp)
			if err != nil {
				log.Println(err)
				return nil
			}
			msg.Data = rsp
		} else {
			rsp := &MsRpcResponse{}
			err := serializer.Deserialize(body, rsp)
			if err != nil {
				log.Println(err)
				return nil
			}
			msg.Data = rsp
		}
		return msg
	}
	return nil
}

type MsTcpClientProxy struct {
	client *MsTcpClient
	option TcpClientOption
}

func NewMsTcpClientProxy(option TcpClientOption) *MsTcpClientProxy {
	return &MsTcpClientProxy{option: option}
}

func (p *MsTcpClientProxy) Call(ctx context.Context, serviceName string, methodName string, args []any) (any, error) {
	client := NewTcpClient(p.option)
	p.client = client
	err := client.Connect()
	if err != nil {
		return nil, err
	}
	for i := 0; i < p.option.Retries; i++ {
		result, err := client.Invoke(ctx, serviceName, methodName, args)
		if err != nil {
			if i >= p.option.Retries-1 {
				log.Println(errors.New("already retry all time"))
				client.Close()
				return nil, err
			}
			continue
		}
		client.Close()
		return result, nil
	}
	return nil, errors.New("retry time is 0")
}

对rpc做了初步实现，属于简单实现，并没有处理更为复杂的心跳，超时，连接管理等，需要大家自行去完善

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：/a/429139.html

如若内容造成侵权/违法违规/事实不符，请联系我们进行投诉反馈qq邮箱809451989@qq.com，一经查实，立即删除！

相关文章

智能通用平台（Intelligent General-purpose Platform）

智能通用平台（Intelligent General-purpose Platform）

根据2024年的最新人工智能技术发展趋势&#xff0c;我为您提出的项目需求表如下&#xff1a; 项目名称&#xff1a;智能通用平台&#xff08;Intelligent General-purpose Platform&#xff09;项目概述&#xff1a;结合最新的生成式人工智能、多模态学习和量子计算技术&#…
阅读更多...
Redis基础---Java客户端应用

Redis基础---Java客户端应用

目录 一、介绍 二、Jedis的使用 三、SpringDataRedis的使用 创建&#xff1a; 一、介绍 在Redis官网&#xff0c;提供了多种编程语言的客户端&#xff0c;如Java、C等&#xff0c;官网地址&#xff1a;Clients | Redis 而对于Java的客户端有很多&#xff0c;但是用的最多的就是…
阅读更多...
判定MySQL是否开机自启

判定MySQL是否开机自启

在Linux系统中&#xff0c;判断MySQL是否设置为开机自动启动&#xff0c;可以通过以下几种方式来确认&#xff1a; 使用systemd&#xff1a; 如果你的Linux发行版使用systemd作为其初始化系统&#xff08;大多数现代Linux发行版如Ubuntu、CentOS 7&#xff09;&#xff0c;可以…
阅读更多...
【Web前端入门学习】——HTML

【Web前端入门学习】——HTML

目录 HTML简介HTML文件结构常用文本标签标题标签段落标签有序列表和无序列表表格标签 HTML属性a标签—超链接标签图片标签 HTML区块块元素与行内元素 HTML表单 HTML简介 HTML全称是Hypertext Markup Language超文本标记语言。 HTML的作用&#xff1a; 为网页提供结构&#xff…
阅读更多...
自动驾驶感知面试-coding应用题

自动驾驶感知面试-coding应用题

感知面试手撕代码&#xff1a;这个博主总结的很好&#xff0c;尤其是关于叉积的计算 双线性插值 双线性插值公式记忆方法和Python实现 NMS算法 #include<iostream> #include<vector> #include<algorithm>using namespace std; struct Box {int x1,x2,y1,…
阅读更多...
【代码】Python3｜无GUI环境中使用Seaborn作图的学习路线及代码（阴影折线图）

【代码】Python3｜无GUI环境中使用Seaborn作图的学习路线及代码（阴影折线图）

我有个需求是需要画图&#xff0c;让GPT帮我生成了一下学习计划。 学习路线依照GPT的来的&#xff0c;使用的Prompt工具是https://github.com/JushBJJ/Mr.-Ranedeer-AI-Tutor。 文章目录 PrerequisiteMain Curriculum1.1 Seaborn介绍Seaborn基础保存图形为文件练习 1.2 单变量数…
阅读更多...
什么是Vue的服务端渲染（SSR）？它有什么作用？

什么是Vue的服务端渲染（SSR）？它有什么作用？

Vue的服务端渲染&#xff08;SSR&#xff09;是指将Vue组件在服务器端进行渲染&#xff0c;然后将已经渲染好的页面返回给浏览器&#xff0c;相比于传统的客户端渲染&#xff0c;SSR可以更好地优化SEO和加速首屏加载速度。在传统的客户端渲染中&#xff0c;浏览器需要加载所有的…
阅读更多...
【Spring云原生】Spring官宣，干掉原生JVM，推出 Spring Native！整体提升性能！Native镜像技术在Spring中的应用

【Spring云原生】Spring官宣，干掉原生JVM，推出 Spring Native！整体提升性能！Native镜像技术在Spring中的应用

&#x1f389;&#x1f389;欢迎光临&#x1f389;&#x1f389; &#x1f3c5;我是苏泽&#xff0c;一位对技术充满热情的探索者和分享者。&#x1f680;&#x1f680; &#x1f31f;特别推荐给大家我的最新专栏《Spring 狂野之旅&#xff1a;从入门到入魔》 &#x1f680; 本…
阅读更多...
Linux——MySQL主从复制与读写分离

Linux——MySQL主从复制与读写分离

实验环境 虚拟机 3台 centos7.9 网卡NAT模式 数量 1 组件包mysql-5.6.36.tar.gz cmake-2.8.6.tar.gz 设备 IP 备注 Centos01 192.168.223.123 Amoeba Centos02 192.168.223.124 Master Centos03 192.168.223.125 Slave MySQL安装 主从同时操作 安装所需要的…
阅读更多...
【探索AI】二十二 深度学习之第5周：生成对抗网络（GAN）

【探索AI】二十二 深度学习之第5周：生成对抗网络（GAN）

生成对抗网络&#xff08;GAN&#xff09; GAN的基本原理与结构常见的GAN模型&#xff08;如DCGAN、StyleGAN等&#xff09;图像生成与风格迁移任务实践&#xff1a;使用GAN进行图像生成或风格迁移任务 GAN的基本原理与结构 生成对抗网络&#xff08;GAN&#xff09;的基本原理…
阅读更多...
【经验】f-string 的一些点

【经验】f-string 的一些点

【经验】f-string 的一些点 省几个字别数错了对齐它现在几点 省几个字 让 f-string 给你写表达式&#xff0c;在 f-string 中使用 来自动打印表达式 a 10 b 25 print(f"{a b }") >>> a b 35别数错了 对于过大的数字难以一眼看出来它的数量级&#xf…
阅读更多...
Android 中get请求网络数据 详细举例

Android 中get请求网络数据 详细举例

请求链接 https://api.bilibili.com/x/web-interface/ranking 1.添加网络权限 依赖等 implementation com.squareup.okhttp3:okhttp:4.9.3 implementation com.google.code.gson:gson:2.8.92.写请求类network package com.example.myapplication;import android.graphics.Bi…
阅读更多...
现代信号处理学习笔记（三）信号检测

现代信号处理学习笔记（三）信号检测

通过观测数据判断信号是否存在&#xff0c;这一问题称为信号检测。 目录 前言 一、统计假设检验 1、信号检测的基本概念 2、信号检测理论测度 比率测度 概率测度 3、决策理论空间 二、概率密度函数与误差函数 1、概率密度函数 2、误差函数与补余误差函数 三、检测概…
阅读更多...
TypeError: the JSON object must be str, bytes or bytearray, not dict

TypeError: the JSON object must be str, bytes or bytearray, not dict

参考文章&#xff1a;https://blog.csdn.net/yuan2019035055/article/details/124934362 Python基础系列&#xff08;一&#xff09;搞懂json数据解析与字典之间的关系 代码&#xff1a; 报错信息: TypeError: the JSON object must be str, bytes or bytearray, not dict …
阅读更多...
Python字符串中%d，%.f的用法

Python字符串中%d，%.f的用法

在字符串中&#xff0c;我们经常需要改变其中的一些几个字符&#xff0c;比如 第一个d%就是epoch1&#xff0c;会随着epoch的变化而变化。同理&#xff0c;后面的也是。打印出的结果就是 其中&#xff0c;epoch迭代了3次&#xff0c;从0开始。会发现 d%&#xff0c;是整数&…
阅读更多...
.NET高级面试指南专题十四【 观察者模式介绍，最常用的设计模式之一】

.NET高级面试指南专题十四【 观察者模式介绍，最常用的设计模式之一】

简介&#xff1a; 观察者模式&#xff08;Observer Pattern&#xff09;是一种行为型设计模式&#xff0c;其目的是定义了一种一对多的依赖关系&#xff0c;当一个对象的状态发生变化时&#xff0c;所有依赖于它的对象都会得到通知并自动更新。 原理&#xff1a; 在观察者模式中…
阅读更多...
002-CSS-三角形

002-CSS-三角形

三角形 普通三角形带阴影的三角形&#xff1a;box-shadow带阴影的三角形&#xff1a;filter 普通三角形 &#x1f4a1; Tips&#xff1a;设置 div 盒子宽高为零&#xff0c;使用单边框实现 .triangle {width: 0;height: 0;border: 20px solid transparent;border-top-color: …
阅读更多...
把python完全卸载干净

把python完全卸载干净

1.winR&#xff0c;输入control回车&#xff0c;点击程序和功能&#xff0c;在搜索框输入python&#xff0c;右键点击卸载 2、找到Python安装路径&#xff0c;把所有文件全部删除。 安装路径可以打开CMD输入&#xff1a;where python 3、强制删除Python.exe 打开cmd&#xff…
阅读更多...
使用Java生成JWT（JSON Web Token）的详细指南

使用Java生成JWT（JSON Web Token）的详细指南

介绍 在现代应用程序中&#xff0c;身份验证和授权是至关重要的。JSON Web Token&#xff08;JWT&#xff09;是一种开放标准&#xff08;RFC 7519&#xff09;&#xff0c;它定义了一种紧凑且自包含的方式用于在各方之间安全地传输信息。在本文中&#xff0c;我们将学习如何使…
阅读更多...
大厂报价查询系统性能优化之道！

大厂报价查询系统性能优化之道！

0 前言 机票查询系统&#xff0c;日均亿级流量&#xff0c;要求高吞吐&#xff0c;低延迟架构设计。提升缓存的效率以及实时计算模块长尾延迟&#xff0c;成为制约机票查询系统性能关键。本文介绍机票查询系统在缓存和实时计算两个领域的架构提升。 1 机票搜索服务概述 1.1 …
阅读更多...
最新文章

Copyright @ 2022~2023