workPool

README

workerPool

workerPool实现百万级的并发
高效的worker，每分钟可以达到万级以上的task异步处理
性能可以达到百亿级别以上的处理量，可以作为大规模作业任务处理。
使用场景：一般用在大量的http/grpc服务或大规模的异步作业层,job都可以。

Documentation

Overview

workerPool实现百万级的并发 go程序开发过程中,通过简单的调用go func 函数来开启协程,容易导致程序死锁 并且会无限制的开启groutine,groutine数量激增的情况下并发性能会明显下降 所以需要考虑使用工作池来控制协程数量,以达到高并发的效果.

Index

• type Job
• type WorkQueue
• type Worker
  • func NewWorker() Worker
  • func (w Worker) Run(wq chan WorkQueue)
• type WorkerPool
  • func NewWorkerPool(workerLen int) *WorkerPool
  • func (wp *WorkerPool) Run()

Types

type Job

type Job interface {
	Do() error
}

定义任务接口,只要接口类型实现了Do()方法就实现了job接口 如果Do方法没有错误返回，可以返回nil

type WorkQueue

type WorkQueue chan Job

type Worker

type Worker struct {
	JobQueue chan Job
}

-----------worker--------- 每一个被初始化的worker都会在后期单独占用一个协程 初始化的时候会先把自己的JobQueue传递到Worker通道中， 然后阻塞读取自己的JobQueue，读到一个Job就执行Job对象的Do()方法。

func NewWorker

func NewWorker() Worker

NewWorker初始化Worker

func (Worker) Run

func (w Worker) Run(wq chan WorkQueue)

运行作业池中的任务

type WorkerPool

type WorkerPool struct {
	JobQueue chan Job //WorkerPool的Job通道
	// contains filtered or unexported fields
}

-------------------工作池(WorkerPool)------------ 初始化时会按照传入的num，启动num个后台协程，然后循环读取Job通道里面的数据， 读到一个数据时，再获取一个可用的Worker，并将Job对象传递到该Worker的chan通道 工作池原理： 1. 整个过程中 每个Worker都会被运行在一个协程中，在整个WorkerPool中就会有num可空闲的Worker 2. 当来一条数据的时候，就会在工作池中去一个空闲的Worker去执行该Job，当工作池中没有可用的worker时 就会阻塞等待一个空闲的worker。

func NewWorkerPool

func NewWorkerPool(workerLen int) *WorkerPool

func (*WorkerPool) Run

func (wp *WorkerPool) Run()