Go 异步任务

异步任务在开发中很常见，用来做解耦。本文介绍一下异步队列的实现的几个问题，并且结合三方库实现来分析。

有下面的几个关键点:

1. 用户代码（任务）如何封装
2. 数据的存放（数据存放在哪里？就是一个读取队列）
3. worker的管理（worker的数量，worker执行是否支持超时时间，worker的异常恢复）

带着上面的问题，对比https://github.com/golang-queue/queue的实现，说明一下。

用户代码如何封装

对于任务来说，最重要的是 函数操作，也就是对应的代码逻辑。go中是可以将方法作为参数传递的，方法也是一种类型。所以我们定义下面的方法，方法签名如下：

type TaskFunc func(ctx context.Context) error

还可以配置方法的callback逻辑，比如重试次数，重试间隔，重试error的判断等等

抽象出一个结构体来表示

https://github.com/golang-queue/queue/blob/master/job/job.go#L15

数据的存放

这是很好拓展的地方，可以支持多种存储媒介和中间件，比如基于内存实现的循环队列，redis，rocketmq。

在实现上就是接口抽象功能，依赖倒转。接口有下面的两个功能

1. 存数据
2. 取数据

https://github.com/golang-queue/queue/blob/master/core/worker.go

解释一下QueuedMessage接口和Worker中的Run方法

1. QueuedMessage

   用来做数据转换的。

2. Run

   用来执行函数，表示执行的任务。

worker的管理

worker管理涉及到下面几个方面

1. worker的数量限制
2. worker执行时候的超时时间
3. worker执行时候的异常panic
4. workder从队列中获取需要处理的处理，并且支持请求超时操作
5. 服务关闭之后worker也需要操作

我们来看golang-queue/queue中的实现是什么？

通过metric来记录queue在运行期间具体的情况

https://github.com/golang-queue/queue/blob/master/metric.go#L20

并且通过 channel 来做限制。

每次在goroutine启动和停止的时候通过metric来计数。并且会调用schedule来发信号，给ready发送信号。

goroutine在启动的时候会select ready。

work的异常情况，在调用task的处理函数的时候，肯定要用到defer来做error恢复，并且通过channel来通信，context来实现超时控制。

具体的原理，我们从下面的代码开始来分析。

https://github.com/golang-queue/queue/blob/master/queue.go#L285

// 对于start来说，是一个死循环，会启动一个goroutine从work中获取数据，当前goroutine等待结果，并且启动goroutine来执行，此Goroutine叫做worker。
func (q *Queue) start() {
	// QueuedMessage 表示message
	tasks := make(chan core.QueuedMessage, 1)
	// 启动一个goroutine来处理任务
	// 从work中获取任务，并且启动一个goroutine来处理任务
	for {
		// check worker number
    // 做调度的，就是检查work的数量
		q.schedule()
		
    // 数量不够，需要堵塞
		select {
		// wait worker ready
		case <-q.ready:
		case <-q.quit:
			return
		}

	// 启动一个goRoutine从 work中获取数据
		q.routineGroup.Run(func() {
			for {
				// 从队列中获取一个请求
				t, err := q.worker.Request()
				// 没有消息，或者有错误
				if t == nil || err != nil {
					// 有错误
					if err != nil {
						select {
              // 队列退出，关闭掉task，
						case <-q.quit:
							if !errors.Is(err, ErrNoTaskInQueue) {
								close(tasks)
								return
							}
              // 等待一秒再次从work中抓取新数据
						case <-time.After(time.Second):
							// sleep 1 second to fetch new task
						}
					}
				}
				if t != nil { // 说明取到了消息
					tasks <- t
					return
				}
				// 说明t为nil但是没有错误
				select {
				case <-q.quit:
					if !errors.Is(err, ErrNoTaskInQueue) {
						close(tasks)
						return
					}
				default:
				}
			}
		})
		// 这就是从queue中获取一个task，之后将此task提交给work来实现
		task, ok := <-tasks
		if !ok {
			return
		}
		// 所以，这里并没有维护所谓的goroutine池，因为go的编程是不需要这些玩意的。goroutine已经很轻量级的了，直接提交运行就好了
		// start new task
		q.metric.IncBusyWorker()
		q.routineGroup.Run(func() {
			q.work(task)
		})
	}
}

