Kratos 初始化源碼分析

Kratos 是bilibili開源的一套Go微服務框架，包含大量微服務相關框架及工具。

名字來源於:《戰神》遊戲以希臘神話爲背景，講述由凡人成爲戰神的奎託斯（Kratos）成爲戰神並展開弒神屠殺的冒險歷程。

好！開始吧！

小提示：閱讀源碼時請保持清醒。

首先是按照Kratos tool 生產的工程目錄。

├── CHANGELOG.md

├── OWNERS

├── README.md

├── api # api目錄爲對外保留的proto文件及生成的pb.go文件

│ ├── api.bm.go

│ ├── api.pb.go # 經過go generate生成的pb.go文件

│ ├── api.proto

│ └── client.go

├── cmd

│ └── main.go # cmd目錄爲main所在

├── configs # configs爲配置文件目錄

│ ├── application.toml # 應用的自定義配置文件，多是一些業務開關如：useABtest = true

│ ├── db.toml # db相關配置

│ ├── grpc.toml # grpc相關配置

│ ├── http.toml # http相關配置

│ ├── memcache.toml # memcache相關配置

│ └── redis.toml # redis相關配置

├── go.mod

├── go.sum

└── internal # internal爲項目內部包，包括如下目錄：

│ ├── dao # dao層，用於數據庫、cache、MQ、依賴某業務grpc|http等資源訪問

│ │ ├── dao.bts.go

│ │ ├── dao.go

│ │ ├── db.go

│ │ ├── mc.cache.go

│ │ ├── mc.go

│ │ └── redis.go

│ ├── di # 依賴注入層 採用wire靜態分析依賴

│ │ ├── app.go

│ │ ├── wire.go # wire 聲明

│ │ └── wire_gen.go # go generate 生成的代碼

│ ├── model # model層，用於聲明業務結構體

│ │ └── model.go

│ ├── server # server層，用於初始化grpc和http server

│ │ ├── grpc # grpc層，用於初始化grpc server和定義method

│ │ │ └── server.go

│ │ └── http # http層，用於初始化http server和聲明handler

│ │ └── server.go

│ └── service # service層，用於業務邏輯處理，且爲方便http和grpc共用方法，建議入參和出參保持grpc風格，且使用pb文件生成代碼

│ └── service.go

└── test # 測試資源層 用於存放測試相關資源數據 如docker-compose配置 數據庫初始化語句等

└── docker-compose.yaml

Entry

入口在cmd/main.go下，咱們進去看看。

func  main() {

// 沒什麼好說的，參數解析

flag.Parse()

log.Init(nil) // debug flag: log.dir={path}

defer log.Close()

log.Info("kratos-demo start")

// -conf 參數的解析

paladin.Init()

// 這裏是 `應用的入口`

// 一會分析

_, closeFunc, err := di.InitApp()

if err != nil {

panic(err)

}

// os.Signal 是一個系統信號接收channel

c := make(chan os.Signal, 1)

// syscall 都是一些系統信號

signal.Notify(c, syscall.SIGHUP, syscall.SIGQUIT, syscall.SIGTERM, syscall.SIGINT)

for {

// 一旦有信號進來了，看下面的代碼邏輯，八成是關閉這個應用。

s := <-c

log.Info("get a signal %s", s.String())

switch s {

case syscall.SIGQUIT, syscall.SIGTERM, syscall.SIGINT:

closeFunc()

log.Info("kratos-demo exit")

time.Sleep(time.Second)

return

case syscall.SIGHUP:

default:

return

}

}

}

Initializer

接下來咱們去看di.InitApp()裏作了什麼。

這個方法是經過github.com/google/wire來生成.

若是你不知道wire能夠參考下面的官方原話：

Wire is a code generation tool that automates connecting components using dependency injection. Dependencies between components are represented in Wire as function parameters, encouraging explicit initialization instead of global variables. Because Wire operates without runtime state or reflection, code written to be used with Wire is useful even for hand-written initialization.

簡單來講就是Golang的依賴注入代碼生成器, 它不使用反射.由Google提供.

不過Wire不是咱們的重點, 咱們接着回到di.InitApp()去。

// Injectors from wire.go:

