golang thrift 源碼分析，服務器和客戶端到底是如何工做的

首先編寫thrift文件(rpcserver.thrift)，運行thrift --gen go rpcserver.thrift，生成代碼

namespace go rpc

service RpcService {
    string SayHi(1: string name);
    void SayHello(1: string name);
}

搭建一個以二進制爲傳輸協議的服務器以下：

type rpcService struct{
}

func (this *rpcService)SayHi(name string)(r string, err error){
    fmt.Println("Hi ", name)
    r = "Hello "+name
    err = nil
    return
}

func (this *rpcService)SayHello(name string)err error{
    fmt.Println("Hello ", name)
    err = nil
    return
}

func StartServer(){

    serverTransport, err := thrift.NewTServerSocket("127.0.0.1:8808")
    if err != nil {
        fmt.Println("Error!", err)
        return
    }
    handler := &rpcService{}
    processor := NewRpcServiceProcessor(handler)
    server := thrift.NewTSimpleServer2(processor, serverTransport)
    fmt.Println("thrift server in localhost")

    server.Serve()
}

查看自動生成的代碼recserver.go，咱們發現 NewRpcServiceProcessor函數代碼以下：

func NewRpcServiceProcessor(handler RpcService) *RpcServiceProcessor {

    self4 := &RpcServiceProcessor{handler: handler, processorMap: make(map[string]thrift.TProcessorFunction)}
    self4.processorMap["SayHi"] = &rpcServiceProcessorSayHi{handler: handler}
    self4.processorMap["SayHello"] = &rpcServiceProcessorSayHello{handler: handler}
    return self4
}

也就是說，thrift經過key-value保存了咱們實際將要運行的函數，最終經過handler來執行。

這裏就有點像咱們使用golang系統中的http包中的ListenAndServer()函數時，提早經過Handfunc來設置好函數路由是一個意思。

再看看Serve()函數是如何實現的：

func (p *TSimpleServer) Serve() error {
    err := p.Listen()
    if err != nil {
        return err
    }
    p.AcceptLoop()
    return nil
}

func (p *TSimpleServer) AcceptLoop() error {
    for {
        client, err := p.serverTransport.Accept()
        if err != nil{
        //.......被我刪掉
        }
        if client != nil {
            go func() {
                if err := p.processRequests(client); err != nil {
                    log.Println("error processing request:", err)
                }
            }()
        }
    }
}

Serve()函數負責監聽鏈接到服務器上的client，而且經過processRequests()函數來處理client。實際處理過程當中，服務器會獲取client的processor，而後進一步處理client的請求。這部分先暫停一下，咱們來分析一下client端的工做原理，以後再回過頭來看看會比較清晰一些

 

首先咱們架設client端以下，而且經過client端來發送一個SayHi的操做：   

    transport, err := thrift.NewTSocket(net.JoinHostPort("127.0.0.1", "8808"))
    if err != nil {
        //...
    }
    
    protocolFactory := thrift.NewTBinaryProtocolFactoryDefault()

    client := NewRpcServiceClientFactory(transport, protocolFactory)
    if err := transport.Open(); err != nil {
        //...
    }
    defer transport.Close()
    res, _ := client.SayHi("wordl")

如今問題來了，這個SayHi是如何通知給服務器的呢？不急，看源碼

在咱們調用thrift --gen go XXX命令的時候，thrift已經給咱們生成了SayHi過程的代碼，以下：

func (p *RpcServiceClient) SayHi(name string) (r string, err error) {
    if err = p.sendSayHi(name); err != nil {
        return
    }
    return p.recvSayHi()
}

其中RpcServiceClient類型就是咱們的client，能夠看到先調用了一個sendSayHi，若是沒有錯誤的話，又調用了一個recvSayHi。

其實sendSayHi就是咱們通知服務器執行SayHi()函數的關鍵，而recvSayHi是接受服務器的執行結果的。

一塊兒看下sendSayHi是如何實現的(代碼被我精簡，這保留了關鍵部分，完整代碼能夠本身經過thrift命令生成查看)

func (p *RpcServiceClient) sendSayHi(name string) (err error) {
    oprot := p.OutputProtocol  //獲取傳輸協議
    
    if err = oprot.WriteMessageBegin("SayHi", thrift.CALL, p.SeqId); err != nil { //發送SayHi字符創，告訴服務器未來執行的函數
        return
    }
    args := RpcServiceSayHiArgs{ //構建參數
        Name: name,
    }
    if err = args.Write(oprot); err != nil {  //將參數發送給服務器
        return
    }
    if err = oprot.WriteMessageEnd(); err != nil { //通知服務器發送完畢
        return
    }
    return oprot.Flush()
}

經過這樣的一系列數據傳輸，服務器經過路由解析，即可以正確的知道該執行哪一個函數了。thrift的精髓也正在此，實現了rpc架構，客戶端只須要簡單的調用client.SayHi()，沒必要知道這是本地調用仍是遠程調用。

 

好了，既然請求發出了，咱們如今固然看看服務器是如何響應的，在源碼中，有一個函數是專門響應客戶端請求的：

func (p *RpcServiceProcessor) Process(iprot, oprot thrift.TProtocol) (success bool, err thrift.TException)

前面講解服務器端是如何建立的時候講到過一個processRequests()函數，它在client鏈接上server的時候會被server調用。咱們看看源碼：

func (p *TSimpleServer) processRequests(client TTransport) error {
    processor := p.processorFactory.GetProcessor(client) //....
    for {
        ok, err := processor.Process(inputProtocol, outputProtocol)
        if err{
                  //....
        }
    }
    return nil
}

在去除無關代碼以後咱們看到，服務器首先獲取客戶端的processor，而後調用processor的Process函數，從而執行響應客戶端的請求。

看看Process函數具體是如何實現的：

func (p *RpcServiceProcessor) Process(iprot, oprot thrift.TProtocol) (success bool, err thrift.TException) {
    name, _, seqId, err := iprot.ReadMessageBegin() //獲取客戶端請求執行的函數名稱
    if err != nil {
        return false, err
    }
    if processor, ok := p.GetProcessorFunction(name); ok {
        return processor.Process(seqId, iprot, oprot)  //執行
    }
    //...
}

要注意的是，函數中用紅色標註的Process是另一個函數，這裏可不是遞歸。兩個Process函數的聲明是不同的：

func (p *RpcServiceProcessor) Process(iprot, oprot thrift.TProtocol) (success bool, err thrift.TException)  //RpcServiceProcessor是server的processor

func (p *rpcServiceProcessorSayHi) Process(seqId int32, iprot, oprot thrift.TProtocol) (success bool, err thrift.TException) //rpcServiceProcessorSayHi是具體的handler，

如今到了最關鍵的時候了，咱們看看handler是如何執行process的：

func (p *rpcServiceProcessorSayHi) Process(seqId int32, iprot, oprot thrift.TProtocol) (success bool, err thrift.TException) {
    args := RpcServiceSayHiArgs{} //構建參數
    if err = args.Read(iprot); err != nil { //讀取客戶端發來的參數
        //處理err
    }

    iprot.ReadMessageEnd()  //讀取客戶端的結束消息
    result := RpcServiceSayHiResult{} 
    var retval string
    var err2 error
    if retval, err2 = p.handler.SayHi(args.Name); err2 != nil { //執行函數
        //..處理err
    } else {
        result.Success = &retval
    }
    //...將result發送給客戶端，流程和client發送請求相似，client經過recvSayHi()函數接受result
    return true, err
}

如今，服務器和客戶端到底是如何工做的。你明白了嗎

 

轉載請註明出處，謝謝