community_team/app/mw/mw_limiter.go

package mw

import (
	"applet/app/e"
	"applet/app/md"
	"applet/app/svc"
	"applet/app/utils"
	"applet/app/utils/cache"
	"bytes"
	"fmt"
	"github.com/gin-gonic/gin"
	"io/ioutil"
)

func Limiter(c *gin.Context) {
	limit := 200 // 限流次数
	ttl := 2     // 限流过期时间
	ip := c.ClientIP()
	// 读取token或者ip
	token := c.GetHeader("Authorization")
	mid := c.GetString("mid")
	// 判断是否已经超出限额次数
	method := c.Request.Method
	host := c.Request.Host
	uri := c.Request.URL.String()

	buf := make([]byte, 5120*10)
	num, _ := c.Request.Body.Read(buf)
	body := buf[:num]
	// Write body back
	c.Request.Body = ioutil.NopCloser(bytes.NewBuffer(body))
	//queryValue := utils.SerializeStr(c.Request.URL.Query()) //不建议开启，失去限流的意义

	//TODO::分布式锁阻拦(保证原子性)
	requestIdPrefix := fmt.Sprintf(md.DealAppLimiterRequestIdPrefix, mid, ip)
	cb, err := svc.HandleLimiterDistributedLock(mid, ip, requestIdPrefix)
	if err != nil {
		e.OutErr(c, e.ERR, err.Error())
		return
	}
	if cb != nil {
		defer cb() // 释放锁
	}

	Md5 := utils.Md5(ip + token + method + host + uri + string(body))
	//Md5 := utils.Md5(ip + token + method + host + uri + string(body) + queryValue)
	if cache.Exists(Md5) {
		c.AbortWithStatusJSON(428, gin.H{
			"code": 428,
			"msg":  "don't repeat the request",
			"data": struct{}{},
		})
		return
	}

	// 2s后没返回自动释放
	go cache.SetEx(Md5, "0", ttl)

	key := "NEW_LIMITER_APP_COMM_" + ip
	reqs, _ := cache.GetInt(key)
	if reqs >= limit {
		c.AbortWithStatusJSON(429, gin.H{
			"code": 429,
			"msg":  "too many requests",
			"data": struct{}{},
		})
		return
	}
	if reqs > 0 {
		//go cache.Incr(key)
		go cache.SetEx(key, reqs+1, ttl)
	} else {
		go cache.SetEx(key, 1, ttl)
	}
	c.Next()
	go cache.Del(Md5)
}