扫二维码与项目经理沟通
我们在微信上24小时期待你的声音
解答本文疑问/技术咨询/运营咨询/技术建议/互联网交流
Golang中怎么实现百万级高并发,很多新手对此不是很清楚,为了帮助大家解决这个难题,下面小编将为大家详细讲解,有这方面需求的人可以来学习下,希望你能有所收获。
创新互联公司-专业网站定制、快速模板网站建设、高性价比咸阳网站开发、企业建站全套包干低至880元,成熟完善的模板库,直接使用。一站式咸阳网站制作公司更省心,省钱,快速模板网站建设找我们,业务覆盖咸阳地区。费用合理售后完善,10余年实体公司更值得信赖。
package main import ( "fmt" "runtime" "time" ) func main(){ //这里我们假设数据是int类型,缓存格式设为100 dataChan:=make(chan int,100) go func(){ for{ select{ case data:=<-dataChan: fmt.Println("data:",data) time.Sleep(1 * time.Second)//这里延迟是模拟处理数据的耗时 } } }() //填充数据 for i:=0;i<100;i++{ dataChan<-i } //这里循环打印查看协程个数 for { fmt.Println("runtime.NumGoroutine() :", runtime.NumGoroutine()) time.Sleep(2 * time.Second) } }
这里打印出来的协程个数时2,为什么? 因为main方法独占一个主协程,我们又起了一个协程,所以是两个。
首先我们要抽象出几个概念:
Job: type Job interface { Do() } // 一个数据接口,所有的数据都要实现该接口,才能被传递进来 //实现Job接口的一个数据实例,需要实现一个Do()方法,对数据的处理就在这个Do()方法中。 Job通道: 这里有两个Job通道: 1、WorkerPool的Job channel,用于调用者把具体的数据写入到这里,WorkerPool读取。 2、Worker的Job channel,当WorkerPool读取到Job,并拿到可用的Worker的时候,会将Job实例写入该Worker的Job channel,用来直接执行Do()方法。 Worker: type Worker struct { JobQueue chan Job //Worker的Job通道 } //每一个被初始化的worker都会在后期单独占用一个协程 //初始化的时候会先把自己的JobQueue传递到Worker通道中, //然后阻塞读取自己的JobQueue,读到一个Job就执行Job对象的Do()方法。 工作池(WorkerPool): type WorkerPool struct { workerlen int //WorkerPool中同时 存在Worker的个数 JobQueue chan Job // WorkerPool的Job通道 WorkerQueue chan chan Job } //初始化时会按照传入的num,启动num个后台协程,然后循环读取Job通道里面的数据, //读到一个数据时,再获取一个可用的Worker,并将Job对象传递到该Worker的chan通道
整个过程中 每个Worker都会被运行在一个协程中,在整个WorkerPool中就会有num可空闲的Worker,当来一条数据的时候,就会在工作池中去一个空闲的Worker去执行该Job,当工作池中没有可用的worker时,就会阻塞等待一个空闲的worker。
这是一个粗糙最简单的版本,只是为了演示效果,具体使用需要根据实际情况加一些特殊的处理。
当数据无限多的时候func (wp *WorkerPool) Run() 会无限创建协程,这里需要做一些处理,这里是为了让所有的请求不等待,并且体现一下最大峰值时的协程数。具体因项目而异。
代码地址:https://github.com/wangzhen0625/gonote/tree/master/7goroutune
main.go
package main import ( "fmt" "runtime" "time" ) type Score struct { Num int } func (s *Score) Do() { fmt.Println("num:", s.Num) time.Sleep(1 * 1 * time.Second) } func main() { num := 100 * 100 * 20 // debug.SetMaxThreads(num + 1000) //设置最大线程数 // 注册工作池,传入任务 // 参数1 worker并发个数 p := NewWorkerPool(num) p.Run() datanum := 100 * 100 * 100 * 100 go func() { for i := 1; i <= datanum; i++ { sc := &Score{Num: i} p.JobQueue <- sc } }() for { fmt.Println("runtime.NumGoroutine() :", runtime.NumGoroutine()) time.Sleep(2 * time.Second) } }
job.go
package main type Job interface { Do() }
worker.go
package main type Worker struct { JobQueue chan Job } func NewWorker() Worker { return Worker{JobQueue: make(chan Job)} } func (w Worker) Run(wq chan chan Job) { go func() { for { wq <- w.JobQueue select { case job := <-w.JobQueue: job.Do() } } }() }
workerpool.go
package main import "fmt" type WorkerPool struct { workerlen int JobQueue chan Job WorkerQueue chan chan Job } func NewWorkerPool(workerlen int) *WorkerPool { return &WorkerPool{ workerlen: workerlen, JobQueue: make(chan Job), WorkerQueue: make(chan chan Job, workerlen), } } func (wp *WorkerPool) Run() { fmt.Println("初始化worker") //初始化worker for i := 0; i < wp.workerlen; i++ { worker := NewWorker() worker.Run(wp.WorkerQueue) } // 循环获取可用的worker,往worker中写job go func() { for { select { case job := <-wp.JobQueue: worker := <-wp.WorkerQueue worker <- job } } }() }
看完上述内容是否对您有帮助呢?如果还想对相关知识有进一步的了解或阅读更多相关文章,请关注创新互联行业资讯频道,感谢您对创新互联的支持。
我们在微信上24小时期待你的声音
解答本文疑问/技术咨询/运营咨询/技术建议/互联网交流