Kratos 是bilibili開源的一套Go微服務框架，包含大量微服務相關框架及工具。
名字來源於:《戰神》遊戲以希臘神話為背景，講述由凡人成為戰神的奎託斯（Kratos）成為戰神並展開弒神屠殺的冒險歷程。
好！開始吧！
小提示：閱讀源碼時請保持清醒。
首先是按照Kratos tool 生產的工程目錄。
1├── CHANGELOG.md
2├── OWNERS
3├── README.md
4├── api # api目錄為對外保留的proto文件及生成的pb.go文件
5│ ├── api.bm.go
6│ ├── api.pb.go # 通過go generate生成的pb.go文件
7│ ├── api.proto
8│ └── client.go
9├── cmd
10│ └── main.go # cmd目錄為main所在
11├── configs # configs為配置文件目錄
12│ ├── application.toml # 應用的自定義配置文件，可能是一些業務開關如：useABtest = true
13│ ├── db.toml # db相關配置
14│ ├── grpc.toml # grpc相關配置
15│ ├── http.toml # http相關配置
16│ ├── memcache.toml # memcache相關配置
17│ └── redis.toml # redis相關配置
18├── go.mod
19├── go.sum
20└── internal # internal為項目內部包，包括以下目錄：
21│ ├── dao # dao層，用於數據庫、cache、MQ、依賴某業務grpc|http等資源訪問
22│ │ ├── dao.bts.go
23│ │ ├── dao.go
24│ │ ├── db.go
25│ │ ├── mc.cache.go
26│ │ ├── mc.go
27│ │ └── redis.go
28│ ├── di # 依賴注入層 採用wire靜態分析依賴
29│ │ ├── app.go
30│ │ ├── wire.go # wire 聲明
31│ │ └── wire_gen.go # go generate 生成的代碼
32│ ├── model # model層，用於聲明業務結構體
33│ │ └── model.go
34│ ├── server # server層，用於初始化grpc和http server
35│ │ ├── grpc # grpc層，用於初始化grpc server和定義method
36│ │ │ └── server.go
37│ │ └── http # http層，用於初始化http server和聲明handler
38│ │ └── server.go
39│ └── service # service層，用於業務邏輯處理，且為方便http和grpc共用方法，建議入參和出參保持grpc風格，且使用pb文件生成代碼
40│ └── service.go
41└── test # 測試資源層 用於存放測試相關資源數據 如docker-compose配置 數據庫初始化語句等
42 └── docker-compose.yaml
Entry
入口在cmd/main.go下，我們進去看看。
1func main() {
2 // 沒什麼好說的，參數解析
3 flag.Parse()
4 log.Init(nil) // debug flag: log.dir={path}
5 defer log.Close()
6 log.Info("kratos-demo start")
7 // -conf 參數的解析
8 paladin.Init()
9 // 這裡是 `深坑的入口`
10 // 一會分析
11 _, closeFunc, err := di.InitApp()
12 if err != nil {
13 panic(err)
14 }
15 // os.Signal 是一個系統信號接收channel
16 c := make(chan os.Signal, 1)
17 // syscall 都是一些系統信號
18 signal.Notify(c, syscall.SIGHUP, syscall.SIGQUIT, syscall.SIGTERM, syscall.SIGINT)
19 for {
20 // 一旦有信號進來了，看下面的代碼邏輯，八成是關閉這個應用。
21 s := <-c
22 log.Info("get a signal %s", s.String())
23 switch s {
24 case syscall.SIGQUIT, syscall.SIGTERM, syscall.SIGINT:
25 closeFunc()
26 log.Info("kratos-demo exit")
27 time.Sleep(time.Second)
28 return
29 case syscall.SIGHUP:
30 default:
31 return
32 }
33 }
34}
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()去。
1// Injectors from wire.go:
2func InitApp() (*App, func(), error) {
3
4 // 基本上就是創建一個個實例，和善後它們的函數
5 // 如果途中創建出問題就全體下葬（觸發善後函數）.
6
7 // Redis實例，善後函數
8 redis, cleanup, err := dao.NewRedis()
9 if err != nil {
10 return nil, nil, err
11 }
12
13 // memcache實例，善後函數
14 memcache, cleanup2, err := dao.NewMC()
15 if err != nil {
16 cleanup()
17 return nil, nil, err
18 }
19
20 // 看起來只支持MySQL，善後函數
21 db, cleanup3, err := dao.NewDB()
22 if err != nil {
23 cleanup2()
24 cleanup()
25 return nil, nil, err
26 }
27
28 // 把上面所有的模型對象做一個DAO層封裝
29 daoDao, cleanup4, err := dao.New(redis, memcache, db)
30 if err != nil {
31 cleanup3()
32 cleanup2()
33 cleanup()
34 return nil, nil, err
35 }
36
37 // 這個是個重點，`service`是你的gRPC服務.
38 // 一會我們去分析他
39 serviceService, cleanup5, err := service.New(daoDao)
40 if err != nil {
41 cleanup4()
42 cleanup3()
43 cleanup2()
44 cleanup()
45 return nil, nil, err
46 }
47
48 // 有人會好奇Kratos是怎麼把gRPC和Gin融合在一起的
49 //（沒錯Bilibili的web框架是Gin, 不過這個Gin的一部分核心代碼已經被魔改過了， 在Engine初始化的時候會多加入一個鏈路追蹤的Middleware, 還有一堆路由...）
50 // 秘密就在這裡，等會我們再看
51 engine, err := http.New(serviceService)
52 if err != nil {
53 cleanup5()
54 cleanup4()
55 cleanup3()
56 cleanup2()
57 cleanup()
58 return nil, nil, err
59 }
60
61 // gRPC的初始化的常規操作
62 server, err := grpc.New(serviceService)
63 if err != nil {
64 cleanup5()
65 cleanup4()
66 cleanup3()
67 cleanup2()
68 cleanup()
69 return nil, nil, err
70 }
71
72 // 把上面的服務，engine,gRPC服務,整一塊
73 // 善後函數
74 // 後面稍微分析一下
75 app, cleanup6, err := NewApp(serviceService, engine, server)
76 if err != nil {
77 cleanup5()
78 cleanup4()
79 cleanup3()
80 cleanup2()
81 cleanup()
82 return nil, nil, err
83 }
84
85 // 你可以走了.
86 return app, func() {
87 cleanup6()
88 cleanup5()
89 cleanup4()
90 cleanup3()
91 cleanup2()
92 cleanup()
93 }, nil
94}
95
96//以上代碼全是自動生成，冗餘很正常
接下來我們首先看看serviceService是個什麼東西.(這是什麼魔鬼命名)
進到Service.New(dao)
1
2// Service service.
3type Service struct {
4 // 配置文件映射的Map (這個命名就nm離譜)
5 ac *paladin.Map
6 // 字面意思
7 dao dao.Dao
8}
9
10// New new a service and return.
11func New(d dao.Dao) (s *Service, cf func(), err error) {
12 // 初始化～～～
13 s = &Service{
14 ac: &paladin.TOML{},
15 dao: d,
16 }
17 // 一個關閉的鉤子
18 cf = s.Close
19
20 // 熱加載 application.toml 配置文件
21 // 原理是使用fsnotify監聽文件變更
22 err = paladin.Watch("application.toml", s.ac)
23 return
24}
25
26// -------------- 下面都是你的gRPC業務邏輯-------------
27
28// SayHello grpc demo func.
29func (s *Service) SayHello(ctx context.Context, req *pb.HelloReq) (reply *empty.Empty, err error) {
30 reply = new(empty.Empty)
31 fmt.Printf("hello %s", req.Name)
32 return
33}
34
35// SayHelloURL bm demo func.
36func (s *Service) SayHelloURL(ctx context.Context, req *pb.HelloReq) (reply *pb.HelloResp, err error) {
37 reply = &pb.HelloResp{
38 Content: "hello " + req.Name,
39 }
40 fmt.Printf("hello url %s", req.Name)
41 return
42}
43
44// Ping ping the resource.
45func (s *Service) Ping(ctx context.Context, e *empty.Empty) (*empty.Empty, error) {
46 return &empty.Empty{}, s.dao.Ping(ctx)
47}
48
49// Close close the resource.
50func (s *Service) Close() {}
哇哦，現在我們知道了，Service是由一些gRPC方法，配置項，模型層組成的。
好，乘勝追擊我們再看一看engine, err := http.New(serviceService)做了什麼。
1var svc pb.DemoServer
2
3// New new a bm server.
4func New(s pb.DemoServer) (engine *bm.Engine, err error) {
5 var (
6 cfg bm.ServerConfig
7 ct paladin.TOML
8 )
9 // 讀取你的配置文件
10 if err = paladin.Get("http.toml").Unmarshal(&ct); err != nil {
11 return
12 }
13 // 得到http.toml的Server字段
14 if err = ct.Get("Server").UnmarshalTOML(&cfg); err != nil {
15 return
16 }
17 svc = s
18
19 // 做一個新 engine
20 // (engine 是 Gin 裡的模塊，這裡我就不分析Gin的源碼了)
21 engine = bm.DefaultServer(&cfg)
22
23 // 將gRPC服務註冊到engine, 這個代碼註冊代碼是bm自己生成的
24 // 一會我們分析
25 pb.RegisterDemoBMServer(engine, s)
26
27 // 把你的路由搞進去
28 initRouter(engine)
29
30 // 開始跑
31 err = engine.Start()
32 return
33}
34
35// 路由在這裡登記！
36func initRouter(e *bm.Engine) {
37 e.Ping(ping)
38 g := e.Group("/kratos-demo")
39 {
40 g.GET("/start", howToStart)
41 }
42}
43
44func ping(ctx *bm.Context) {
45 if _, err := svc.Ping(ctx, nil); err != nil {
46 log.Error("ping error(%v)", err)
47 ctx.AbortWithStatus(http.StatusServiceUnavailable)
48 }
49}
50
51// example for http request handler.
52func howToStart(c *bm.Context) {
53 k := &model.Kratos{
54 Hello: "Golang 大法好 !!!我好你個頭！",
55 }
56 c.JSON(k, nil)
57}
如果你使用過gin這個web框架, 上面的代碼你一定很熟悉，對吧？
bm就是gin，只是部分代碼被Bilibili魔改了，整體架構是不變的。
OK，我們看看RegisterDemoBMServer裡做了什麼.
1// DemoBMServer is the server API for Demo service.
2type DemoBMServer interface {
3 Ping(ctx context.Context, req *google_protobuf1.Empty) (resp *google_protobuf1.Empty, err error)
4
5 SayHello(ctx context.Context, req *HelloReq) (resp *google_protobuf1.Empty, err error)
6
7 SayHelloURL(ctx context.Context, req *HelloReq) (resp *HelloResp, err error)
8}
9
10// 我們寫的gRPC服務
11var DemoSvc DemoBMServer
12// ------------------------------------------------
13
14// 我們仔細分析這些方法不難發現
15// 他們都會調用 `BindWith` 和對應的gRPC方法
16
17// 先使用BindWith: 將request中的`Body` 轉化為go中的 `struct`
18// 然後使用gRPC方法處理請求數據
19// 最後返回
20
21// 本質就是通過http調用gRPC服務
22
23func demoPing(c *bm.Context) {
24 p := new(google_protobuf1.Empty)
25 if err := c.BindWith(p, binding.Default(c.Request.Method, c.Request.Header.Get("Content-Type"))); err != nil {
26 return
27 }
28 resp, err := DemoSvc.Ping(c, p)
29 c.JSON(resp, err)
30}
31
32func demoSayHello(c *bm.Context) {
33 p := new(HelloReq)
34 if err := c.BindWith(p, binding.Default(c.Request.Method, c.Request.Header.Get("Content-Type"))); err != nil {
35 return
36 }
37 resp, err := DemoSvc.SayHello(c, p)
38 c.JSON(resp, err)
39}
40
41func demoSayHelloURL(c *bm.Context) {
42 p := new(HelloReq)
43 if err := c.BindWith(p, binding.Default(c.Request.Method, c.Request.Header.Get("Content-Type"))); err != nil {
44 return
45 }
46 resp, err := DemoSvc.SayHelloURL(c, p)
47 c.JSON(resp, err)
48}
49//-------------------------------------
50
51// RegisterDemoBMServer Register the blademaster route
52func RegisterDemoBMServer(e *bm.Engine, server DemoBMServer) {
53 // server 是我們之前編寫的gRPC服務
54 DemoSvc = server
55
56 // 將一些方法註冊到路由裡去
57 e.GET("/demo.service.v1.Demo/Ping", demoPing)
58 e.GET("/demo.service.v1.Demo/SayHello", demoSayHello)
59 e.GET("/kratos-demo/say_hello", demoSayHelloURL)
60}
哇，原來只是把一些gRPC的服務綁定到gin的路由裡了呀。
借用gin來調用gRPC.
grpc.New()就不分析了。
然後是AppNew()
1//go:generate kratos tool wire
2type App struct {
3 svc *service.Service
4 http *bm.Engine
5 grpc *warden.Server
6}
7
8func NewApp(svc *service.Service, h *bm.Engine, g *warden.Server) (app *App, closeFunc func(), err error) {
9 app = &App{
10 svc: svc,
11 http: h,
12 grpc: g,
13 }
14
15 // 一個關閉context的回調
16 closeFunc = func() {
17 ctx, cancel := context.WithTimeout(context.Background(), 35*time.Second)
18 if err := g.Shutdown(ctx); err != nil {
19 log.Error("grpcSrv.Shutdown error(%v)", err)
20 }
21 if err := h.Shutdown(ctx); err != nil {
22 log.Error("httpSrv.Shutdown error(%v)", err)
23 }
24 cancel()
25 }
26 return
27}
到這裡初始化是結束了。
Summary
kratos的初始化流程:
- 讀取配置文件
- 實例化Dao層
- 實例化gRPC服務
- 實例化gin的engine
- 註冊gPRC到engine
- 啟動engine
- 啟動gRPC服務端
- 獲得整個程序關閉的回調
我分得應該還是比較細的。
後面應該還會分析warden，它是Kratos在grpc原版上的一個封裝版本。
溜了溜了…