func (q *Queue) work(task core.QueuedMessage) {
	var err error
	// 来处理一些内部的错误，在这里会减去worker的数量，并且重新schedule
	defer func() {
		q.metric.DecBusyWorker()
		e := recover()
		if e != nil {
			q.logger.Errorf("panic error: %v", e)
		}
		q.schedule()

		// increase success or failure number
		if err == nil && e == nil {
			q.metric.IncSuccessTask()
		} else {
			q.metric.IncFailureTask()
		}
	}()
	// 运行任务，可以看到这里的代码就是为了包装一下
	if err = q.run(task); err != nil {
		q.logger.Errorf("runtime error: %s", err.Error())
	}
}

func (q *Queue) run(task core.QueuedMessage) error {
	data := task.(*job.Message)
	if data.Task == nil {
		data = job.Decode(task.Bytes())
		data.Data = data.Payload
	}

	return q.handle(data)
}

func (q *Queue) handle(m *job.Message) error {
	// create channel with buffer size 1 to avoid goroutine leak
	// 这是go中很创建的做法，一个channel中有数据，但并没有被其他的任何的goroutine操作的话，也是会被gc掉的
	done := make(chan error, 1) // 完成的信号channel
	panicChan := make(chan interface{}, 1) // panic的channel
	startTime := time.Now() 
	ctx, cancel := context.WithTimeout(context.Background(), m.Timeout)
	defer func() {
		cancel()
	}()

	// run the job 启动goroutine来运行一个job
	go func() {
		// handle panic issue
		defer func() {
			if p := recover(); p != nil {
				panicChan <- p
			}
		}()

		// run custom process function
		var err error
		// 做重试逻辑，这里的重试逻辑还可以指定重试的错误，比如as，基于那种类型的错误来做重试操作等。
		b := &backoff.Backoff{
			Min:    m.RetryMin,
			Max:    m.RetryMax,
			Factor: m.RetryFactor,
			Jitter: m.Jitter,
		}
		delay := m.RetryDelay
   // backoff都是通过for循环来做的
	loop:
		for {
      // 两种形式，一种是直接function，一直是通过message
			if m.Task != nil {
				err = m.Task(ctx)
			} else {
				err = q.worker.Run(ctx, m)
			}

 	    // 不需要重试就直接返回，如果有错误就开始重试，并且利用time来做重试时间的控制
			if err == nil || m.RetryCount == 0 {
				break
			}
			m.RetryCount--

			if m.RetryDelay == 0 {
				delay = b.Duration()
			}
			// 这里用select来做操作
			select {
			case <-time.After(delay): // retry delay
				q.logger.Infof("retry remaining times: %d, delay time: %s", m.RetryCount, delay)
			case <-ctx.Done(): // timeout reached // ctx完成就直接返回
				err = ctx.Err()
				break loop
			}
		}

		done <- err
	}()
	// 当前的goroutine在等待结果，
	select {
	case p := <-panicChan:
		panic(p)
	case <-ctx.Done(): // timeout reached
		return ctx.Err()
	case <-q.quit: // shutdown service
		// cancel job
		cancel()

		leftTime := m.Timeout - time.Since(startTime)
		// wait job
		select {
		case <-time.After(leftTime):
			return context.DeadlineExceeded
		case err := <-done: // job finish
			return err
		case p := <-panicChan:
			panic(p)
		}
	case err := <-done: // job finish
		return err
	}
}

有个问题，如何保证程序退出的时候这些work可以执行结束呢？利用waitGroup实现。

https://github.com/golang-queue/queue/blob/master/thread.go