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
7
8
9

// ItemExponentialFailureRateLimiter does a simple baseDelay*2^<num-failures> limit
// dealing with max failures and expiration are up to the caller
type ItemExponentialFailureRateLimiter struct {
	failuresLock sync.Mutex            //鎖，防止資源競爭
	failures     map[interface{}]int   //計算某個物件呼叫了延遲的次數

	baseDelay time.Duration            //基礎延遲的時間
	maxDelay  time.Duration            //最多要延遲多久
}

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

//不知道為什麼要設定一個空的變數
var _ RateLimiter = &ItemExponentialFailureRateLimiter{}

//傳入基礎要等待的時間，最大等待時間，以及初始化failures map資料集
func NewItemExponentialFailureRateLimiter(baseDelay time.Duration, maxDelay time.Duration) RateLimiter {
	return &ItemExponentialFailureRateLimiter{
		failures:  map[interface{}]int{},
		baseDelay: baseDelay,
		maxDelay:  maxDelay,
	}
}

//預設等待時間為 1 Millisecond，最大等待時間為1000秒
func DefaultItemBasedRateLimiter() RateLimiter {
	return NewItemExponentialFailureRateLimiter(time.Millisecond, 1000*time.Second)
}

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

func (r *ItemExponentialFailureRateLimiter) When(item interface{}) time.Duration {
	r.failuresLock.Lock()                    //鎖不做解釋
	defer r.failuresLock.Unlock()            //解鎖不做解釋

	exp := r.failures[item]                  //查看map裡面物件放入的次數
	r.failures[item] = r.failures[item] + 1    //放入次數＋1

	// The backoff is capped such that 'calculated' value never overflows.
        // backoff=base * 2^（物件放入的次數），代表延遲時間為指數型成長。
	backoff := float64(r.baseDelay.Nanoseconds()) * math.Pow(2, float64(exp))  
        //如果backoff 大於int64（溢位）就回傳最大延遲時間
	if backoff > math.MaxInt64 {
		return r.maxDelay
	}
        //封裝為延遲時間
	calculated := time.Duration(backoff)
        //延遲時間超過最大延遲時間就回傳最大延遲時間
	if calculated > r.maxDelay {
		return r.maxDelay
	}

	return calculated
}

1
2
3
4
5
6

func (r *ItemExponentialFailureRateLimiter) NumRequeues(item interface{}) int {
	r.failuresLock.Lock()                    //鎖不做解釋
	defer r.failuresLock.Unlock()            //解鎖不做解釋

	return r.failures[item]                  //回傳重試次數
}

1
2
3
4
5
6

func (r *ItemExponentialFailureRateLimiter) Forget(item interface{}) {
	r.failuresLock.Lock()                    //鎖不做解釋
	defer r.failuresLock.Unlock()            //解鎖不做解釋

	delete(r.failures, item)                 //刪除map裡的物件（重新計算）
}

1
2
3
4
5
6

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

1

NewRateLimitingQueue(NewItemExponentialFailureRateLimiter(time.Millisecond, 1000*time.Second)