go-sentinel流量控制(五)：github优秀开源项目源码分析-六虎

go-admin 怎么运用go-sentinel

go-admin
依据Gin + Vue + Element UI & Arco Design & Ant Design 的前后端别离权限办理体系脚手架（包含了：多租户的支持，根底用户办理功用，jwt鉴权，代码生成器，RBAC资源操控，表单构建，守时任务等

界说中间件

源码在/common/middleware/sentinel.go

Sentinel 体系BBR自适应限流从全体维度对运用进口流量进行操控，结合运用的 Load、CPU 运用率、总体均匀 RT、进口 QPS 和并发线程数等几个维度的监控目标，经过自适应的流控战略，让体系的进口流量和体系的负载达到一个平衡，让体系尽可能跑在最大吞吐量的一起保证体系全体的稳定性。

package middleware
import (
	"github.com/alibaba/sentinel-golang/core/system"
	sentinel "github.com/alibaba/sentinel-golang/pkg/adapters/gin"
	"github.com/gin-gonic/gin"
	log "github.com/go-admin-team/go-admin-core/logger"
)
// Sentinel 限流
func Sentinel() gin.HandlerFunc {
	if _, err := system.LoadRules([]*system.Rule{
		{
			MetricType:   system.InboundQPS,
			TriggerCount: 200,
			Strategy:     system.BBR,
		},
	}); err != nil {
		log.Fatalf("Unexpected error: %+v", err)
	}
	return sentinel.SentinelMiddleware(
		sentinel.WithBlockFallback(func(ctx *gin.Context) {
			ctx.AbortWithStatusJSON(200, map[string]interface{}{
				"msg":  "too many request; the quota used up!",
				"code": 500,
			})
		}),
	)
}

限流规矩: 在Sentinel()函数内部，运用system.LoadRules函数界说了限流规矩：

MetricType设定为system.InboundQPS，表明约束恳求的速率。
TriggerCount设置为 200，表明每秒答应的恳求数量为 200。
Strategy设定为system.BBR，这是一种流量操控战略，表明运用 BBR 算法进行限流。

回调函数: 运用sentinel.SentinelMiddleware函数创建了一个 Gin 中间件，其中包含了一个WithBlockFallback选项，该选项界说了当恳求被限流时的回调函数。在这个回调函数中，会回来一个 JSON 呼应，提示客户端恳求过多，已经达到配额上限。

运用中间件

/cmd/api/server.go

func initRouter() {
	var r *gin.Engine
	h := sdk.Runtime.GetEngine()
	if h == nil {
		h = gin.New()
		sdk.Runtime.SetEngine(h)
	}
	switch h.(type) {
	case *gin.Engine:
		r = h.(*gin.Engine)
	default:
		log.Fatal("not support other engine")
		//os.Exit(-1)
	}
	if config.SslConfig.Enable {
		r.Use(handler.TlsHandler())
	}
	//r.Use(middleware.Metrics())
	r.Use(common.Sentinel()).
		Use(common.RequestId(pkg.TrafficKey)).
		Use(api.SetRequestLogger)
	common.InitMiddleware(r)
}

jupiter 怎么运用go-sentinel

jupiter
Jupiter 是斗鱼开源的一套微服务办理框架，供给丰富的后台功用，办理运用的资源、装备，运用的功能、装备等可视化

界说装备

/pkg/core/sentinel/config.go

对官方库的二次封装。支持从 ETCD 数据源和其他数据源加载规矩，并进行相应的初始化作业

type Config struct { ... }: 界说了 Sentinel 的装备结构体Config，其中包含了一系列装备项，如启用状况、数据源、熔断降级规矩、流量操控规矩和体系维护规矩等。
func StdConfig() Config { ... }: 供给了一个规范的装备获取函数，用于获取规范的 Sentinel 装备。
func RawConfig(key string) Config { ... }: 依据给定的装备键获取原始装备，假如装备不存在或解析失利，则回来默许装备。
func DefaultConfig() Config { ... }: 回来默许的 Sentinel 装备。
func (e Config) exitHandler(entry *SentinelEntry, ctx *EntryContext) error { ... }: 界说了 Sentinel 规矩履行完成后的退出处理函数，用于计算履行结果并处理反常。
func (c Config) Entry(resource string, opts ...EntryOption) (*SentinelEntry, *BlockError) { ... }: 依据给定的资源和选项创建 Sentinel 规矩的进口。
func (c Config) Build() error { ... }: 构建 Sentinel 装备，主要进行 Sentinel 的初始化作业，包含加载规矩和注册状况监听器。
func (c Config) loadRules() { ... }: 加载 Sentinel 规矩，依据装备的数据源加载对应的规矩。
func checkSrcComplianceJson(src []byte) (bool, error) { ... }: 检查 JSON 格局的数据源是否合规。
func initRules(client *clientv3.Client, key string, h datasource.PropertyHandler) error { ... }: 依据给定的客户端和键初始化规矩。
func circuitBreakerRuleJsonArrayParser(src []byte) (interface{}, error) { ... }: 解析 JSON 格局的熔断降级规矩。

