go微服务之四- Kratos Transporter 传输层.md · tking/micro-service

本文主要介绍Kratos Transport和源码的分析。

kratos 框架对传输层进行了抽象，用户可以通过实现接口来接入实现，框架默认实现了gRPC和HTTP两种通信协议传输层。

抽象接口

Server Interface

// 服务的启动和停止，用于管理服务生命周期。
type Server interface {
    Start(context.Context) error
    Stop(context.Context) error
}

Transporter

type Transporter interface {
    // 代表实现的通讯协议的种类，如内置的 http grpc，也可以实现其他的类型如 mqtt，websocket
    Kind() Kind
    // 提供的服务终端地址
    Endpoint() string
    // 用于标识服务的完整方法路径
    // 示例: /helloworld.Greeter/SayHello
    Operation() string
    // http 的请求头或者 grpc 的元数据
    Header() Header
}

Endpointer

type Endpointer interface {
    // 用于实现注册到注册中心的终端地址，如果不实现这个方法则不会注册到注册中心
    Endpoint() (*url.URL, error)
}

Example For Http

使用kratos-transport 创建一个http server 和 client ，在 client和server之间通讯

server

server的完整代码 server.go 如下：

package main

import (
	"context"
	"encoding/json"
	"fmt"
	"log"
	"net/http"
	"os"
	"os/signal"
	"syscall"
	"time"

	transporthttp "github.com/go-kratos/kratos/v2/transport/http"
)

func NewServer() *transporthttp.Server {
	var opts = []transporthttp.ServerOption{
		transporthttp.Address("0.0.0.0:8000"),
		transporthttp.Timeout(time.Second * 10),
	}
	svr := transporthttp.NewServer(opts...)
	svr.Handle("/", &helloHandler{})
	return svr
}

func main() {
	ctx := context.Background()
	srv := NewServer()

	err := srv.Start(ctx)
	if err != nil {
		if err == http.ErrServerClosed {
			log.Print("Server closed under request")
		} else {
			log.Fatal("Server closed unexpected")
		}
	}

	done := make(chan os.Signal)
	signal.Notify(done, os.Interrupt, syscall.SIGINT, syscall.SIGTERM)
	go func() {
		<-done
		if err := srv.Stop(ctx); err != nil {
			log.Fatal("Shutdown NewServer:", err)
		}
	}()
}

type Foo struct {
	Bar string
}

type helloHandler struct{}

func (*helloHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
	err := r.ParseForm()
	if err != nil {
		return
	}
	formData := make(map[string]interface{})
	json.NewDecoder(r.Body).Decode(&formData)
	fmt.Println(formData)
	for k, v := range formData {
		fmt.Println("key", k, "value", v)
	}

	reply := &Foo{
		Bar: "NiceToMeetYouToo",
	}
	replyBytes, err := json.Marshal(reply)
	if err != nil {
		return
	}
	fmt.Fprintf(w, string(replyBytes))
}

启动服务

 go run main.go

输出

API server listening at: [::]:60870
INFO msg=[HTTP] server listening on: [::]:8000

client

client 的完整代码 client.go 如下：


package client

import (
   "context"
   "github.com/go-kratos/kratos/v2/transport/http"
   "testing"
   "time"

   "github.com/stretchr/testify/assert"
)

type Foo struct {
   Bar string
}

func TestNewClient(t *testing.T) {
   client, err := http.NewClient(
   	context.Background(),
   	http.WithEndpoint("0.0.0.0:8000"),
   	http.WithMiddleware(),
   	http.WithTimeout(time.Second*10),
   )
   assert.Nil(t, err)

   values := map[string]string{
   	"foo":   "bar",
   	"hello": "world",
   } 
   assert.Nil(t, err)
   reply := &Foo{}
   err = client.Invoke(context.Background(), "POST", "/", values, reply)
   assert.Nil(t, err)

   t.Log("reply", reply)
}

运行测试用例，

server 输出：

key foo value bar
key hello value world

client 输出：

=== RUN   TestNewClient
    client_test.go:42: reply &{NiceToMeetYouToo}

源码分析

Transport Server

// 服务的启动和停止，用于管理服务生命周期。
type Server interface {
   Start(context.Context) error
   Stop(context.Context) error
}

Start：启动服务、监听端口、处理请求 Stop：停止服务

如上例 server.go 中，

svr := transporthttp.NewServer(opts...)

这个用创建一个 Server Interface的Http实例

// Server is an HTTP server wrapper.
type Server struct {
	*http.Server
	lis         net.Listener
	tlsConf     *tls.Config
	endpoint    *url.URL
	err         error
	network     string
	address     string
	timeout     time.Duration
	filters     []FilterFunc
	ms          []middleware.Middleware
	dec         DecodeRequestFunc
	enc         EncodeResponseFunc
	ene         EncodeErrorFunc
	strictSlash bool
	router      *mux.Router
	log         *log.Helper
}

可以看到，Server是对标准库 net/http 中http.Server 的一个封装。

// ServerOption is an HTTP server option.
type ServerOption func(*Server)

可通过 option 设置服务参数，比如例子中的

var opts = []transporthttp.ServerOption{
		transporthttp.Address("0.0.0.0:8000"),
		transporthttp.Timeout(time.Second * 10),
	}

Start 启动一个net/http.Server

// Start start the HTTP server.
func (s *Server) Start(ctx context.Context) error { 
    ...
	s.log.Infof("[HTTP] server listening on: %s", s.lis.Addr().String())
	var err error
	if s.tlsConf != nil {
		err = s.ServeTLS(s.lis, "", "")
	} else {
		err = s.Serve(s.lis)
	}
    ...
}

启动流程

func (a *App) Run() error {
	...
	for _, srv := range a.opts.servers {
		srv := srv
		eg.Go(func() error {
			<-ctx.Done() // wait for stop signal
			return srv.Stop(ctx)
		})
		wg.Add(1)
		eg.Go(func() error {
			wg.Done()
			return srv.Start(ctx)
		})
	}
	wg.Wait()
    ...
}

kratos 的App 层无需关心底层服务协议的实现，只需管理好应用配置、服务生命周期、加载顺序即可。

Transport Client

Transport Client 是会 http CLient封装

// Client is an HTTP client.
type Client struct {
	opts     clientOptions
	target   *Target
	r        *resolver
	cc       *http.Client
	insecure bool
}

如client.go 中的一个Http Post

    values := map[string]string{
	}  
	reply := &Foo{}
	err = client.Invoke(context.Background(), "POST", "/", values, reply)

Invoke封装了post，使用更加简洁。

总结

kratos 的App 层无需关心底层服务协议的实现，只需管理好应用配置、服务生命周期、加载顺序即可。抽象的服务协议可以专注应用配置、服务生命周期等。层次清晰，职责清楚。

参考

https:/github.com/go-kratos/kratos/v2/transport
https://go-kratos.dev/docs/component/transport/overview

tking / micro-service

抽象接口

Server Interface

Transporter

Endpointer

Example For Http

server

client

源码分析

Transport Server

启动流程

Transport Client

总结

参考

简介

发行版

贡献者

近期动态

tking / micro-service .gitee-modal { width: 500px !important; }

抽象接口

Server Interface

Transporter

Endpointer

Example For Http

server

client

源码分析

Transport Server

启动流程

Transport Client

总结

参考

简介

发行版

贡献者

近期动态

搜索帮助

tking / micro-service