您好,登錄后才能下訂單哦!
本篇內容介紹了“golang中tunny的原理和作用是什么”的有關知識,在實際案例的操作過程中,不少人都會遇到這樣的困境,接下來就讓小編帶領大家學習一下如何處理這些情況吧!希望大家仔細閱讀,能夠學有所成!
本文主要研究一下tunny
type Worker interface { // Process will synchronously perform a job and return the result. Process(interface{}) interface{} // BlockUntilReady is called before each job is processed and must block the // calling goroutine until the Worker is ready to process the next job. BlockUntilReady() // Interrupt is called when a job is cancelled. The worker is responsible // for unblocking the Process implementation. Interrupt() // Terminate is called when a Worker is removed from the processing pool // and is responsible for cleaning up any held resources. Terminate() }
Worker接口定義了Process、BlockUntilReady、Interrupt、Terminate方法
type closureWorker struct { processor func(interface{}) interface{} } func (w *closureWorker) Process(payload interface{}) interface{} { return w.processor(payload) } func (w *closureWorker) BlockUntilReady() {} func (w *closureWorker) Interrupt() {} func (w *closureWorker) Terminate() {}
closureWorker定義了processor屬性,它實現了Worker接口的Process、BlockUntilReady、Interrupt、Terminate方法,其中Process方法委托給processor
type callbackWorker struct{} func (w *callbackWorker) Process(payload interface{}) interface{} { f, ok := payload.(func()) if !ok { return ErrJobNotFunc } f() return nil } func (w *callbackWorker) BlockUntilReady() {} func (w *callbackWorker) Interrupt() {} func (w *callbackWorker) Terminate() {}
callbackWorker定義了processor屬性,它實現了Worker接口的Process、BlockUntilReady、Interrupt、Terminate方法,其中Process方法執行的是payload函數
type Pool struct { queuedJobs int64 ctor func() Worker workers []*workerWrapper reqChan chan workRequest workerMut sync.Mutex } func New(n int, ctor func() Worker) *Pool { p := &Pool{ ctor: ctor, reqChan: make(chan workRequest), } p.SetSize(n) return p } func NewFunc(n int, f func(interface{}) interface{}) *Pool { return New(n, func() Worker { return &closureWorker{ processor: f, } }) } func NewCallback(n int) *Pool { return New(n, func() Worker { return &callbackWorker{} }) }
Pool定義了queuedJobs、ctor、workers、reqChan、workerMut屬性;New方法根據n和ctor創建Pool;NewFunc方法根據n和f來創建closureWorker;NewCallback方法創建callbackWorker
func (p *Pool) Process(payload interface{}) interface{} { atomic.AddInt64(&p.queuedJobs, 1) request, open := <-p.reqChan if !open { panic(ErrPoolNotRunning) } request.jobChan <- payload payload, open = <-request.retChan if !open { panic(ErrWorkerClosed) } atomic.AddInt64(&p.queuedJobs, -1) return payload }
Process方法首先遞增queuedJobs,然后從reqChan讀取request,然后往jobChan寫入payload,之后再等待retChan,最后遞減queuedJobs
func (p *Pool) SetSize(n int) { p.workerMut.Lock() defer p.workerMut.Unlock() lWorkers := len(p.workers) if lWorkers == n { return } // Add extra workers if N > len(workers) for i := lWorkers; i < n; i++ { p.workers = append(p.workers, newWorkerWrapper(p.reqChan, p.ctor())) } // Asynchronously stop all workers > N for i := n; i < lWorkers; i++ { p.workers[i].stop() } // Synchronously wait for all workers > N to stop for i := n; i < lWorkers; i++ { p.workers[i].join() } // Remove stopped workers from slice p.workers = p.workers[:n] }
SetSize方法首先通過workerMut加鎖,然后根據lWorkers創建newWorkerWrapper,之后執行worker.stop,再執行worker.join(),然后清空workers
func (p *Pool) Close() { p.SetSize(0) close(p.reqChan) }
Close方法執行SetSize(0)及close(p.reqChan)
func TestFuncJob(t *testing.T) { pool := NewFunc(10, func(in interface{}) interface{} { intVal := in.(int) return intVal * 2 }) defer pool.Close() for i := 0; i < 10; i++ { ret := pool.Process(10) if exp, act := 20, ret.(int); exp != act { t.Errorf("Wrong result: %v != %v", act, exp) } } }
TestFuncJob通過NewFunc創建pool,
tunny的Worker接口定義了Process、BlockUntilReady、Interrupt、Terminate方法;NewFunc方法創建的是closureWorker,NewCallback方法創建的是callbackWorker。
“golang中tunny的原理和作用是什么”的內容就介紹到這里了,感謝大家的閱讀。如果想了解更多行業相關的知識可以關注億速云網站,小編將為大家輸出更多高質量的實用文章!
免責聲明:本站發布的內容(圖片、視頻和文字)以原創、轉載和分享為主,文章觀點不代表本網站立場,如果涉及侵權請聯系站長郵箱:is@yisu.com進行舉報,并提供相關證據,一經查實,將立刻刪除涉嫌侵權內容。