feat: add 使用goroutine写高并发逻辑

backend/goroutine.md

@@ -0,0 +1,151 @@
+# 使用goroutine写高并发逻辑
+
+我们使用消费者和生产者的逻辑构建一个通用的函数库, 所需的功能模块清单.
+
+1. JobQueue chan Job, 用来接受job的队列
+2. WorkerPool chan chan Job, 消费job的worker管道`集合`.
+3. Len in, 控制worker的数目
+
+## Job, Payload
+
+Job作为管道中传输的单位, 而payload是job行为定义.
+
+```go
+// Payload For example usage
+type Payload interface {
+	Do() (err error)
+}
+
+// Job represents the job to be run
+type Job struct {
+	Payload Payload
+}
+
+// NewJob represents the job
+func NewJob(payload Payload) Job {
+	return Job{Payload: payload}
+}
+```
+
+
+## Dispatcher
+我们创建一个Dispatcher负责构建初始化JobQueue, WorkerPool同时准备len数目的worker用来`接通`生产端和消费端.
+
+```go
+// Dispatcher -> Processing Data
+type Dispatcher struct {
+	// JobQueue A buffered channel that we can send work requests on.
+	JobQueue chan Job
+	// A pool of workers channels that are registered with the dispatcher
+	WorkerPool chan chan Job
+	// Len represents num of worker
+	Len int
+}
+
+// NewDispatcher represents Processing Data
+func NewDispatcher(maxWorkers int) *Dispatcher {
+	pool := make(chan chan Job, maxWorkers)
+	jobQueue := make(chan Job, 1)
+	return &Dispatcher{JobQueue: jobQueue, WorkerPool: pool, Len: maxWorkers}
+}
+
+// Run starting n number of workers
+func (d *Dispatcher) Run() {
+	for i := 0; i < d.Len; i++ {
+		worker := NewWorker(d.WorkerPool)
+		worker.Start()
+	}
+	go d.dispatch()
+}
+
+func (d *Dispatcher) dispatch() {
+	for {
+		select {
+		case job := <-d.JobQueue:
+			logrus.Info("Store a job into jobChannel")
+			go func(job Job) {
+				jobChannel := <-d.WorkerPool
+				jobChannel <- job
+			}(job)
+		}
+	}
+}
+
+// AddJob represents Generate Data
+func (d *Dispatcher) AddJob(j Job) {
+	d.JobQueue <- j
+}
+```
+
+## Worker
+```go
+// Worker represents the worker that executes the job
+type Worker struct {
+	WorkerPool chan chan Job
+	JobChannel chan Job
+	quit       chan bool
+}
+
+// NewWorker represents the worker
+func NewWorker(workerPool chan chan Job) Worker {
+	return Worker{
+		WorkerPool: workerPool,
+		JobChannel: make(chan 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 {
+			w.WorkerPool <- w.JobChannel
+			select {
+			case job := <-w.JobChannel:
+				if err := job.Payload.Do(); err != nil {
+					logrus.Error("Error do payload function", err.Error())
+				}
+			case <-w.quit:
+				return
+			}
+		}
+	}()
+}
+
+// Stop signals the worker to stop listening for work requests.
+func (w Worker) Stop() {
+	go func() {
+		w.quit <- true
+	}()
+}
+```
+
+## Test
+
+```go
+// Example:
+// MyPayloadTest For example usage
+type MyPayloadTest struct {
+	Num int
+}
+
+// Do do something
+func (p MyPayloadTest) Do() (err error) {
+	//Do a job sample
+	logrus.Info("I am working Do", p.Num)
+	time.Sleep(time.Second * 2)
+	err = nil
+	return
+}
+
+func main() {
+	dispatcher := NewDispatcher(10)
+	dispatcher.Run()
+	for i := 1; i < 40; i++ {
+		job := NewJob(MyPayloadTest{i})
+		dispatcher.AddJob(job)
+	}
+
+	time.Sleep(time.Second * 50)
+}
+```