Kitex 消息類型

消息類型

支持 PingPong、Oneway、Streaming。

協(xié)議支持 

目前 Kitex 支持的消息類型、編解碼協(xié)議和傳輸協(xié)議

• PingPong：客戶端發(fā)起一個(gè)請(qǐng)求后會(huì)等待一個(gè)響應(yīng)才可以進(jìn)行下一次請(qǐng)求
• Oneway：客戶端發(fā)起一個(gè)請(qǐng)求后不等待一個(gè)響應(yīng)
• Streaming：客戶端發(fā)起一個(gè)或多個(gè)請(qǐng)求 , 等待一個(gè)或多個(gè)響應(yīng)

Thrift

目前 Thrift 支持 PingPong 和 Oneway。Kitex 計(jì)劃支持 Thrift Streaming。

Example

IDL 定義 :

namespace go echo

struct Request {
    1: string Msg
}

struct Response {
    1: string Msg
}

service EchoService {
    Response Echo(1: Request req); // pingpong method
    oneway void VisitOneway(1: Request req); // oneway method
}

生成的代碼組織結(jié)構(gòu) :

.
└── kitex_gen
    └── echo
        ├── echo.go
        ├── echoservice
        │   ├── client.go
        │   ├── echoservice.go
        │   ├── invoker.go
        │   └── server.go
        └── k-echo.go

Server 的處理代碼形如 :

package main

import (
    "xx/echo"
    "xx/echo/echoservice"
)

type handler struct {}

func (handler) Echo(ctx context.Context, req *echo.Request) (r *echo.Response, err error) {
    //...
    return &echo.Response{ Msg: "world" }
}

func (handler) VisitOneway(ctx context.Context, req *echo.Request) (err error) {
    //...
    return nil
}

func main() {
    svr, err := echoservice.NewServer(handler{})
    if err != nil {
        panic(err)
    }
    svr.Run()
}

PingPong

Client 側(cè)代碼 :

package main

import (
    "xx/echo"
    "xx/echo/echoservice"
)

func main() {
    cli, err := echoservice.NewClient("destServiceName")
    if err != nil {
        panic(err)
    }
    req := echo.NewRequest()
    req.Msg = "hello"
    resp, err := cli.Echo(req)
    if err != nil {
        panic(err)
    }
    // resp.Msg == "world"
}

Oneway

Client 側(cè)代碼 :

package main

import (
    "xx/echo"
    "xx/echo/echoservice"
)

func main() {
    cli, err := echoservice.NewClient("destServiceName")
    if err != nil {
        panic(err)
    }
    req := echo.NewRequest()
    req.Msg = "hello"
    err = cli.VisitOneway(req)
    if err != nil {
        panic(err)
    }
    // no response return
}

Protobuf

Kitex 支持兩種承載 Protobuf 負(fù)載的協(xié)議：

• Kitex Protobuf
• 只支持 PingPong，若 IDL 定義了 stream 方法，將默認(rèn)使用 gRPC 協(xié)議
• gRPC 協(xié)議
• 可以與 gRPC 互通，與 gRPC service 定義相同，支持 Unary(PingPong)、 Streaming 調(diào)用

Example 

以下給出 Streaming 的使用示例。

IDL 定義 :

syntax = "proto3";

option go_package = "echo";

package echo;

message Request {
  string msg = 1;
}

message Response {
  string msg = 1;
}

service EchoService {
  rpc ClientSideStreaming(stream Request) returns (Response) {} // 客戶端側(cè) streaming
  rpc ServerSideStreaming(Request) returns (stream Response) {} // 服務(wù)端側(cè) streaming
  rpc BidiSideStreaming(stream Request) returns (stream Response) {} // 雙向流
}

生成的代碼組織結(jié)構(gòu) :

.
└── kitex_gen
    └── echo
        ├── echo.pb.go
        └── echoservice
            ├── client.go
            ├── echoservice.go
            ├── invoker.go
            └── server.go

Server 側(cè)代碼 :

package main

import (
    "sync"

    "xx/echo"
    "xx/echo/echoservice"
}

type handler struct{}

func (handler) ClientSideStreaming(stream echo.EchoService_ClientSideStreamingServer) (err error) {
    for {
        req, err := stream.Recv()
        if err != nil {
            return err
        }
    }
}

func (handler) ServerSideStreaming(req *echo.Request, stream echo.EchoService_ServerSideStreamingServer) (err error) {
      _ = req
      for {
          resp := &echo.Response{Msg: "world"}
          if err := stream.Send(resp); err != nil {
              return err
          }
      }
}

func (handler) BidiSideStreaming(stream echo.EchoService_BidiSideStreamingServer) (err error) {
    var once sync.Once
    go func() {
        for {
            req, err2 := stream.Recv()
            log.Println("received:", req.GetMsg())
            if err2 != nil {
                once.Do(func() {
                    err = err2
                })
                break
            }
        }
    }()
    for {
        resp := &echo.Response{Msg: "world"}
        if err2 := stream.Send(resp); err2 != nil {
            once.Do(func() {
                err = err2
            })
            return
        }
    }
    return
}

func main() {
    svr, err := echoservice.NewServer(handler{})
    if err != nil {
        panic(err)
    }
    svr.Run()
}

Streaming

客戶端流

package main

import (
    "xx/echo"
    "xx/echo/echoservice"
}

func main() {
    cli, err := echoservice.NewClient("destServiceName")
    if err != nil {
        panic(err)
    }
    cliStream, err := cli.ClientSideStreaming(context.Background())
    if err != nil {
        panic(err)
    }
    for {
        req := &echo.Request{Msg: "hello"}
        if err := cliStream.Send(req); err != nil {
            panic(err)
        }
    }

}

服務(wù)器端流

package main

import (
    "xx/echo"
    "xx/echo/echoservice"
}

func main() {
    cli, err := echoservice.NewClient("destServiceName")
    if err != nil {
        panic(err)
    }
    req := &echo.Request{Msg: "hello"}
    svrStream, err := cli.ServerSideStreaming(context.Background(), req)
    if err != nil {
        panic(err)
    }
    for {
        resp, err := svrStream.Recv()
        if err != nil {
            panic(err)
        }
        // resp.Msg == "world"
    }

}

雙向流

package main

import (
    "xx/echo"
    "xx/echo/echoservice"
}

func main() {
    cli, err := echoservice.NewClient("destServiceName")
    if err != nil {
        panic(err)
    }
    req := &echo.Request{Msg: "hello"}
    bidiStream, err := cli.BidiSideStreaming(context.Background())
    if err != nil {
        panic(err)
    }
    go func() {
        for {
            req := &echo.Request{Msg: "hello"}
            err := bidiStream.Send(req)
            if err != nil {
                panic(err)
            }
        }
    }()
    for {
        resp, err := bidiStream.Recv()
        if err != nil {
            panic(err)
        }
        // resp.Msg == "world"
    }
}