diff --git a/README.md b/README.md index 91fdd91..fef041e 100644 --- a/README.md +++ b/README.md @@ -1,17 +1,17 @@

- -A goroutine pool in Go + +A goroutine pool for Go

- - - - + + + +
- - - - - + + + + +

English | [🇨🇳中文](README_ZH.md) @@ -136,61 +136,6 @@ func main() { } ``` -### Integrate with http server -```go -package main - -import ( - "io/ioutil" - "net/http" - - "github.com/panjf2000/ants/v2" -) - -type Request struct { - Param []byte - Result chan []byte -} - -func main() { - pool, _ := ants.NewPoolWithFunc(100000, func(payload interface{}) { - request, ok := payload.(*Request) - if !ok { - return - } - reverseParam := func(s []byte) []byte { - for i, j := 0, len(s)-1; i < j; i, j = i+1, j-1 { - s[i], s[j] = s[j], s[i] - } - return s - }(request.Param) - - request.Result <- reverseParam - }) - defer pool.Release() - - http.HandleFunc("/reverse", func(w http.ResponseWriter, r *http.Request) { - param, err := ioutil.ReadAll(r.Body) - if err != nil { - http.Error(w, "request error", http.StatusInternalServerError) - } - defer r.Body.Close() - - request := &Request{Param: param, Result: make(chan []byte)} - - // Throttle the requests traffic with ants pool. This process is asynchronous and - // you can receive a result from the channel defined outside. - if err := pool.Invoke(request); err != nil { - http.Error(w, "throttle limit error", http.StatusInternalServerError) - } - - w.Write(<-request.Result) - }) - - http.ListenAndServe(":8080", nil) -} -``` - ### Functional options for ants pool ```go @@ -199,7 +144,9 @@ type Option func(opts *Options) // Options contains all options which will be applied when instantiating a ants pool. type Options struct { - // ExpiryDuration sets the expired time (second) of every worker. + // ExpiryDuration is a period for the scavenger goroutine to clean up those expired workers, + // the scavenger scans all workers every `ExpiryDuration` and clean up those workers that haven't been + // used for more than `ExpiryDuration`. ExpiryDuration time.Duration // PreAlloc indicates whether to make memory pre-allocation when initializing Pool. @@ -217,6 +164,10 @@ type Options struct { // PanicHandler is used to handle panics from each worker goroutine. // if nil, panics will be thrown out again from worker goroutines. PanicHandler func(interface{}) + + // Logger is the customized logger for logging info, if it is not set, + // default standard logger from log package is used. + Logger Logger } // WithOptions accepts the whole options config. @@ -260,6 +211,13 @@ func WithPanicHandler(panicHandler func(interface{})) Option { opts.PanicHandler = panicHandler } } + +// WithLogger sets up a customized logger. +func WithLogger(logger Logger) Option { + return func(opts *Options) { + opts.Logger = logger + } +} ``` `ants.Options`contains all optional configurations of ants pool, which allows you to customize the goroutine pool by invoking option functions to set up each configuration in `NewPool`/`NewPoolWithFunc`method. diff --git a/README_ZH.md b/README_ZH.md index 6d5220a..4cf5f43 100644 --- a/README_ZH.md +++ b/README_ZH.md @@ -1,17 +1,17 @@

- + Go 语言的 goroutine 池

- - - - + + + +
- - - - - + + + + +

[英文](README.md) | 🇨🇳中文 @@ -136,61 +136,6 @@ func main() { } ``` -### 与 http server 集成 -```go -package main - -import ( - "io/ioutil" - "net/http" - - "github.com/panjf2000/ants/v2" -) - -type Request struct { - Param []byte - Result chan []byte -} - -func main() { - pool, _ := ants.NewPoolWithFunc(100000, func(payload interface{}) { - request, ok := payload.(*Request) - if !ok { - return - } - reverseParam := func(s []byte) []byte { - for i, j := 0, len(s)-1; i < j; i, j = i+1, j-1 { - s[i], s[j] = s[j], s[i] - } - return s - }(request.Param) - - request.Result <- reverseParam - }) - defer pool.Release() - - http.HandleFunc("/reverse", func(w http.ResponseWriter, r *http.Request) { - param, err := ioutil.ReadAll(r.Body) - if err != nil { - http.Error(w, "request error", http.StatusInternalServerError) - } - defer r.Body.Close() - - request := &Request{Param: param, Result: make(chan []byte)} - - // Throttle the requests traffic with ants pool. This process is asynchronous and - // you can receive a result from the channel defined outside. - if err := pool.Invoke(request); err != nil { - http.Error(w, "throttle limit error", http.StatusInternalServerError) - } - - w.Write(<-request.Result) - }) - - http.ListenAndServe(":8080", nil) -} -``` - ### Pool 配置 ```go @@ -199,7 +144,9 @@ type Option func(opts *Options) // Options contains all options which will be applied when instantiating a ants pool. type Options struct { - // ExpiryDuration sets the expired time (second) of every worker. + // ExpiryDuration is a period for the scavenger goroutine to clean up those expired workers, + // the scavenger scans all workers every `ExpiryDuration` and clean up those workers that haven't been + // used for more than `ExpiryDuration`. ExpiryDuration time.Duration // PreAlloc indicates whether to make memory pre-allocation when initializing Pool. @@ -217,6 +164,10 @@ type Options struct { // PanicHandler is used to handle panics from each worker goroutine. // if nil, panics will be thrown out again from worker goroutines. PanicHandler func(interface{}) + + // Logger is the customized logger for logging info, if it is not set, + // default standard logger from log package is used. + Logger Logger } // WithOptions accepts the whole options config. @@ -260,6 +211,13 @@ func WithPanicHandler(panicHandler func(interface{})) Option { opts.PanicHandler = panicHandler } } + +// WithLogger sets up a customized logger. +func WithLogger(logger Logger) Option { + return func(opts *Options) { + opts.Logger = logger + } +} ``` 通过在调用`NewPool`/`NewPoolWithFunc`之时使用各种 optional function，可以设置`ants.Options`中各个配置项的值，然后用它来定制化 goroutine pool. diff --git a/options.go b/options.go index 87b3aa6..5235f96 100644 --- a/options.go +++ b/options.go @@ -15,7 +15,9 @@ func loadOptions(options ...Option) *Options { // Options contains all options which will be applied when instantiating a ants pool. type Options struct { - // ExpiryDuration sets the expired time of every worker. + // ExpiryDuration is a period for the scavenger goroutine to clean up those expired workers, + // the scavenger scans all workers every `ExpiryDuration` and clean up those workers that haven't been + // used for more than `ExpiryDuration`. ExpiryDuration time.Duration // PreAlloc indicates whether to make memory pre-allocation when initializing Pool. diff --git a/pool.go b/pool.go index 1f72f77..8b3946e 100644 --- a/pool.go +++ b/pool.go @@ -61,8 +61,8 @@ type Pool struct { options *Options } -// periodicallyPurge clears expired workers periodically. -func (p *Pool) periodicallyPurge() { +// purgePeriodically clears expired workers periodically which runs in an individual goroutine, as a scavenger. +func (p *Pool) purgePeriodically() { heartbeat := time.NewTicker(p.options.ExpiryDuration) defer heartbeat.Stop() @@ -133,7 +133,7 @@ func NewPool(size int, options ...Option) (*Pool, error) { p.cond = sync.NewCond(p.lock) // Start a goroutine to clean up expired workers periodically. - go p.periodicallyPurge() + go p.purgePeriodically() return p, nil } @@ -187,7 +187,7 @@ func (p *Pool) Release() { // Reboot reboots a released pool. func (p *Pool) Reboot() { if atomic.CompareAndSwapInt32(&p.state, CLOSED, OPENED) { - go p.periodicallyPurge() + go p.purgePeriodically() } } diff --git a/pool_func.go b/pool_func.go index 099040d..7c16a98 100644 --- a/pool_func.go +++ b/pool_func.go @@ -63,8 +63,8 @@ type PoolWithFunc struct { options *Options } -// periodicallyPurge clears expired workers periodically. -func (p *PoolWithFunc) periodicallyPurge() { +// purgePeriodically clears expired workers periodically which runs in an individual goroutine, as a scavenger. +func (p *PoolWithFunc) purgePeriodically() { heartbeat := time.NewTicker(p.options.ExpiryDuration) defer heartbeat.Stop() @@ -148,7 +148,7 @@ func NewPoolWithFunc(size int, pf func(interface{}), options ...Option) (*PoolWi p.cond = sync.NewCond(p.lock) // Start a goroutine to clean up expired workers periodically. - go p.periodicallyPurge() + go p.purgePeriodically() return p, nil } @@ -206,7 +206,7 @@ func (p *PoolWithFunc) Release() { // Reboot reboots a released pool. func (p *PoolWithFunc) Reboot() { if atomic.CompareAndSwapInt32(&p.state, CLOSED, OPENED) { - go p.periodicallyPurge() + go p.purgePeriodically() } }