关于golang:聊聊dapr的Limiter

序

本文次要钻研一下 dapr 的 Limiter

Limiter

dapr/pkg/concurrency/limiter.go

const (
    // DefaultLimit is the default concurrency limit
    DefaultLimit = 100
)

// Limiter object
type Limiter struct {
    limit         int
    tickets       chan int
    numInProgress int32
}

Limiter 定义了 limit、tickets、numInProgress 属性

NewLimiter

dapr/pkg/concurrency/limiter.go

// NewLimiter allocates a new ConcurrencyLimiter
func NewLimiter(limit int) *Limiter {
    if limit <= 0 {limit = DefaultLimit}

    // allocate a limiter instance
    c := &Limiter{
        limit:   limit,
        tickets: make(chan int, limit),
    }

    // allocate the tickets:
    for i := 0; i < c.limit; i++ {c.tickets <- i}

    return c
}

NewLimiter 办法依据 limit 来创立 Limiter，并挨个调配 ticket

Execute

dapr/pkg/concurrency/limiter.go

// Execute adds a function to the execution queue.
// if num of go routines allocated by this instance is < limit
// launch a new go routine to execute job
// else wait until a go routine becomes available
func (c *Limiter) Execute(job func(param interface{}), param interface{}) int {
    ticket := <-c.tickets
    atomic.AddInt32(&c.numInProgress, 1)
    go func(param interface{}) {defer func() {
            c.tickets <- ticket
            atomic.AddInt32(&c.numInProgress, -1)
        }()

        // run the job
        job(param)
    }(param)
    return ticket
}

Execute 办法首先获取 ticket，而后递增 numInProgress，之后异步执行 job，执行完后偿还 ticket

Wait

dapr/pkg/concurrency/limiter.go

// Wait will block all the previously Executed jobs completed running.
//
// IMPORTANT: calling the Wait function while keep calling Execute leads to
//            un-desired race conditions
func (c *Limiter) Wait() {
    for i := 0; i < c.limit; i++ {<-c.tickets}
}

Wait 办法遍历 limit，挨个期待 tickets 返回

小结

dapr 的 Limiter 定义了 limit、tickets、numInProgress 属性；它定义了 Execute、Wait 办法，同时提供 NewLimiter 的工厂办法。

doc

• dapr