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  結構體

// 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 設定

1. 在 Client 使用方法的時候直接作為 引數傳遞，將剛才所說的函式-返回一個實現了 CallOption 介面的結構體。
2. 在 生成 Client 的時候就傳遞設定。具體如下

1. 通過 dialOptions.go 中的 函式 grpc.WithDefaultCallOptions() 
2. 這個函式會將 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...)
	})
}

有沒有感覺有點不好理解？給你們一個例項

1. 使用的第一種方法

response, err := myclient.MyCall(ctx, request, grpc.CallContentSubtype("mycodec"))

1. 使用第二種方法

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{}