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

type RateLimiter interface {
	// When gets an item and gets to decide how long that item should wait
    //當一個物件放入的時候，需要回傳延遲多久（可自定義規則，等等會看到）
	When(item interface{}) time.Duration
    
    
	// Forget indicates that an item is finished being retried.  Doesn't matter whether its for perm failing	    
    // or for success, we'll stop tracking it
    //當一個物件完成的時候可以，要忘記曾經延遲過（重新計算）
    Forget(item interface{})
    
    
	// NumRequeues returns back how many failures the item has had
    // 回傳物件已經放入幾次（重試了幾次，白話一點呼叫NumRequeues幾次）
	NumRequeues(item interface{}) int
}

1
2
3
4
5
6

// MaxOfRateLimiter calls every RateLimiter and returns the worst case response
// When used with a token bucket limiter, the burst could be apparently exceeded in cases where particular items
// were separately delayed a longer time.
type MaxOfRateLimiter struct {
	limiters []RateLimiter     //定義一個實作集合RateLimiter比如前幾個章節有提到的 ItemExponentialFailureRateLimiter、ItemFastSlowRateLimiter都可以放到這個集合內
}

1
2
3
4

//傳入實作集合RateLimiter比如前幾個章節有提到的 ItemExponentialFailureRateLimiter、ItemFastSlowRateLimiter都可以放到這個集合內
func NewMaxOfRateLimiter(limiters ...RateLimiter) RateLimiter {
	return &MaxOfRateLimiter{limiters: limiters}    
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

func (r *MaxOfRateLimiter) When(item interface{}) time.Duration {
	ret := time.Duration(0)
	for _, limiter := range r.limiters {
		curr := limiter.When(item)    //把集合內所有的when都跑跑看，看看延遲時間哪個比較大，用大的延遲時間為主。
		if curr > ret {
			ret = curr
		}
	}

	return ret
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

func (r *MaxOfRateLimiter) NumRequeues(item interface{}) int {
	ret := 0
	for _, limiter := range r.limiters {
		curr := limiter.NumRequeues(item)    //把集合內所有的NumRequeues都跑跑看，看看哪個用了最多次，以最多次那個為主。 
		if curr > ret {
			ret = curr
		}
	}

	return ret
}

1
2
3
4
5

func (r *MaxOfRateLimiter) Forget(item interface{}) {
	for _, limiter := range r.limiters {    //做完了，讓集合內所有的人重新處理這個物件。
		limiter.Forget(item)
	}
}

1
2
3
4
5
6

func NewRateLimitingQueue(rateLimiter RateLimiter) RateLimitingInterface {
	return &rateLimitingType{
		DelayingInterface: NewDelayingQueue(),        //前一小節有提到過delating work queue的newfunction
		rateLimiter:       rateLimiter,               //自行實作的rateLimiter
	}
}

1
2
3
4

NewRateLimitingQueue(
    NNewItemFastSlowRateLimiter(5*time.Millisecond, 3*time.Second, 3),
    NewItemExponentialFailureRateLimiter(1*time.Millisecond, 1*time.Second)
)