分布式系统高可用实战之限流器（Go 版本实现）

// 采用channel作为底层数据结构，从而达到阻塞的获取和使用信号量
type Semaphore struct {
    innerChan chan struct{}
}
// 初始化信号量，本质初始化一个channel，channel的初始化大小为 信号量数值
func NewSemaphore(num uint64) *Semaphore {
    return &Semaphore{
        innerChan: make(chan struct{}, num),
    }
}
// 获取信号量，本质是 向channel放入元素，如果同时有很多协程并发获取信号量，则channel则会full阻塞，从而达到控制并发协程数的目的，也即是信号量的控制
func (s *Semaphore) Acquire() {
    for {
        select {
        case s.innerChan <- struct{}{}:
            return
        default:
            log.Error("semaphore acquire is blocking")
            time.Sleep(100 * time.Millisecond)
        }
    }
}
// 释放信号量 本质是 从channel中获取元素，由于有acquire的放入元素，所以此处一定能回去到元素 也就能释放成功，default只要是出于安全编程的目的
func (s *Semaphore) Release() {
    select {
    case <-s.innerChan:
        return
    default:
        return
    }
}

// 此处截取自研的熔断器代码中的限流实现，这是非阻塞的实现
func (sp *servicePanel) incLimit() error {
    // 如果大于限制的条件则返回错误
    if sp.currentLimitCount.Load() > sp.currLimitFunc(nil) {
        return ErrCurrentLimit
    }
    sp.currentLimitCount.Inc()
    return nil
}

func (sp *servicePanel) clearLimit() {
    // 定期每秒重置计数器，从而达到每秒限制的并发数
    // 比如限制1000req/s，在这里指每秒清理1000的计数值
// 令牌桶是定期放，这里是逆思维，每秒清空，实现不仅占用内存低而且效率高
    t := time.NewTicker(time.Second)
    for {
        select {
        case <-t.C:
            sp.currentLimitCount.Store(0)
        }
    }
}

func (sp *servicePanel) incLimit() error {
    // 如果大于1则返回错误
    if sp.currentLimitCount.Load() > 1 {
        return ErrCurrentLimit
    }
    sp.currentLimitCount.Inc()
    return nil
}

func (sp *servicePanel) clearLimit() {
    // 1s除以每秒限流个数
    t := time.NewTicker(time.Second/time.Duration(sp.currLimitFunc(nil)))
    for {
        select {
        case <-t.C:
            sp.currentLimitCount.Store(0)
        }
    }
}

#第一版本
go get github.com/uber-go/ratelimit@v0.1.0
#改进版本
go get github.com/uber-go/ratelimit@master

func ExampleRatelimit() {
    rl := ratelimit.New(100) // per second

    prev := time.Now()
    for i := 0; i < 10; i++ {
        now := rl.Take()
        if i > 0 {
            fmt.Println(i, now.Sub(prev))
        }
        prev = now
    }
}

    // Output:
    // 1 10ms
    // 2 10ms
    // 3 10ms
    // 4 10ms
    // 5 10ms
    // 6 10ms
    // 7 10ms
    // 8 10ms
    // 9 10ms

// New returns a Limiter that will limit to the given RPS.
func New(rate int, opts ...Option) Limiter {
    l := &limiter{
        perRequest: time.Second / time.Duration(rate),
        maxSlack:   -10 * time.Second / time.Duration(rate),
    }
    for _, opt := range opts {
        opt(l)
    }
    if l.clock == nil {
        l.clock = clock.New()
    }
    return l
}

// Take blocks to ensure that the time spent between multiple
// Take calls is on average time.Second/rate.
func (t *limiter) Take() time.Time {
    t.Lock()
    defer t.Unlock()

    now := t.clock.Now()

    // If this is our first request, then we allow it.
    if t.last.IsZero() {
        t.last = now
        return t.last
    }

    // sleepFor calculates how much time we should sleep based on
    // the perRequest budget and how long the last request took.
    // Since the request may take longer than the budget, this number
    // can get negative, and is summed across requests.
    t.sleepFor += t.perRequest - now.Sub(t.last)

    // We shouldn't allow sleepFor to get too negative, since it would mean that
    // a service that slowed down a lot for a short period of time would get
    // a much higher RPS following that.
    if t.sleepFor < t.maxSlack {
        t.sleepFor = t.maxSlack
    }

    // If sleepFor is positive, then we should sleep now.
    if t.sleepFor > 0 {
        t.clock.Sleep(t.sleepFor)
        t.last = now.Add(t.sleepFor)
        t.sleepFor = 0
    } else {
        t.last = now
    }

    return t.last
}

// Take blocks to ensure that the time spent between multiple
// Take calls is on average time.Second/rate.
func (t *limiter) Take() time.Time {
    newState := state{}
    taken := false
    for !taken {
        now := t.clock.Now()

        previousStatePointer := atomic.LoadPointer(&t.state)
        oldState := (*state)(previousStatePointer)

        newState = state{}
        newState.last = now

        // If this is our first request, then we allow it.
        if oldState.last.IsZero() {
            taken = atomic.CompareAndSwapPointer(&t.state, previousStatePointer, unsafe.Pointer(&newState))
            continue
        }

        // sleepFor calculates how much time we should sleep based on
        // the perRequest budget and how long the last request took.
        // Since the request may take longer than the budget, this number
        // can get negative, and is summed across requests.
        newState.sleepFor += t.perRequest - now.Sub(oldState.last)
        // We shouldn't allow sleepFor to get too negative, since it would mean that
        // a service that slowed down a lot for a short period of time would get
        // a much higher RPS following that.
        if newState.sleepFor < t.maxSlack {
            newState.sleepFor = t.maxSlack
        }
        if newState.sleepFor > 0 {
            newState.last = newState.last.Add(newState.sleepFor)
        }
        taken = atomic.CompareAndSwapPointer(&t.state, previousStatePointer, unsafe.Pointer(&newState))
    }
    t.clock.Sleep(newState.sleepFor)
    return newState.last
}