使用go-kit中間件時(shí)，一般在endpoint中進(jìn)行中間件的開發(fā)。
在endpoint層插入hystrix中間件的插入。
endpoint.go
|
1
2
3
4
5
6
7
|
func MakeEndpoint (svc services.StringService) endpoint.Endpoint { return func(ctx context.Context, request interface{}) (response interface{}, err error) { req := request.(*StringService.Request) rep , err := svc.Diff(ctx , req) return rep , err }} |
go-kit使我們更注重對(duì)服務(wù)邏輯的開發(fā)，對(duì)中間過程的省略會(huì)減少很多的錯(cuò)誤發(fā)生。
main.go
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
|
package mainimport ( "balencegrpc/services" "balencegrpc/router" "balencegrpc/proto" "balencegrpc/discover" "balencegrpc/endpoints" "github.com/go-kit/kit/circuitbreaker" "google.golang.org/grpc" "google.golang.org/grpc/health/grpc_health_v1" "log" "net")func main() { //svc := new() svc := services.ServiceI{} endpoint := circuitbreaker.Hystrix("testname")(endpoints.MakeEndpoint(svc)) lis , err := net.Listen("tcp" , ":8081") if err != nil { log.Println(err) return } router := router.NewRouter(svc , endpoint) grpcserver := grpc.NewServer() c := discover.Service{} grpc_health_v1.RegisterHealthServer(grpcserver , &c) StringService.RegisterStringServiceServer(grpcserver , router) rs := discover.NewService() rs.Register("127.0.0.1" , 8081) grpcserver.Serve(lis)} |
我們?cè)趍ain.go中進(jìn)行添加hystrix對(duì)endpoint進(jìn)行封裝，，使用hystrix進(jìn)行服務(wù)熔斷的處理。其實(shí)使用go-kit封裝的。也可以自行封裝一個(gè)中間件，在創(chuàng)建endpoint時(shí)進(jìn)行封裝
補(bǔ)充：go-kit微服務(wù)熔斷機(jī)制的實(shí)現(xiàn)
在微服務(wù)架構(gòu)中，每一個(gè)微服務(wù)都是一個(gè)獨(dú)立的業(yè)務(wù)功能單元，而一個(gè)應(yīng)用一般由多個(gè)微服務(wù)組成，微服務(wù)之間的交互是通過RPC（遠(yuǎn)程過程調(diào)用）完成。
比如，我們的應(yīng)用是微服務(wù)A調(diào)用微服務(wù)B和微服務(wù)C來完成的，而微服務(wù)B又需要調(diào)用微服務(wù)D，微服務(wù)D又需要調(diào)用微服務(wù)E。如果在調(diào)用的鏈路上對(duì)微服務(wù)E的調(diào)用，響應(yīng)時(shí)間過長(zhǎng)或者服務(wù)不可用，那么對(duì)微服務(wù)D的調(diào)用就會(huì)占用越來越多的系統(tǒng)資源，進(jìn)而引起微服務(wù)D的系統(tǒng)崩潰，微服務(wù)D的不可用，又會(huì)連鎖反應(yīng)的引起微服務(wù)B崩潰，進(jìn)而微服務(wù)A崩潰，最終導(dǎo)致整個(gè)應(yīng)用不可用。這也就是所謂的“雪崩效應(yīng)”。
介紹
go-kit 提供了三種熔斷
1、 gobreaker
2、 handy
3、 hystrix-go
hystrix用的比較多，我們來介紹下go-kit中hystrix的使用方法
go-kit的hystrix
Middleware的實(shí)現(xiàn)
1、 Hystrix返回Middleware 此中間件會(huì)在原來的endPoint包一層Hystrix的endPoint
2、 hystrix通過傳入的commanName獲取對(duì)應(yīng)的Hystrix的設(shè)置，并設(shè)置run失敗時(shí)運(yùn)行的fallback函數(shù)為nil
3、 我們也可以自己實(shí)現(xiàn)middleware包裝endPoint
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
|
func Hystrix(commandName string) endpoint.Middleware { return func(next endpoint.Endpoint) endpoint.Endpoint { return func(ctx context.Context, request interface{}) (response interface{}, err error) { var resp interface{} if err := hystrix.Do(commandName, func() (err error) { resp, err = next(ctx, request) return err }, nil); err != nil { return nil, err } return resp, nil } } } |
客戶端hystrix配置
1、Timeout 【請(qǐng)求超時(shí)的時(shí)間】
2、ErrorPercentThreshold【允許出現(xiàn)的錯(cuò)誤比例】
3、SleepWindow【熔斷開啟多久嘗試發(fā)起一次請(qǐng)求】
4、MaxConcurrentRequests【允許的最大并發(fā)請(qǐng)求數(shù)】
5、RequestVolumeThreshold 【波動(dòng)期內(nèi)的最小請(qǐng)求數(shù)，默認(rèn)波動(dòng)期10S】
|
1
2
3
4
5
6
7
8
|
commandName := "my-endpoint" hystrix.ConfigureCommand(commandName, hystrix.CommandConfig{ Timeout: 1000 * 30, ErrorPercentThreshold: 1, SleepWindow: 10000, MaxConcurrentRequests: 1000, RequestVolumeThreshold: 5, }) |
增加熔斷中間件的包裝
|
1
2
3
|
breakerMw := circuitbreaker.Hystrix(commandName)//增加熔斷中間件 reqEndPoint = breakerMw(reqEndPoint) |
實(shí)例
1、protobuf文件及生成對(duì)應(yīng)的go文件
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
|
syntax = "proto3"; // 請(qǐng)求書詳情的參數(shù)結(jié)構(gòu) book_id 32位整形message BookInfoParams { int32 book_id = 1;} // 書詳情信息的結(jié)構(gòu) book_name字符串類型message BookInfo { int32 book_id = 1; string book_name = 2;} // 請(qǐng)求書列表的參數(shù)結(jié)構(gòu) page、limit 32位整形message BookListParams { int32 page = 1; int32 limit = 2;} // 書列表的結(jié)構(gòu) BookInfo結(jié)構(gòu)數(shù)組message BookList { repeated BookInfo book_list = 1;}// 定義 獲取書詳情 和 書列表服務(wù) 入?yún)⒊鰠⒎謩e為上面所定義的結(jié)構(gòu)service BookService { rpc GetBookInfo (BookInfoParams) returns (BookInfo) {} rpc GetBookList (BookListParams) returns (BookList) {}} |
生成對(duì)應(yīng)的go語言代碼文件：protoc --go_out=plugins=grpc:. book.proto （其中：protobuf文件名為：book.proto）
注：由于演示熔斷機(jī)制，也就是Server出現(xiàn)問題的時(shí)候進(jìn)行熔斷，因此本文Server端代碼可以不用。
2、Client端代碼
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
|
package main import ( "MyKit" "context" "fmt" "github.com/afex/hystrix-go/hystrix" "github.com/go-kit/kit/circuitbreaker" "github.com/go-kit/kit/endpoint" "github.com/go-kit/kit/log" "github.com/go-kit/kit/sd" "github.com/go-kit/kit/sd/etcdv3" "github.com/go-kit/kit/sd/lb" "google.golang.org/grpc" "io" "time") func main() { var ( //注冊(cè)中心地址 etcdServer = "127.0.0.1:2379" //監(jiān)聽的服務(wù)前綴 prefix = "/services/book/" ctx = context.Background() ) //對(duì)hystrix進(jìn)行配置 commandName:="my_endpoint" hystrix.ConfigureCommand(commandName,hystrix.CommandConfig{ Timeout:1000*3, //超時(shí) MaxConcurrentRequests:100, //最大并發(fā)的請(qǐng)求數(shù) RequestVolumeThreshold:5,//請(qǐng)求量閾值 SleepWindow:10000, //熔斷開啟多久嘗試發(fā)起一次請(qǐng)求 ErrorPercentThreshold:1, //誤差閾值百分比 }) breakerMw:=circuitbreaker.Hystrix(commandName) //定義熔斷器中間件 options := etcdv3.ClientOptions{ DialTimeout: time.Second * 3, DialKeepAlive: time.Second * 3, } //連接注冊(cè)中心 client, err := etcdv3.NewClient(ctx, []string{etcdServer}, options) if err != nil { panic(err) } logger := log.NewNopLogger() //創(chuàng)建實(shí)例管理器, 此管理器會(huì)Watch監(jiān)聽etc中prefix的目錄變化更新緩存的服務(wù)實(shí)例數(shù)據(jù) instancer, err := etcdv3.NewInstancer(client, prefix, logger) if err != nil { panic(err) } //創(chuàng)建端點(diǎn)管理器， 此管理器根據(jù)Factory和監(jiān)聽的到實(shí)例創(chuàng)建endPoint并訂閱instancer的變化動(dòng)態(tài)更新Factory創(chuàng)建的endPoint endpointer := sd.NewEndpointer(instancer, reqFactory, logger) //reqFactory自定義的函數(shù)，主要用于端點(diǎn)層（endpoint）接受并顯示數(shù)據(jù) //創(chuàng)建負(fù)載均衡器 balancer := lb.NewRoundRobin(endpointer) /** 我們可以通過負(fù)載均衡器直接獲取請(qǐng)求的endPoint，發(fā)起請(qǐng)求 reqEndPoint,_ := balancer.Endpoint() */ /** 也可以通過retry定義嘗試次數(shù)進(jìn)行請(qǐng)求 */ reqEndPoint := lb.Retry(3, 100*time.Second, balancer) //請(qǐng)求次數(shù)為3，時(shí)間為10S（時(shí)間需要多于服務(wù)器限流時(shí)間3s） //增加熔斷中間件 reqEndPoint=breakerMw(reqEndPoint) //現(xiàn)在我們可以通過 endPoint 發(fā)起請(qǐng)求了 req := struct{}{} for i:=0;i<20;i++ { //發(fā)生20次請(qǐng)求 ctx=context.Background() if _, err = reqEndPoint(ctx, req); err != nil { //panic(err) fmt.Println("當(dāng)前時(shí)間: ", time.Now().Format("2006-01-02 15:04:05.99"),"\t第",i+1,"次") fmt.Println(err) time.Sleep(1*time.Second) } }} //通過傳入的 實(shí)例地址 創(chuàng)建對(duì)應(yīng)的請(qǐng)求endPointfunc reqFactory(instanceAddr string) (endpoint.Endpoint, io.Closer, error) { return func(ctx context.Context, request interface{}) (interface{}, error) { conn, err := grpc.Dial(instanceAddr, grpc.WithInsecure()) if err != nil { fmt.Println(err) panic("connect error") } defer conn.Close() bookClient := book.NewBookServiceClient(conn) bi, _ := bookClient.GetBookInfo(context.Background(), &book.BookInfoParams{BookId: 1}) fmt.Println("獲取書籍詳情") fmt.Println("bookId: 1", " => ", "bookName:", bi.BookName) fmt.Println("請(qǐng)求服務(wù)成功: ", instanceAddr,"當(dāng)前時(shí)間為:",time.Now().Format("2006-01-02 15:04:05.99")) /*bl, _ := bookClient.GetBookList(context.Background(), &book.BookListParams{Page: 1, Limit: 10}) fmt.Println("獲取書籍列表") for _, b := range bl.BookList { fmt.Println("bookId:", b.BookId, " => ", "bookName:", b.BookName) }*/ return nil, nil }, nil, nil} |
3、運(yùn)行及分析
直接運(yùn)行Client端（不用啟動(dòng)etcd、Server），效果如下：
通過上面的輸出記錄可以驗(yàn)證我們的配置：
1、 前5條波動(dòng)期內(nèi)的錯(cuò)誤，沒有觸發(fā)circuit開啟（RequestVolumeThreshold:5,//請(qǐng)求量閾值）
2、 circuit開啟后請(qǐng)求熔斷生效（輸出內(nèi)容：hystrix: circuit open）
3、 circuit開啟10S后，SleepWindow測(cè)試發(fā)起請(qǐng)求設(shè)置生效（第16次輸出的內(nèi)容；設(shè)置：SleepWindow:10000, //熔斷開啟多久嘗試發(fā)起一次請(qǐng)求）
以上為個(gè)人經(jīng)驗(yàn)，希望能給大家一個(gè)參考，也希望大家多多支持服務(wù)器之家。如有錯(cuò)誤或未考慮完全的地方，望不吝賜教。
原文鏈接：https://blog.csdn.net/Xiang_lhh/article/details/115495321
