gRPC 原始碼詳解(一)配置化的結構體
grpc 原始碼結構詳解
DialOptions
DialOptions 是最重要的一環,負責配置每一次 rpc 請求的時候的一應選擇。
結構
先來看看這個的結構
// dialOptions configure a Dial call. dialOptions are set by the DialOption // values passed to Dial. type dialOptions struct { unaryInt UnaryClientInterceptor streamInt StreamClientInterceptor chainUnaryInts []UnaryClientInterceptor chainStreamInts []StreamClientInterceptor cp Compressor dc Decompressor bs backoff.Strategy block bool insecure bool timeout time.Duration scChan <-chan ServiceConfig authority string copts transport.ConnectOptions callOptions []CallOption // This is used by v1 balancer dial option WithBalancer to support v1 // balancer, and also by WithBalancerName dial option. balancerBuilder balancer.Builder // This is to support grpclb. resolverBuilder resolver.Builder channelzParentID int64 disableServiceConfig bool disableRetry bool disableHealthCheck bool healthCheckFunc internal.HealthChecker minConnectTimeout func() time.Duration defaultServiceConfig *ServiceConfig // defaultServiceConfig is parsed from defaultServiceConfigRawJSON. defaultServiceConfigRawJSON *string }
由於命名非常規範,加上註釋很容易看懂每一個 field 配置的哪一條屬性。如果掠過看的 大概有 壓縮解壓器,超時阻塞設定,認證安全轉發,負載均衡,服務持久化的資訊儲存 ,配置,心跳檢測等。
其一應函式方法都是設定 其中欄位的。
如何設定
這裡是 grpc 設計較好的地方,通過函式設定,同時設有生成函式的函式。什麼意思呢?首先結合圖來理解,這也是整個 grpc 設定的精華部分
這裡的意思是 , DialOptions 是一個匯出介面,實現函式是 apply 同時接受引數 dialOptions 來修改它。
而實際上,是使用 newFuncDialOption 函式包裝一個 修改 dialOptions 的方法給 funcDialOption 結構體,在實際 Dial 呼叫的時候 是使用閉包 呼叫 funcDialOption 結構體的 apply 方法。
可以在這裡看一下 Dial 方法的原始碼(Dial 呼叫的是 DialContext
起作用的就是 opt.apply()
func DialContext(ctx context.Context, target string, opts ...DialOption) (conn *ClientConn, err error) { cc := &ClientConn{ target: target, csMgr: &connectivityStateManager{}, conns: make(map[*addrConn]struct{}), dopts: defaultDialOptions(), blockingpicker: newPickerWrapper(), czData: new(channelzData), firstResolveEvent: grpcsync.NewEvent(), } ··· for _, opt := range opts { opt.apply(&cc.dopts) } ··· }
這裡的 options 可以說是 client 發起 rpc 請求的核心中轉站。
另一個重要的介面,同時也集中在 dialOptions 結構體中初始化處理的是
callOptions []CallOption
CallOption
CallOption 是一個介面,定義在 rpc_util 包內
結構
// CallOption configures a Call before it starts or extracts information from
// a Call after it completes.
type CallOption interface {
// before is called before the call is sent to any server. If before
// returns a non-nil error, the RPC fails with that error.
before(*callInfo) error
// after is called after the call has completed. after cannot return an
// error, so any failures should be reported via output parameters.
after(*callInfo)
}
操作的是 callInfo 結構裡的資料,其被包含在 dialOptions
結構體中,
即每一次 dial 的時候進行呼叫。
callInfo
// callInfo contains all related configuration and information about an RPC.
type callInfo struct {
compressorType string
failFast bool
stream ClientStream
maxReceiveMessageSize *int
maxSendMessageSize *int
creds credentials.PerRPCCredentials
contentSubtype string
codec baseCodec
maxRetryRPCBufferSize int
}
可以看到 callInfo 中欄位用來表示 單次呼叫中獨有的自定義選項如 壓縮,流控,認證,編解碼器等。
一個實現
簡單看一個 CallOption 介面的實現
// Header returns a CallOptions that retrieves the header metadata
// for a unary RPC.
func Header(md *metadata.MD) CallOption {
return HeaderCallOption{HeaderAddr: md}
}
// HeaderCallOption is a CallOption for collecting response header metadata.
// The metadata field will be populated *after* the RPC completes.
// This is an EXPERIMENTAL API.
type HeaderCallOption struct {
HeaderAddr *metadata.MD
}
func (o HeaderCallOption) before(c *callInfo) error { return nil }
func (o HeaderCallOption) after(c *callInfo) {
if c.stream != nil {
*o.HeaderAddr, _ = c.stream.Header()
}
}
重點看到,實際操作是在 before 和 after 方法中執行,它們會在 Client 發起請求的時候自動執行,顧名思義,一個在呼叫前執行,一個在呼叫後執行。
實現注意
這裡可以看出,這裡也是通過函式返回一個擁有這兩個方法的結構體,注意這一個設計,可以作為你自己的 Option 設計的時候的參考。
兩種方法
有兩種方法讓 Client 接受你的 CallOption 設定
- 在 Client 使用方法的時候直接作為 引數傳遞,將剛才所說的函式-返回一個實現了 CallOption 介面的結構體。
- 在 生成 Client 的時候就傳遞設定。具體如下
- 通過 dialOptions.go 中的 函式 grpc.WithDefaultCallOptions()
- 這個函式會將 CallOption 設定到 dialOptions 中的欄位 []CallOption 中。
// WithDefaultCallOptions returns a DialOption which sets the default
// CallOptions for calls over the connection.
func WithDefaultCallOptions(cos ...CallOption) DialOption {
return newFuncDialOption(func(o *dialOptions) {
o.callOptions = append(o.callOptions, cos...)
})
}
有沒有感覺有點不好理解?給你們一個例項
- 使用的第一種方法
response, err := myclient.MyCall(ctx, request, grpc.CallContentSubtype("mycodec"))
- 使用第二種方法
myclient := grpc.Dial(ctx, target, grpc.WithDefaultCallOptions(grpc.CallContentSubtype("mycodec")))
這裡假設 我們設定了一個 mycodec 的譯碼器。馬上下面解釋它的設計。
另
值得注意的是, 我好像只提到了在 Client 呼叫時設定,callOption 只在客戶端設定的情況是不是讓大家感到困惑。
實際上 gRPC server 端會自動檢測 callOption 的設定,並檢測自己是否支援此項選擇,如果不支援則會返回失敗。也就是說,在 Server 端註冊的所有 Codec 譯碼器之後,Client 直接使用相應的設定就好了。
Codec
在 gRPC 中 Codec 有兩個介面定義,一個是 baseCodec 包含正常的 Marshal 和 Unmarshal 方法,另一個是擁有名字的 Codec 定義在 encoding 包內,這是由於在註冊 registry 的時候會使用到這個方法。
介面
type Codec interface {
// Marshal returns the wire format of v.
Marshal(v interface{}) ([]byte, error)
// Unmarshal parses the wire format into v.
Unmarshal(data []byte, v interface{}) error
// String returns the name of the Codec implementation. This is unused by
// gRPC.
String() string
}
就是這個方法
// RegisterCodec registers the provided Codec for use with all gRPC clients and
// servers.
//
// The Codec will be stored and looked up by result of its Name() method, which
// should match the content-subtype of the encoding handled by the Codec. This
// is case-insensitive, and is stored and looked up as lowercase. If the
// result of calling Name() is an empty string, RegisterCodec will panic. See
// Content-Type on
// https://github.com/grpc/grpc/blob/master/doc/PROTOCOL-HTTP2.md#requests for
// more details.
//
// NOTE: this function must only be called during initialization time (i.e. in
// an init() function), and is not thread-safe. If multiple Compressors are
// registered with the same name, the one registered last will take effect.
func RegisterCodec(codec Codec) {
if codec == nil {
panic("cannot register a nil Codec")
}
if codec.Name() == "" {
panic("cannot register Codec with empty string result for Name()")
}
contentSubtype := strings.ToLower(codec.Name())
registeredCodecs[contentSubtype] = codec
}
Compressor
同時 encoding 包中還定義了 Compressor 介面,參照 Codec 理解即可。
// Compressor is used for compressing and decompressing when sending or
// receiving messages.
type Compressor interface {
// Compress writes the data written to wc to w after compressing it. If an
// error occurs while initializing the compressor, that error is returned
// instead.
Compress(w io.Writer) (io.WriteCloser, error)
// Decompress reads data from r, decompresses it, and provides the
// uncompressed data via the returned io.Reader. If an error occurs while
// initializing the decompressor, that error is returned instead.
Decompress(r io.Reader) (io.Reader, error)
// Name is the name of the compression codec and is used to set the content
// coding header. The result must be static; the result cannot change
// between calls.
Name() string
}
MetaData
這個包對應 context 中的 Value field 也就是 key-value 形式的儲存
在其他包中簡寫是 MD
結構
type MD map[string][]string
函式
實現了完善的儲存功能,從單一讀寫到批量(採用 pair 模式,...string 作為引數,len(string)%2==1 時會報錯,由於會有孤立的沒有配對的元資訊。
另外幾個函式是實現了從 context 中的讀取和寫入(這裡的寫入是 使用 context.WithValue 方法,即生成 parent context 的 copy。
注意⚠️
- 值得注意的是,在 MetaData 結構體中, value 的結構是 []string 。
- 同時 key 不可以以 "grpc-" 開頭,這是因為在 grpc 的 internal 包中已經保留了。
- 更為重要的是 在 context 中的讀取方式,其實是 MetaData 結構對應的是 context Value 中的 value 值,而 key 值設為 一個空結構體同時區分輸入輸入
type mdIncomingKey struct{}
type mdOutgoingKey struct{}