Golang 任务队列策略 -- 读《JOB QUEUES IN GO》

go process(job)

func worker(jobChan <-chan Job) {
 for job := range jobChan {
 process(job)
 }
}
// make a channel with a capacity of 100.
jobChan := make(chan Job, 100)
// start the worker
go worker(jobChan)
// enqueue a job
jobChan <- job

// make a channel with a capacity of 100.
jobChan := make(chan Job, 100)

// enqueue a job
jobChan <- job

// TryEnqueue tries to enqueue a job to the given job channel. Returns true if
// the operation was successful, and false if enqueuing would not have been
// possible without blocking. Job is not enqueued in the latter case.
func TryEnqueue(job Job, jobChan <-chan Job) bool {
 select {
 case jobChan <- job:
 return true
 default:
 return false
 }
}

if !TryEnqueue(job, chan) {
 http.Error(w, "max capacity reached", 503)
 return
}

close(jobChan)

for job := range jobChan {...}

package main
import (
 "fmt"
)
var jobChan chan int
func worker(jobChan <- chan int) {
 for job := range jobChan{
 fmt.Printf("执行任务 %d \n", job)
 }
}
func main() {
 jobChan = make(chan int, 100)
 //入队
 for i := 1; i <= 10; i++{
 jobChan <- i
 }
 
 close(jobChan)
 go worker(jobChan)
}

package main
import (
 "fmt"
 "sync"
)
var jobChan chan int
var wg sync.WaitGroup
func worker(jobChan <- chan int) {
 defer wg.Done()
 for job := range jobChan{
 fmt.Printf("执行任务 %d \n", job)
 }
}
func main() {
 jobChan = make(chan int, 100)
 //入队
 for i := 1; i <= 10; i++{
 jobChan <- i
 }
 wg.Add(1)
 close(jobChan)
 go worker(jobChan)
 wg.Wait()
}

执行任务 1 
执行任务 2 
执行任务 3 
执行任务 4 
执行任务 5 
执行任务 6 
执行任务 7 
执行任务 8 
执行任务 9 
执行任务 10 
Process finished with exit code 0

package main
import (
 "fmt"
 "sync"
 "time"
)
var jobChan chan int
var wg sync.WaitGroup
func worker(jobChan <-chan int) {
 defer wg.Done()
 for job := range jobChan {
 fmt.Printf("执行任务 %d \n", job)
 time.Sleep(1 * time.Second)
 }
}
func main() {
 jobChan = make(chan int, 100)
 //入队
 for i := 1; i <= 10; i++ {
 jobChan <- i
 }
 wg.Add(1)
 close(jobChan)
 go worker(jobChan)
 res := WaitTimeout(&wg, 5*time.Second)
 if res {
 fmt.Println("执行完成退出")
 } else {
 fmt.Println("执行超时退出")
 }
}
//超时机制
func WaitTimeout(wg *sync.WaitGroup, timeout time.Duration) bool {
 ch := make(chan struct{})
 go func() {
 wg.Wait()
 close(ch)
 }()
 select {
 case <-ch:
 return true
 case <-time.After(timeout):
 return false
 }
}

执行任务 1 
执行任务 2 
执行任务 3 
执行任务 4 
执行任务 5 
执行超时退出
Process finished with exit code 0

package main
import (
 "context"
 "fmt"
 "sync"
 "time"
)
var jobChan chan int
var ctx context.Context
var cancel context.CancelFunc
func worker(jobChan <-chan int, ctx context.Context) {
 for {
 select {
 case <-ctx.Done():
 return
 case job := <-jobChan:
 fmt.Printf("执行任务 %d \n", job)
 time.Sleep(1 * time.Second)
 }
 }
}
func main() {
 jobChan = make(chan int, 100)
 //带有取消功能的 contex
 ctx, cancel = context.WithCancel(context.Background())
 //入队
 for i := 1; i <= 10; i++ {
 jobChan <- i
 }
 close(jobChan)
 go worker(jobChan, ctx)
 time.Sleep(2 * time.Second)
 //調用cancel
 cancel()
}

执行任务 1 
执行任务 2 
Process finished with exit code 0

package main
import (
 "fmt"
 "time"
)
var jobChan chan int
func worker(jobChan <-chan int, cancelChan <-chan struct{}) {
 for {
 select {
 case <-cancelChan:
 return
 case job := <-jobChan:
 fmt.Printf("执行任务 %d \n", job)
 time.Sleep(1 * time.Second)
 }
 }
}
func main() {
 jobChan = make(chan int, 100)
 //通过chan 取消操作
 cancelChan := make(chan struct{})
 //入队
 for i := 1; i <= 10; i++ {
 jobChan <- i
 }
 close(jobChan)
 go worker(jobChan, cancelChan)
 time.Sleep(2 * time.Second)
 //关闭chan
 close(cancelChan)
}

有疑问加站长微信联系(非本文作者)

• 请尽量让自己的回复能够对别人有帮助
• 支持 Markdown 格式, **粗体**、~~删除线~~、`单行代码`
• 支持 @ 本站用户;支持表情(输入 : 提示),见 Emoji cheat sheet
• 图片支持拖拽、截图粘贴等方式上传