1
2

// Middleware is a chainable behavior modifier for endpoints.
type Middleware func(Endpoint) Endpoint

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

package main

import (
	"context"
	"errors"
	"github.com/go-kit/kit/endpoint"
	"golang.org/x/time/rate"
)

var ErrLimitExceed = errors.New("Rate limit exceed!")


// NewTokenBucketLimitterWithBuildIn 使用x/time/rate创建限流中间件
func NewTokenBucketLimiter(bkt *rate.Limiter) endpoint.Middleware {
	return func(next endpoint.Endpoint) endpoint.Endpoint {
		return func(ctx context.Context, request interface{}) (response interface{}, err error) {
			if !bkt.Allow() {
				return nil, ErrLimitExceed
			}
			return next(ctx, request)
		}
	}
}

1
2
3

//add ratelimit,refill every second,set capacity 3
rateBucket := rate.NewLimiter(rate.Every(time.Second*1), 3)
endpoint = NewTokenBucketLimiter(rateBucket)(endpoint)

1
2
3
4
5
6

import	kitratelim "github.com/go-kit/kit/ratelimit"

//add ratelimit,refill every second,set capacity 3
rateBucket := rate.NewLimiter(rate.Every(time.Second*1), 3)
endpoint = kitratelim.NewErroringLimiter(rateBucket)(endpoint)
	

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

import	"github.com/go-kit/kit/circuitbreaker"

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)

1
2
3

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

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

// Hystrix returns an endpoint.Middleware that implements the circuit
// breaker pattern using the afex/hystrix-go package.
//
// When using this circuit breaker, please configure your commands separately.
//
// See https://godoc.org/github.com/afex/hystrix-go/hystrix for more
// information.
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
		}
	}
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22

func main() {
	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)
	//现在我们可以通过 endPoint 发起请求了
	req := struct{}{}
	for i := 1; i <= 20; i++ {
		if _, err = reqEndPoint(ctx, req); err != nil {
			fmt.Println("当前时间: ", time.Now().Format("2006-01-02 15:04:05.99"))
			fmt.Println(err)
			time.Sleep(1 * time.Second)
		}
	}
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

import	kithttp "github.com/go-kit/kit/transport/http"

options := []kithttp.ServerOption{
	kithttp.ServerErrorHandler(transport.NewLogErrorHandler(logger)),
	kithttp.ServerErrorEncoder(kithttp.DefaultErrorEncoder), //错误处理
	zipkinServer,
}

http.Handle("/calculate", kithttp.NewServer(
	endpoint,
	decodeArithmeticRequest,
	encodeArithmeticResponse,
	options...,
))

1

kithttp.ServerErrorEncoder(kithttp.DefaultErrorEncoder), //错误处理

 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

// DefaultErrorEncoder writes the error to the ResponseWriter, by default a
// content type of text/plain, a body of the plain text of the error, and a
// status code of 500. If the error implements Headerer, the provided headers
// will be applied to the response. If the error implements json.Marshaler, and
// the marshaling succeeds, a content type of application/json and the JSON
// encoded form of the error will be used. If the error implements StatusCoder,
// the provided StatusCode will be used instead of 500.
func DefaultErrorEncoder(_ context.Context, err error, w http.ResponseWriter) {
	contentType, body := "text/plain; charset=utf-8", []byte(err.Error())
	if marshaler, ok := err.(json.Marshaler); ok {
		if jsonBody, marshalErr := marshaler.MarshalJSON(); marshalErr == nil {
			contentType, body = "application/json; charset=utf-8", jsonBody
		}
	}
	w.Header().Set("Content-Type", contentType)
	if headerer, ok := err.(Headerer); ok {
		for k, values := range headerer.Headers() {
			for _, v := range values {
				w.Header().Add(k, v)
			}
		}
	}
	code := http.StatusInternalServerError //500
	if sc, ok := err.(StatusCoder); ok {
		code = sc.StatusCode()
	}
	w.WriteHeader(code)
	w.Write(body)
}