thrift客户端

设计

TClient ,
TStandardClient、WrappedTClient 实现了TClient接口的Call方法

demo

https://github.com/weikeqin/thrift-tutorial-go-demo

以client调用add方法为例

调用下游Add()方法代码

func main() {
	// 创建thrift client
	thriftClient := getThriftClient()
	// 创建 calculatorClient   tutorial是由idl生成的
	calculatorClient := tutorial.NewCalculatorClient(thriftClient)
	// 调用Add方法
	sum, _ := calculatorClient.Add(defaultCtx, 1, 2)
	fmt.Print("1+2=", sum, "\n")
}
// 获取一个thriftClient
func getThriftClient() *thrift.TStandardClient {

	addr := flag.String("addr", "localhost:9090", "Address to listen to")

	var transportFactory thrift.TTransportFactory
	transportFactory = thrift.NewTTransportFactory()

	cfg := &thrift.TConfiguration{
		TLSConfig: &tls.Config{
			InsecureSkipVerify: true,
		},
	}

	// 传输方式
	var transport thrift.TTransport
	transport = thrift.NewTSocketConf(*addr, cfg)
	// 传输方式增强
	transport, err := transportFactory.GetTransport(transport)
	if err != nil {
		fmt.Println(err)
	}

	//defer transport.Close()
	if err := transport.Open(); err != nil {
		fmt.Println(err)
	}

	// 协议工厂
	var protocolFactory thrift.TProtocolFactory
	protocolFactory = thrift.NewTBinaryProtocolFactoryConf(nil)

	iprot := protocolFactory.GetProtocol(transport)
	oprot := protocolFactory.GetProtocol(transport)

	// 创建thrift client,
	// 这个client使用TStandardClient类型,  TStandardClient 实现了 TClient 接口
	thriftClient := thrift.NewTStandardClient(iprot, oprot)
	return thriftClient
}

CalculatorClient::Add

CalculatorClient 可以看做Calculator接口的代理

// gen-go/tutorial/tutorial.go 
package tutorial


// Parameters:
//  - Num1
//  - Num2
func (p *CalculatorClient) Add(ctx context.Context, num1 int32, num2 int32) (_r int32, _err error) {
  // 业务方法入参包装  `CalculatorAddArgs`实现了`TStruct`接口
  var _args3 CalculatorAddArgs
  _args3.Num1 = num1
  _args3.Num2 = num2
  // 业务返回结果   
  // call的第4个入参类型是`TStruct`,`CalculatorAddResult`实现了`TStruct`接口,可以直接传递
  var _result5 CalculatorAddResult
  // 通用返回结果里的headers
  var _meta4 thrift.ResponseMeta

  // 调用 TStandardClient::Call 方法
  _meta4, _err = p.Client_().Call(ctx, "add", &_args3, &_result5)
  p.SetLastResponseMeta_(_meta4)
  if _err != nil {
    return
  }
  // 返回业务处理结果
  return _result5.GetSuccess(), nil
}

源码解读-go

TClient

TClient

TClient定义了客户端

package thrift
// thrift.client.go 

type TClient interface {
	Call(ctx context.Context, method string, args, result TStruct) (ResponseMeta, error)
}

TStandardClient

TStandardClient有3个方法 Send Recv Call

package thrift
// thrift.client.go 

// 
type TStandardClient struct {
	seqId        int32      // 序列Id
	iprot, oprot TProtocol  // 输入传输协议 输出传输协议
}

创建NewTStandardClient方法 NewTStandardClient

// TStandardClient 实现了 TClient 接口,使用 Thrift 的标准消息格式。
// 并发使用是不安全的
// 
// TStandardClient implements TClient, and uses the standard message format for Thrift.
// It is not safe for concurrent use.
func NewTStandardClient(inputProtocol, outputProtocol TProtocol) *TStandardClient {
	return &TStandardClient{
		iprot: inputProtocol,
		oprot: outputProtocol,
	}
}

Call

// param  method  方法名
// param  args    业务入参
// param  result  返回(业务)结果
func (p *TStandardClient) Call(ctx context.Context, method string, args, result TStruct) (ResponseMeta, error) {
    // 序列Id+1  注意:这儿不是并发安全的 
    p.seqId++
	seqId := p.seqId

    // 调用Send方法
	if err := p.Send(ctx, p.oprot, seqId, method, args); err != nil {
		return ResponseMeta{}, err
	}

	// method is oneway
	if result == nil {
		return ResponseMeta{}, nil
	}

    // 调用Recv方法,把返回结果写入result
    err := p.Recv(ctx, p.iprot, seqId, method, result)
    // 处理返回headers 
	var headers THeaderMap
	if hp, ok := p.iprot.(*THeaderProtocol); ok {
		headers = hp.transport.readHeaders
	}
	return ResponseMeta{
		Headers: headers,
	}, err
}

从以上的代码可以看到
1、TStandardClient并发使用是不安全的,所以使用的时候得每次新建一个
2、TStandardClient是同步调用

Send

// client给Server发请求的方法 
func (p *TStandardClient) Send(ctx context.Context, oprot TProtocol, seqId int32, method string, args TStruct) error {

	// 设置headers
	// Set headers from context object on THeaderProtocol
	if headerProt, ok := oprot.(*THeaderProtocol); ok {
		headerProt.ClearWriteHeaders()
		for _, key := range GetWriteHeaderList(ctx) {
			if value, ok := GetHeader(ctx, key); ok {
				headerProt.SetWriteHeader(key, value)
			}
		}
	}


	// 使用配置的传输协议写入数据 
	// method是方法名  CALL是TMessageType对应的枚举  seqId是序列Id 
    // 调用server接口 其实就是给server发一个请求,按照约定的传输协议(TProtocol)和传输方式(TTransport)传输对应的二进制数据 
    // 这个例子里传输协议是 TBinaryProtocol  传输方式是 TSocket
	if err := oprot.WriteMessageBegin(ctx, method, CALL, seqId); err != nil {
		return err
	}
	if err := args.Write(ctx, oprot); err != nil {
		return err
	}
	if err := oprot.WriteMessageEnd(ctx); err != nil {
		return err
    }
    
    // 传输协议 调 传输方式 flush 
	return oprot.Flush(ctx)
}

