go-kit组件使用hystrix中间件的操作

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
 }
}

package main

import (
 "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)
}

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  
      }  
   }  
}

commandName := "my-endpoint"  
hystrix.ConfigureCommand(commandName, hystrix.CommandConfig{  
  Timeout: 1000 * 30,  
  ErrorPercentThreshold: 1,  
  SleepWindow: 10000,  
  MaxConcurrentRequests: 1000,  
  RequestVolumeThreshold: 5,  
})

breakerMw := circuitbreaker.Hystrix(commandName)
//增加熔断中间件  
reqEndPoint = breakerMw(reqEndPoint)

 
syntax = "proto3"; 
// 请求书详情的参数结构  book_id 32位整形
message BookInfoParams {
    int32 book_id = 1;
} 
 
// 书详情信息的结构   book_name字符串类型
message BookInfo {
    int32 book_id = 1;
    string  book_name = 2;
}
 
// 请求书列表的参数结构  page、limit   32位整形
message BookListParams {
    int32 page = 1;
    int32 limit = 2;
} 
 
// 书列表的结构    BookInfo结构数组
message BookList {
    repeated BookInfo book_list = 1;
}
// 定义 获取书详情  和 书列表服务   入参出参分别为上面所定义的结构
service BookService {
    rpc GetBookInfo (BookInfoParams) returns (BookInfo) {}
    rpc GetBookList (BookListParams) returns (BookList) {}
}

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 (
		//注册中心地址
		etcdServer = "127.0.0.1:2379"
		//监听的服务前缀
		prefix = "/services/book/"
		ctx    = context.Background()
	)
	//对hystrix进行配置
	commandName:="my_endpoint"
	hystrix.ConfigureCommand(commandName,hystrix.CommandConfig{
		Timeout:1000*3, //超时
		MaxConcurrentRequests:100, //最大并发的请求数
		RequestVolumeThreshold:5,//请求量阈值
		SleepWindow:10000, //熔断开启多久尝试发起一次请求
		ErrorPercentThreshold:1, //误差阈值百分比
	})
	breakerMw:=circuitbreaker.Hystrix(commandName) //定义熔断器中间件
	options := etcdv3.ClientOptions{
		DialTimeout:   time.Second * 3,
		DialKeepAlive: time.Second * 3,
	}
	//连接注册中心
	client, err := etcdv3.NewClient(ctx, []string{etcdServer}, options)
	if err != nil {
		panic(err)
	}
	logger := log.NewNopLogger()
	//创建实例管理器, 此管理器会Watch监听etc中prefix的目录变化更新缓存的服务实例数据
	instancer, err := etcdv3.NewInstancer(client, prefix, logger)
	if err != nil {
		panic(err)
	}
	//创建端点管理器， 此管理器根据Factory和监听的到实例创建endPoint并订阅instancer的变化动态更新Factory创建的endPoint
	endpointer := sd.NewEndpointer(instancer, reqFactory, logger) //reqFactory自定义的函数，主要用于端点层（endpoint）接受并显示数据
	//创建负载均衡器
	balancer := lb.NewRoundRobin(endpointer)
 
	/**
	我们可以通过负载均衡器直接获取请求的endPoint，发起请求
	reqEndPoint,_ := balancer.Endpoint()
	*/
 
	/**
	也可以通过retry定义尝试次数进行请求
	*/
	reqEndPoint := lb.Retry(3, 100*time.Second, balancer) //请求次数为3，时间为10S（时间需要多于服务器限流时间3s）
 
	//增加熔断中间件
	reqEndPoint=breakerMw(reqEndPoint)
 
	//现在我们可以通过 endPoint 发起请求了
	req := struct{}{}
	for i:=0;i<20;i++ {  //发生20次请求
		ctx=context.Background()
		if _, err = reqEndPoint(ctx, req); err != nil {
			//panic(err)
			fmt.Println("当前时间: ", time.Now().Format("2006-01-02 15:04:05.99"),"\t第",i+1,"次")
			fmt.Println(err)
			time.Sleep(1*time.Second)
		}
	}
}
 
//通过传入的 实例地址  创建对应的请求endPoint
func 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("请求服务成功: ", instanceAddr,"当前时间为:",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
}