Golang高并发学习【通过Go来处理每分钟达百万的数据请求】
golang 的主要场景之一就是用来做高并发的处理，之前的公司有个 golang 写的服务用来生成二维码，并上传到 upyun 上。
并发很好，设计良好，当时还看了下源码，然后今天继续看了一下相关的文章。
然后找到这篇译文：
【译文】通过 Go 来处理每分钟达百万的数据请求 https://blog.csdn.net/tybaoerge/article/details/50392386
这篇文章非常好的描述了什么是高并发，怎么实现高并发：在有限的资源下，通过合理的使用 goroutine，channel，来完成业务操作。
只要是 2 部分，调度器（dispatcher），执行者（worker）
源码分析
调度器 dispatcher
调度器的作用是从 JobQueue 中获取新的 job，并将 job 交给闲置的 worker。
package worker
import ( "fmt" )
type Dispatcher struct { // A pool of workers channels that are registered with the dispatcher // 执行池， WorkerPool chan chan Job maxWorkers int }
func NewDispatcher(maxWorkers int) Dispatcher { pool := make(chan chan Job, maxWorkers) return &Dispatcher{WorkerPool: pool, maxWorkers: maxWorkers} } // 初始化worker池，并启动woker池，并开始接受新的job // func (d Dispatcher) Run() { // starting n number of workers for i := 0; i < d.maxWorkers; i++ { worker := NewWorker(d.WorkerPool) worker.Start() }
go d.dispatch() }
// 开始调度，接收新的job func (d *Dispatcher) dispatch() { for { select { case job := <-JobQueue: // a job request has been received go func(job Job) { fmt.Println("[UPYUN] Dispatcher get JOB") // try to obtain a worker job channel that is available. // this will block until a worker is idle // 从pool中获取空闲的job channel jobChannel := <-d.WorkerPool
// dispatch the job to the worker job channel // 将job塞入 job channel中 jobChannel <- job }(job) } } }
工人执行者 Worker
package worker
import ( "fmt" "os" )
var ( MaxWorker = os.Getenv("MAX_WORKERS") MaxQueue = os.Getenv("MAX_QUEUE") )
type Payload struct { FilePath string }
// Job represents the job to be run type Job struct { Payload Payload }
// A buffered channel that we can send work requests on. var JobQueue chan Job
// Worker represents the worker that executes the job type Worker struct { WorkerPool chan (chan Job) JobChannel chan Job quit chan bool }
func NewWorker(workerPool chan chan Job) Worker { return Worker{ WorkerPool: workerPool, JobChannel: make(chan Job), // 分配器中，会将任务交给jobChannel，下面会从这里读取到job quit: make(chan bool)} }
// Start method starts the run loop for the worker, listening for a quit channel in // case we need to stop it func (w Worker) Start() { go func() { for { // register the current worker into the worker queue. // 将限制的JobChannel（chan） 丢入WorkerPool w.WorkerPool <- w.JobChannel
select { case job := <-w.JobChannel: // 当闲置的jobChannel中有job时， job.Payload // job开始工作 case <-w.quit: return } } }() }
// Stop signals the worker to stop listening for work requests. func (w Worker) Stop() { go func() { w.quit <- true }() }
感觉整个代码最抽象的其实是
chan chan Job
大家知道它是“管道中的管道” 的意思。
用图来模拟就是这样一种感觉
WorkerPool chan
| chan Job |
| chan Job |
- + - + - + -
但是实际用起来，还是蛮抽象的。
我想着，这样写似乎跟下面的代码没有太大区别啊：
伪代码：
maxPool = 100 workerPool = make(chan Job, maxPool)
// 塞入队列 for { workerPool <- Job }
// 取出队列 for i=0; i<maxPool; i++ { go func() { for { select { case job := <- workerPool: job.Do() } } } }