可以看到,client 的send方法主要作用有2个
1、(在传输协议是THeaderProtocol的情况下)设置headers
2、按照约定的传输一些写数据,并flush

Recv

// 接收请求方法
func (p *TStandardClient) Recv(ctx context.Context, iprot TProtocol, seqId int32, method string, result TStruct) error {
    // 
	rMethod, rTypeId, rSeqId, err := iprot.ReadMessageBegin(ctx)
	if err != nil {
		return err
	}

    // 消息校验 校验方法名、序列Id、消息类型
	if method != rMethod {  // 方法和消息里的方法不一致,抛异常
		return NewTApplicationException(WRONG_METHOD_NAME, fmt.Sprintf("%s: wrong method name", method))
	} else if seqId != rSeqId { // 序列Id不一致,抛异常 
		return NewTApplicationException(BAD_SEQUENCE_ID, fmt.Sprintf("%s: out of order sequence response", method))
	} else if rTypeId == EXCEPTION { // 异常类消息,抛异常
		var exception tApplicationException
		if err := exception.Read(ctx, iprot); err != nil {
			return err
		}

		if err := iprot.ReadMessageEnd(ctx); err != nil {
			return err
		}

		return &exception
	} else if rTypeId != REPLY { // 返回结果里消息类型不是回复消息,抛异常
		return NewTApplicationException(INVALID_MESSAGE_TYPE_EXCEPTION, fmt.Sprintf("%s: invalid message type", method))
	}

    // 读取消息
	if err := result.Read(ctx, iprot); err != nil {
		return err
	}

    // 
	return iprot.ReadMessageEnd(ctx)
}

WrappedTClient

Wrapper提供了一种包装机制,使得在执行某方法前先执行Wrapper
因此可以在客户端和服务器做很多功能:接口监控、熔断限流、鉴权、Filter等。

// file thrift/middleware.go
package thrift

// WrappedTClient is a convenience struct that implements the TClient interface
// using inner Wrapped function.
//
// This is provided to aid in developing ClientMiddleware.
type WrappedTClient struct {
	Wrapped func(ctx context.Context, method string, args, result TStruct) (ResponseMeta, error)
}

新建WrappedTClient

// file thrift/middleware.go
package thrift

// 验证 WrappedTClient 实现了 TClient
// verify that WrappedTClient implements TClient
var (
	_ TClient = WrappedTClient{}
	_ TClient = (*WrappedTClient)(nil)
)


// WrappedTClient 是一个使用内部 Wrapped 函数实现 TClient 接口的便利结构。
// 这是为了帮助开发 ClientMiddleware。
// 
// WrapClient wraps the given TClient in the given middlewares.
//
// Middlewares will be called in the order that they are defined:
//
//		1. Middlewares[0]
//		2. Middlewares[1]
//		...
//		N. Middlewares[n]
func WrapClient(client TClient, middlewares ...ClientMiddleware) TClient {
	// 反向添加中间件,因此列表中的第一个是最外面的。
	// Add middlewares in reverse so the first in the list is the outermost.
	for i := len(middlewares) - 1; i >= 0; i-- {
		// middlewares[i]是ClientMiddleware,是一个函数,client是对应的入参 
		client = middlewares[i](client)
	}
	return client
}

Call

// file thrift/middleware.go
package thrift

// Call 通过调用 c.Wrapped的TClient接口实现 来方法调用 并返回结果 。
// Call implements the TClient interface by calling and returning c.Wrapped.
func (c WrappedTClient) Call(ctx context.Context, method string, args, result TStruct) (ResponseMeta, error) {
	// 
	return c.Wrapped(ctx, method, args, result)
}

WrappedTClient Demo

func main() {
	var thriftClient thrift.TClient
	thriftClient = getThriftClient()

	// 扩展点:日志中间件  
	clientMiddleware := clientLoggingMiddleware()
	// 包装后的client
	wrapClient := thrift.WrapClient(thriftClient, clientMiddleware)

	// 创建 calculatorClient   tutorial是由idl生成的
	calculatorClient := tutorial.NewCalculatorClient(wrapClient)
	// 调用Add方法
	sum, _ := calculatorClient.Add(defaultCtx, 1, 2)
	fmt.Print("1+2=", sum, "\n")
}
// 生成一个日志中间件 Demo用  
func clientLoggingMiddleware() thrift.ClientMiddleware {

	clientMiddleware := func(next thrift.TClient) thrift.TClient {
		wrappedTClient := thrift.WrappedTClient{
			Wrapped: func(ctx context.Context, method string, args, result thrift.TStruct) (thrift.ResponseMeta, error) {
				// 调用前打印 方法名及参数
				log.Printf("Before: %q    Args: %#v ", method, args)
				// 方法调用
				headers, err := next.Call(ctx, method, args, result)
				// 调用后打印
				log.Printf("After: %q    Result: %#v", method, result)
				if err != nil {
					log.Printf("Error: %v", err)
				}
				return headers, err
			},
		}
		return wrappedTClient
	}

	return clientMiddleware
}

参考资料

[1] github-thrift-0.16.0-go
[2] Golang微服务:Micro限流、熔断