溫馨提示×
您好，登錄后才能下訂單哦！
點擊 登錄注冊 即表示同意《億速云用戶服務條款》
您好，登錄后才能下訂單哦！
本篇內容介紹了“Golang中tinyrpc框架怎么使用”的有關知識，在實際案例的操作過程中，不少人都會遇到這樣的困境，接下來就讓小編帶領大家學習一下如何處理這些情況吧！希望大家仔細閱讀，能夠學有所成！
tinyrpc基于TCP協議，支持各種壓縮格式，基于protocol buffer的序列化協議。其rpc是基于golang原生的net/rpc開發而成。
tinyrpc基于net/rpc開發而成，在此基礎上集成了額外的能力。項目結構如圖：
功能目錄如下：
codec 編碼模塊
compressor 壓縮模塊
header 請求/響應頭模塊
protoc-gen-tinyrpc 代碼生成插件
serializer 序列化模塊
客戶端是以net/rpc的rpc.Client為基礎構建，在此基礎上定義了Option以配置壓縮方式和序列化方式:
type Option func(o *options)
type options struct {
compressType compressor.CompressType
serializer serializer.Serializer
}在創建客戶端的時候將配置好的壓縮算法和序列化方式作為創建客戶端的參數：
func NewClient(conn io.ReadWriteCloser, opts ...Option) *Client {
options := options{
compressType: compressor.Raw,
serializer: serializer.Proto,
}
for _, option := range opts {
option(&options)
}
return &Client{rpc.NewClientWithCodec(
codec.NewClientCodec(conn, options.compressType, options.serializer))}
}服務端是以net/rpc的rpc.Server為基礎構建，在此基礎上擴展了Server的定義：
type Server struct {
*rpc.Server
serializer.Serializer
}在創建客戶端和開啟服務時傳入序列化方式：
func NewServer(opts ...Option) *Server {
options := options{
serializer: serializer.Proto,
}
for _, option := range opts {
option(&options)
}
return &Server{&rpc.Server{}, options.serializer}
}
func (s *Server) Serve(lis net.Listener) {
log.Printf("tinyrpc started on: %s", lis.Addr().String())
for {
conn, err := lis.Accept()
if err != nil {
continue
}
go s.Server.ServeCodec(codec.NewServerCodec(conn, s.Serializer))
}
}壓縮算法的實現中首先是定義了壓縮的接口：
type Compressor interface {
Zip([]byte) ([]byte, error)
Unzip([]byte) ([]byte, error)
}壓縮的接口包含壓縮和解壓方法。
壓縮算法使用的是uint類型，使用iota來初始化，并且使用map來進行所有壓縮算法實現的管理：
type CompressType uint16
const (
Raw CompressType = iota
Gzip
Snappy
Zlib
)
// Compressors which supported by rpc
var Compressors = map[CompressType]Compressor{
Raw: RawCompressor{},
Gzip: GzipCompressor{},
Snappy: SnappyCompressor{},
Zlib: ZlibCompressor{},
}序列化部分代碼非常簡單，提供了一個接口：
type Serializer interface {
Marshal(message interface{}) ([]byte, error)
Unmarshal(data []byte, message interface{}) error
}目前只有ProtoSerializer一個實現，ProtoSerializer內部的實現是基于"google.golang.org/protobuf/proto"來實現的，并沒有什么特殊的處理，因此就不花費筆墨詳述了。
tinyrpc定義了自己的請求頭和響應頭：
// RequestHeader request header structure looks like:
// +--------------+----------------+----------+------------+----------+
// | CompressType | Method | ID | RequestLen | Checksum |
// +--------------+----------------+----------+------------+----------+
// | uint16 | uvarint+string | uvarint | uvarint | uint32 |
// +--------------+----------------+----------+------------+----------+
type RequestHeader struct {
sync.RWMutex
CompressType compressor.CompressType
Method string
ID uint64
RequestLen uint32
Checksum uint32
}請求頭由壓縮類型，方法，id，請求長度和校驗碼組成。
// ResponseHeader request header structure looks like:
// +--------------+---------+----------------+-------------+----------+
// | CompressType | ID | Error | ResponseLen | Checksum |
// +--------------+---------+----------------+-------------+----------+
// | uint16 | uvarint | uvarint+string | uvarint | uint32 |
// +--------------+---------+----------------+-------------+----------+
type ResponseHeader struct {
sync.RWMutex
CompressType compressor.CompressType
ID uint64
Error string
ResponseLen uint32
Checksum uint32
}響應頭由壓縮類型，id，錯誤信息，返回長度和校驗碼組成。
為了實現頭的重用，tinyrpc為頭構建了緩存池：
var (
RequestPool sync.Pool
ResponsePool sync.Pool
)
func init() {
RequestPool = sync.Pool{New: func() interface{} {
return &RequestHeader{}
}}
ResponsePool = sync.Pool{New: func() interface{} {
return &ResponseHeader{}
}}
}在使用時get出來，生命周期結束后放回池子，并且在put之前需要進行重置：
h := header.RequestPool.Get().(*header.RequestHeader)
defer func() {
h.ResetHeader()
header.RequestPool.Put(h)
}()// ResetHeader reset request header
func (r *RequestHeader) ResetHeader() {
r.Lock()
defer r.Unlock()
r.ID = 0
r.Checksum = 0
r.Method = ""
r.CompressType = 0
r.RequestLen = 0
}
// ResetHeader reset response header
func (r *ResponseHeader) ResetHeader() {
r.Lock()
defer r.Unlock()
r.Error = ""
r.ID = 0
r.CompressType = 0
r.Checksum = 0
r.ResponseLen = 0
}搞清楚了頭的結構以及對象池的復用邏輯，那么具體的頭的編碼與解碼就是很簡單的拆裝工作，就不在此一行一行解析了，大家有興趣可以自行去閱讀。
由于tinyrpc是基于net/rpc開發，那么其codec模塊自然也是依賴于net/rpc的ClientCodec和ServerCodec接口來實現的。
客戶端是基于ClientCodec實現的能力：
type ClientCodec interface {
WriteRequest(*Request, any) error
ReadResponseHeader(*Response) error
ReadResponseBody(any) error
Close() error
}client定義了一個clientCodec類型，并且實現了ClientCodec的接口方法：
type clientCodec struct {
r io.Reader
w io.Writer
c io.Closer
compressor compressor.CompressType // rpc compress type(raw,gzip,snappy,zlib)
serializer serializer.Serializer
response header.ResponseHeader // rpc response header
mutex sync.Mutex // protect pending map
pending map[uint64]string
}WriteRequest實現：
// WriteRequest Write the rpc request header and body to the io stream
func (c *clientCodec) WriteRequest(r *rpc.Request, param interface{}) error {
c.mutex.Lock()
c.pending[r.Seq] = r.ServiceMethod
c.mutex.Unlock()
if _, ok := compressor.Compressors[c.compressor]; !ok {
return NotFoundCompressorError
}
reqBody, err := c.serializer.Marshal(param)
if err != nil {
return err
}
compressedReqBody, err := compressor.Compressors[c.compressor].Zip(reqBody)
if err != nil {
return err
}
h := header.RequestPool.Get().(*header.RequestHeader)
defer func() {
h.ResetHeader()
header.RequestPool.Put(h)
}()
h.ID = r.Seq
h.Method = r.ServiceMethod
h.RequestLen = uint32(len(compressedReqBody))
h.CompressType = compressor.CompressType(c.compressor)
h.Checksum = crc32.ChecksumIEEE(compressedReqBody)
if err := sendFrame(c.w, h.Marshal()); err != nil {
return err
}
if err := write(c.w, compressedReqBody); err != nil {
return err
}
c.w.(*bufio.Writer).Flush()
return nil
}可以看到代碼的實現還是比較清晰的，主要分為幾個步驟：
將數據進行序列化構成請求體
選擇相應的壓縮算法進行壓縮
從Pool中獲取請求頭實例將數據全部填入其中構成最后的請求頭
分別通過io操作發送處理過的請求頭和請求體
ReadResponseHeader實現：
// ReadResponseHeader read the rpc response header from the io stream
func (c *clientCodec) ReadResponseHeader(r *rpc.Response) error {
c.response.ResetHeader()
data, err := recvFrame(c.r)
if err != nil {
return err
}
err = c.response.Unmarshal(data)
if err != nil {
return err
}
c.mutex.Lock()
r.Seq = c.response.ID
r.Error = c.response.Error
r.ServiceMethod = c.pending[r.Seq]
delete(c.pending, r.Seq)
c.mutex.Unlock()
return nil
}此方法作用是讀取返回的響應頭，并解析成具體的結構體
ReadResponseBody實現：
func (c *clientCodec) ReadResponseBody(param interface{}) error {
if param == nil {
if c.response.ResponseLen != 0 {
if err := read(c.r, make([]byte, c.response.ResponseLen)); err != nil {
return err
}
}
return nil
}
respBody := make([]byte, c.response.ResponseLen)
err := read(c.r, respBody)
if err != nil {
return err
}
if c.response.Checksum != 0 {
if crc32.ChecksumIEEE(respBody) != c.response.Checksum {
return UnexpectedChecksumError
}
}
if c.response.GetCompressType() != c.compressor {
return CompressorTypeMismatchError
}
resp, err := compressor.Compressors[c.response.GetCompressType()].Unzip(respBody)
if err != nil {
return err
}
return c.serializer.Unmarshal(resp, param)
}此方法是用于讀取返回的響應結構體，流程如下：
讀取流獲取響應體
根據響應頭中的校驗碼來比對響應體是否完整
根據壓縮算法來解壓具體的結構體
進行反序列化
服務端是基于ServerCodec實現的能力：
type ServerCodec interface {
ReadRequestHeader(*Request) error
ReadRequestBody(any) error
WriteResponse(*Response, any) error
// Close can be called multiple times and must be idempotent.
Close() error
}和客戶端類似，server定義了一個serverCodec類型，并且實現了ServerCodec的接口方法：
type serverCodec struct {
r io.Reader
w io.Writer
c io.Closer
request header.RequestHeader
serializer serializer.Serializer
mutex sync.Mutex // protects seq, pending
seq uint64
pending map[uint64]*reqCtx
}ReadRequestHeader實現：
// ReadRequestHeader read the rpc request header from the io stream
func (s *serverCodec) ReadRequestHeader(r *rpc.Request) error {
s.request.ResetHeader()
data, err := recvFrame(s.r)
if err != nil {
return err
}
err = s.request.Unmarshal(data)
if err != nil {
return err
}
s.mutex.Lock()
s.seq++
s.pending[s.seq] = &reqCtx{s.request.ID, s.request.GetCompressType()}
r.ServiceMethod = s.request.Method
r.Seq = s.seq
s.mutex.Unlock()
return nil
}此方法用于讀取請求頭并解析成結構體
ReadRequestBody實現：
// ReadRequestBody read the rpc request body from the io stream
func (s *serverCodec) ReadRequestBody(param interface{}) error {
if param == nil {
if s.request.RequestLen != 0 {
if err := read(s.r, make([]byte, s.request.RequestLen)); err != nil {
return err
}
}
return nil
}
reqBody := make([]byte, s.request.RequestLen)
err := read(s.r, reqBody)
if err != nil {
return err
}
if s.request.Checksum != 0 {
if crc32.ChecksumIEEE(reqBody) != s.request.Checksum {
return UnexpectedChecksumError
}
}
if _, ok := compressor.
Compressors[s.request.GetCompressType()]; !ok {
return NotFoundCompressorError
}
req, err := compressor.
Compressors[s.request.GetCompressType()].Unzip(reqBody)
if err != nil {
return err
}
return s.serializer.Unmarshal(req, param)
}此方法用于讀取請求體，流程和讀取響應體差不多，大致如下：
讀取流并解析成請求體
根據請求頭中的校驗碼進行校驗
根據壓縮算法進行解壓
反序列化
WriteResponse實現：
// WriteResponse Write the rpc response header and body to the io stream
func (s *serverCodec) WriteResponse(r *rpc.Response, param interface{}) error {
s.mutex.Lock()
reqCtx, ok := s.pending[r.Seq]
if !ok {
s.mutex.Unlock()
return InvalidSequenceError
}
delete(s.pending, r.Seq)
s.mutex.Unlock()
if r.Error != "" {
param = nil
}
if _, ok := compressor.
Compressors[reqCtx.compareType]; !ok {
return NotFoundCompressorError
}
var respBody []byte
var err error
if param != nil {
respBody, err = s.serializer.Marshal(param)
if err != nil {
return err
}
}
compressedRespBody, err := compressor.
Compressors[reqCtx.compareType].Zip(respBody)
if err != nil {
return err
}
h := header.ResponsePool.Get().(*header.ResponseHeader)
defer func() {
h.ResetHeader()
header.ResponsePool.Put(h)
}()
h.ID = reqCtx.requestID
h.Error = r.Error
h.ResponseLen = uint32(len(compressedRespBody))
h.Checksum = crc32.ChecksumIEEE(compressedRespBody)
h.CompressType = reqCtx.compareType
if err = sendFrame(s.w, h.Marshal()); err != nil {
return err
}
if err = write(s.w, compressedRespBody); err != nil {
return err
}
s.w.(*bufio.Writer).Flush()
return nil
}此方法用于寫入響應體，大致與寫入請求體差不多，流程如下：
將響應體序列化
使用壓縮算法將響應體進行壓縮
使用Pool管理響應頭
分別發送返回頭和返回體
“Golang中tinyrpc框架怎么使用”的內容就介紹到這里了，感謝大家的閱讀。如果想了解更多行業相關的知識可以關注億速云網站，小編將為大家輸出更多高質量的實用文章！
免責聲明：本站發布的內容（圖片、視頻和文字）以原創、轉載和分享為主，文章觀點不代表本網站立場，如果涉及侵權請聯系站長郵箱：is@yisu.com進行舉報，并提供相關證據，一經查實，將立刻刪除涉嫌侵權內容。