package sentinel
import (
	"encoding/json"
	"fmt"
	sentinel "github.com/alibaba/sentinel-golang/api"
	"github.com/alibaba/sentinel-golang/core/base"
	"github.com/alibaba/sentinel-golang/core/circuitbreaker"
	"github.com/alibaba/sentinel-golang/core/config"
	"github.com/alibaba/sentinel-golang/core/flow"
	"github.com/alibaba/sentinel-golang/core/system"
	"github.com/alibaba/sentinel-golang/ext/datasource"
	"github.com/douyu/jupiter/pkg"
	"github.com/douyu/jupiter/pkg/client/etcdv3"
	"github.com/douyu/jupiter/pkg/conf"
	"github.com/douyu/jupiter/pkg/core/constant"
	"github.com/douyu/jupiter/pkg/xlog"
	clientv3 "go.etcd.io/etcd/client/v3"
	"go.uber.org/zap"
)
type Config struct {
	Enable     bool   `toml:"enable"`
	Datasource string `toml:"datasource"`
	EtcdRawKey string `toml:"etcdRawKey"`
	// 熔断降级
	CbKey   string                `toml:"cbKey"`
	CbRules []*CircuitBreakerRule `toml:"cbRules"`
	// 流量操控
	FlowKey   string       `toml:"flowKey"`
	FlowRules []*flow.Rule `toml:"flowRules"`
	// 体系维护
	SystemKey   string         `toml:"systemKey"`
	SystemRules []*system.Rule `toml:"systemRules"`
}
func StdConfig() Config {
	return RawConfig(constant.ConfigKey("sentinel"))
}
func RawConfig(key string) Config {
	config := DefaultConfig()
	if conf.Get(constant.ConfigKey("sentinel")) == nil {
		return config
	}
	if err := conf.UnmarshalKey(key, &config, conf.TagName("toml")); err != nil {
		xlog.Jupiter().Warn("unmarshal config", zap.String("key", key), zap.Error(err))
		return config
	}
	return config
}
func DefaultConfig() Config {
	return Config{
		Enable:     false,
		Datasource: SENTINEL_DATASOURCE_ETCD,
		EtcdRawKey: "app.registry.etcd",
		// 熔断降级规矩 /wsd-sentinel/{language}/{app}/{idc}/{env}/{ruleType}=${value}
		CbKey: "/wsd-sentinel/go/%s/%s/%s/degrade",
		// 流量操控规矩 /wsd-sentinel/{language}/{app}/{idc}/{env}/{ruleType}=${value}
		FlowKey: "/wsd-sentinel/go/%s/%s/%s/flow",
		// 体系维护规矩 /wsd-sentinel/{language}/{app}/{idc}/{env}/{ruleType}=${value}
		SystemKey: "/wsd-sentinel/go/%s/%s/%s/system",
	}
}
func (e Config) exitHandler(entry *SentinelEntry, ctx *EntryContext) error {
	if ctx.Err() != nil {
		sentinelExceptionsThrown.WithLabelValues(labels(entry.Resource().Name())...).Inc()
	} else {
		sentinelSuccess.WithLabelValues(labels(entry.Resource().Name())...).Inc()
	}
	sentinelRt.WithLabelValues(labels(entry.Resource().Name())...).Observe(float64(ctx.Rt()) / 1000)
	return ctx.Err()
}
func (c Config) Entry(resource string, opts ...EntryOption) (*SentinelEntry, *BlockError) {
	if !c.Enable {
		return base.NewSentinelEntry(nil, nil, nil), nil
	}
	a, b := sentinel.Entry(resource, opts...)
	sentinelReqeust.WithLabelValues(labels(resource)...).Inc()
	if b != nil {
		sentinelBlocked.WithLabelValues(labels(resource)...).Inc()
		return a, b
	}
	a.WhenExit(c.exitHandler)
	return a, b
}
func (c Config) Build() error {
	if !c.Enable {
		xlog.Jupiter().Info("disable sentinel feature")
		return nil
	}
	if err := sentinel.InitDefault(); err != nil {
		xlog.Jupiter().Error("sentinel.InitDefault failed", zap.Error(err))
		return err
	}
	defaultConfig := config.NewDefaultConfig()
	defaultConfig.Sentinel.App.Name = pkg.Name()
	err := sentinel.InitWithConfig(defaultConfig)
	if err != nil {
		return err
	}
	circuitbreaker.RegisterStateChangeListeners(&stateChangeTestListener{})
	c.loadRules()
	return nil
}
func (c Config) loadRules() {
	xlog.Jupiter().Info("load sentinel rules", zap.String("datasource", c.Datasource))
	switch c.Datasource {
	case SENTINEL_DATASOURCE_ETCD:
		cli, err := etcdv3.RawConfig(c.EtcdRawKey).Singleton()
		if err != nil {
			panic(err)
		}
		err = initRules(cli.Client, c.CbKey, datasource.NewCircuitBreakerRulesHandler(circuitBreakerRuleJsonArrayParser))
		if err != nil {
			xlog.Jupiter().Warn("sentinel etcd Initialize failed", xlog.FieldErr(err))
		}
		err = initRules(cli.Client, c.SystemKey, datasource.NewSystemRulesHandler(datasource.SystemRuleJsonArrayParser))
		if err != nil {
			xlog.Jupiter().Warn("sentinel etcd Initialize failed", xlog.FieldErr(err))
		}
		err = initRules(cli.Client, c.FlowKey, datasource.NewFlowRulesHandler(datasource.FlowRuleJsonArrayParser))
		if err != nil {
			xlog.Jupiter().Warn("sentinel etcd Initialize failed", xlog.FieldErr(err))
		}
	default:
		var err error
		_, err = flow.LoadRules(c.FlowRules)
		if err != nil {
			xlog.Jupiter().Warn("sentinel flow.LoadRules failed", xlog.FieldErr(err))
		}
		_, err = system.LoadRules(c.SystemRules)
		if err != nil {
			xlog.Jupiter().Warn("sentinel system.LoadRules failed", xlog.FieldErr(err))
		}
		rules := convertCbRules(c.CbRules)
		_, err = circuitbreaker.LoadRules(rules)
		if err != nil {
			xlog.Jupiter().Warn("sentinel circuitbreaker.LoadRules failed", xlog.FieldErr(err))
		}
	}
}
func checkSrcComplianceJson(src []byte) (bool, error) {
	if len(src) == 0 {
		return false, nil
	}
	return true, nil
}
func initRules(client *clientv3.Client, key string, h datasource.PropertyHandler) error {
	datasource, err := newDataSource(client,
		fmt.Sprintf(key, pkg.Name(), pkg.AppZone(), conf.GetString("app.mode")),
		h)
	if err != nil {
		return err
	}
	return datasource.Initialize()
}
func circuitBreakerRuleJsonArrayParser(src []byte) (interface{}, error) {
	if valid, err := checkSrcComplianceJson(src); !valid {
		return nil, err
	}
	rules := make([]*CircuitBreakerRule, 0, 8)
	if err := json.Unmarshal(src, &rules); err != nil {
		desc := fmt.Sprintf("Fail to convert source bytes to []*CircuitBreakerRule, err: %s", err.Error())
		xlog.Jupiter().Warn("json.Unmarshal", zap.ByteString("src", src), zap.Error(err))
		return nil, datasource.NewError(datasource.ConvertSourceError, desc)
	}
	xlog.Jupiter().Info("circuitBreakerRuleJsonArrayParser finished", zap.Any("rules", rules))
	return convertCbRules(rules), nil
}