func  InitApp() (*App, func(), error) {

  

// 基本上就是建立一個個實例，和藹後它們的函數

// 若是途中建立出問題就全體下葬（觸發善後函數）.

  

// Redis實例，善後函數

redis, cleanup, err := dao.NewRedis()

if err != nil {

return  nil, nil, err

}

  

// memcache實例，善後函數

memcache, cleanup2, err := dao.NewMC()

if err != nil {

cleanup()

return  nil, nil, err

}

  

// 看起來只支持MySQL，善後函數

db, cleanup3, err := dao.NewDB()

if err != nil {

cleanup2()

cleanup()

return  nil, nil, err

}

  

// 把上面全部的模型對象作一個DAO層封裝

daoDao, cleanup4, err := dao.New(redis, memcache, db)

if err != nil {

cleanup3()

cleanup2()

cleanup()

return  nil, nil, err

}

  

// 這個是個重點，\`service\`是你的gRPC服務.

// 一會咱們去分析他

serviceService, cleanup5, err := service.New(daoDao)

if err != nil {

cleanup4()

cleanup3()

cleanup2()

cleanup()

return  nil, nil, err

}

  

// 有人會好奇Kratos是怎麼把gRPC和Gin融合在一塊兒的

//（沒錯Bilibili的web框架是Gin, 不過這個Gin的一部分核心代碼已經被魔改過了， 在Engine初始化的時候會多加入一個鏈路追蹤的Middleware, 還有一堆路由...）

// 祕密就在這裏，等會咱們再看

engine, err := http.New(serviceService)

if err != nil {

cleanup5()

cleanup4()

cleanup3()

cleanup2()

cleanup()

return  nil, nil, err

}

// gRPC的初始化的常規操做

server, err := grpc.New(serviceService)

if err != nil {

cleanup5()

cleanup4()

cleanup3()

cleanup2()

cleanup()

return  nil, nil, err

}

// 把上面的服務，engine,gRPC服務,整一塊

// 善後函數

// 後面稍微分析一下

app, cleanup6, err := NewApp(serviceService, engine, server)

if err != nil {

cleanup5()

cleanup4()

cleanup3()

cleanup2()

cleanup()

return  nil, nil, err

}

// 你能夠走了.

return app, func() {

cleanup6()

cleanup5()

cleanup4()

cleanup3()

cleanup2()

cleanup()

}, nil

}

  

//以上代碼全是自動生成，冗餘很正常

接下來咱們首先看看serviceService是個什麼東西.(這是什麼魔鬼命名)

進到Service.New(dao)

// Service service.

type  Service  struct {

// 配置文件映射的Map (這個命名就nm離譜)

ac *paladin.Map

// 字面意思

dao dao.Dao

}

  

// New new a service and return.

func  New(d dao.Dao) (s *Service, cf func(), err error) {

// 初始化～～～

s = &Service{

ac: &paladin.TOML{},

dao: d,

}

// 一個關閉的鉤子

cf = s.Close

  

// 熱加載 application.toml 配置文件

// 原理是使用fsnotify監聽文件變動

err = paladin.Watch("application.toml", s.ac)

return

}

  

// -------------- 下面都是你的gRPC業務邏輯-------------

  

// SayHello grpc demo func.

func (s *Service) SayHello(ctx context.Context, req *pb.HelloReq) (reply *empty.Empty, err error) {

reply = new(empty.Empty)

fmt.Printf("hello %s", req.Name)

return

}

  

// SayHelloURL bm demo func.

func (s *Service) SayHelloURL(ctx context.Context, req *pb.HelloReq) (reply *pb.HelloResp, err error) {

reply = &pb.HelloResp{

Content: "hello " + req.Name,

}

fmt.Printf("hello url %s", req.Name)

return

}

  

// Ping ping the resource.

func (s *Service) Ping(ctx context.Context, e *empty.Empty) (*empty.Empty, error) {

return &empty.Empty{}, s.dao.Ping(ctx)

}

  

// Close close the resource.

func (s *Service) Close() {}

哇哦，如今咱們知道了，Service是由一些gRPC方法，配置項，模型層組成的。

好，乘勝追擊咱們再看一看engine, err := http.New(serviceService)作了什麼。

var  svc pb.DemoServer

  

// New new a bm server.

func  New(s pb.DemoServer) (engine *bm.Engine, err error) {

var (

cfg bm.ServerConfig

ct paladin.TOML

)

// 讀取你的配置文件

if  err = paladin.Get("http.toml").Unmarshal(&ct); err != nil {

return

}

// 獲得http.toml的Server字段

if  err = ct.Get("Server").UnmarshalTOML(&cfg); err != nil {

return

}

svc = s

  

// 作一個新 engine

// (engine 是 Gin 裏的模塊，這裏我就不分析Gin的源碼了)

engine = bm.DefaultServer(&cfg)

  

// 將gRPC服務註冊到engine, 這個代碼註冊代碼是bm本身生成的

// 一會咱們分析

pb.RegisterDemoBMServer(engine, s)

  

// 把你的路由搞進去

initRouter(engine)

  

// 開始跑

err = engine.Start()

return

}

  