SentinelConfig

[jupiter.sentinel]
  enable = true
  datasource = "files"
  cbKey = "/wsd-sentinel/go/%s/%s/%s/degrade"
  etcdRawKey = "app.registry.etcd"
[[jupiter.sentinel.cbRules]]
  enable = true
  resource = "recomend"
  strategy = 0
  retryTimeoutMs = 3000
  minRequestAmount = 10
  statIntervalMs = 5000
  maxAllowedRtMs = 1000
  statSlidingWindowBucketCount = 10
  threshold = 0.4

装备项

sentinel详细字段解析：

enable：表明是否开启熔断降级功用，默许为 false
datasource：表明规矩来源类型，可选etcd和files，默许是 etcd
cbKey：熔断降级etcd的key规矩，只在datasource为etcd时有效，默许为/wsd-sentinel/go/%s/%s/%s/degrade
etcdRawKey：etcd的装备名，只在datasource为etcd时有效，默许为app.registry.etcd

cbRules详细字段解析：

Enable: 表明是否开启此熔断规矩。
Resource: 熔断器规矩生效的埋点资源的称号；
Strategy: 熔断战略，现在支持SlowRequestRatio、ErrorRatio、ErrorCount三种；
1. 挑选以慢调用份额 (SlowRequestRatio) 作为阈值，需求设置答应的最大呼应时刻（MaxAllowedRtMs），恳求的呼应时刻大于该值则计算为慢调用。经过 Threshold 字段设置触发熔断的慢调用份额，取值规模为 [0.0, 1.0]。规矩装备后，在单位计算时长内恳求数目大于设置的最小恳求数目，而且慢调用的份额大于阈值，则接下来的熔断时长内恳求会主动被熔断。经过熔断时长后熔断器会进入勘探康复状况，若接下来的一个恳求呼应时刻小于设置的最大 RT 则完毕熔断，若大于设置的最大 RT 则会再次被熔断。
2. 挑选以过错份额 (ErrorRatio) 作为阈值，需求设置触发熔断的反常份额（Threshold），取值规模为 [0.0, 1.0]。规矩装备后，在单位计算时长内恳求数目大于设置的最小恳求数目，而且反常的份额大于阈值，则接下来的熔断时长内恳求会主动被熔断。经过熔断时长后熔断器会进入勘探康复状况，若接下来的一个恳求没有过错则完毕熔断，否则会再次被熔断。代码中能够经过 api.TraceError(entry, err) 函数来记载 error。
3. 挑选以过错份额 (ErrorCount) 作为阈值，需求设置触发熔断的反常份额（Threshold），取值规模为 [1, $]。规矩装备后，在单位计算时长内恳求数目大于设置的最小恳求数目，而且反常的次数大于阈值，则接下来的熔断时长内恳求会主动被熔断。经过熔断时长后熔断器会进入勘探康复状况，若接下来的一个恳求没有过错则完毕熔断，否则会再次被熔断。代码中能够经过 api.TraceError(entry, err) 函数来记载 error。
RetryTimeoutMs: 即熔断触发后持续的时刻（单位为 ms）。资源进入熔断状况后，在装备的熔断时长内，恳求都会快速失利。熔断完毕后进入勘探康复模式（HALF-OPEN）。
MinRequestAmount: 静默数量，假如当时计算周期内对资源的访问数量小于静默数量，那么熔断器就处于静默期。换言之，也便是触发熔断的最小恳求数目，若当时计算周期内的恳求数小于此值，即使达到熔断条件规矩也不会触发。
StatIntervalMs: 计算的时刻窗口长度（单位为 ms）。
MaxAllowedRtMs: 仅对慢调用熔断战略生效，MaxAllowedRtMs 是判别恳求是否是慢调用的临界值，也便是假如恳求的response time小于或等于MaxAllowedRtMs，那么就不是慢调用；假如response time大于MaxAllowedRtMs，那么当时恳求就归于慢调用。
Threshold: 关于慢调用熔断战略, Threshold表明是慢调用份额的阈值(小数表明，比方0.1表明10%)，也便是假如当时资源的慢调用份额假如高于Threshold，那么熔断器就会断开；否则保持闭合状况。关于过错份额战略，Threshold表明的是过错份额的阈值(小数表明，比方0.1表明10%)。关于过错数战略，Threshold是过错计数的阈值。

go-sentinel流量控制(五)：github优秀开源项目源码分析

go-admin 怎么运用go-sentinel

界说中间件

运用中间件

jupiter 怎么运用go-sentinel

界说装备

SentinelConfig

装备项

相关文章

Xmake 和 C/C++ 包管理

GO：如何编写一个 Worker Pool

[ANR] 为什么SharedPreference会引发ANR

深入浅出区块链 Day 12

作者信息