// 路由在這裏登記！

func  initRouter(e *bm.Engine) {

e.Ping(ping)

g := e.Group("/kratos-demo")

{

g.GET("/start", howToStart)

}

}

  

func  ping(ctx *bm.Context) {

if  _, err := svc.Ping(ctx, nil); err != nil {

log.Error("ping error(%v)", err)

ctx.AbortWithStatus(http.StatusServiceUnavailable)

}

}

  

// example for http request handler.

func  howToStart(c *bm.Context) {

k := &model.Kratos{

Hello: "Golang 大法好 !!!我好你個頭！",

}

c.JSON(k, nil)

}

若是你使用過gin這個web框架, 上面的代碼你必定很熟悉，對吧？

bm就是gin，只是部分代碼被Bilibili魔改了，總體架構是不變的。

OK，咱們看看RegisterDemoBMServer裏作了什麼.

// DemoBMServer is the server API for Demo service.

type  DemoBMServer  interface {

Ping(ctx context.Context, req *google_protobuf1.Empty) (resp *google_protobuf1.Empty, err error)

  

SayHello(ctx context.Context, req *HelloReq) (resp *google_protobuf1.Empty, err error)

  

SayHelloURL(ctx context.Context, req *HelloReq) (resp *HelloResp, err error)

}

  

// 咱們寫的gRPC服務

var  DemoSvc DemoBMServer

// ------------------------------------------------

  

// 咱們仔細分析這些方法不難發現

// 他們都會調用 `BindWith` 和對應的gRPC方法

  

// 先使用BindWith: 將request中的`Body` 轉化爲go中的 `struct`

// 而後使用gRPC方法處理請求數據

// 最後返回

  

// 本質就是經過http調用gRPC服務

  

func  demoPing(c *bm.Context) {

p := new(google_protobuf1.Empty)

if  err := c.BindWith(p, binding.Default(c.Request.Method, c.Request.Header.Get("Content-Type"))); err != nil {

return

}

resp, err := DemoSvc.Ping(c, p)

c.JSON(resp, err)

}

  

func  demoSayHello(c *bm.Context) {

p := new(HelloReq)

if  err := c.BindWith(p, binding.Default(c.Request.Method, c.Request.Header.Get("Content-Type"))); err != nil {

return

}

resp, err := DemoSvc.SayHello(c, p)

c.JSON(resp, err)

}

  

func  demoSayHelloURL(c *bm.Context) {

p := new(HelloReq)

if  err := c.BindWith(p, binding.Default(c.Request.Method, c.Request.Header.Get("Content-Type"))); err != nil {

return

}

resp, err := DemoSvc.SayHelloURL(c, p)

c.JSON(resp, err)

}

//-------------------------------------

  

// RegisterDemoBMServer Register the blademaster route

func  RegisterDemoBMServer(e *bm.Engine, server DemoBMServer) {

// server 是咱們以前編寫的gRPC服務

DemoSvc = server

// 將一些方法註冊到路由裏去

e.GET("/demo.service.v1.Demo/Ping", demoPing)

e.GET("/demo.service.v1.Demo/SayHello", demoSayHello)

e.GET("/kratos-demo/say_hello", demoSayHelloURL)

}

哇，原來只是把一些gRPC的服務綁定到gin的路由裏了呀。

借用gin來調用gRPC.

grpc.New()就不分析了。

而後是AppNew()

//go:generate kratos tool wire

type  App  struct {

svc *service.Service

http *bm.Engine

grpc *warden.Server

}

  

func  NewApp(svc *service.Service, h *bm.Engine, g *warden.Server) (app *App, closeFunc func(), err error) {

app = &App{

svc: svc,

http: h,

grpc: g,

}

  

// 一個關閉context的回調

closeFunc = func() {

ctx, cancel := context.WithTimeout(context.Background(), 35*time.Second)

if  err := g.Shutdown(ctx); err != nil {

log.Error("grpcSrv.Shutdown error(%v)", err)

}

if  err := h.Shutdown(ctx); err != nil {

log.Error("httpSrv.Shutdown error(%v)", err)

}

cancel()

}

return

}

到這裏初始化是結束了。

Summary

kratos的初始化流程:

• 讀取配置文件
• 實例化Dao層
• 實例化gRPC服務
• 實例化gin的engine
• 註冊gPRC到engine
• 啓動engine
• 啓動gRPC服務端
• 得到整個程序關閉的回調

我分得應該仍是比較細的。

後面應該還會分析warden，它是Kratos在grpc原版上的一個封裝版本。實際上官方對warden已經有很詳細的分析流程了

溜了溜了...