go-parallel: 一个并发任务处理工具

在日常开发过程中，通常我们会面对需要并发处理任务的需求，例如获取1000个商品的商品信息，但是接口只支持单次批量最多30个商品的限制，或者需要对多路召回商品的结果进行聚合操作。在上述场景下，我们通常需要对总体任务进行拆分，并通过并发的方式进行处理和计算。同时为了保证最终接口的性能，会对并发流程做超时处理。通常这需要我们花费部分精力去处理并发任务的分发和任务超时退出等相关逻辑的开发和验证，稍不注意就会导致协程泄漏。go-paralle这个并发任务管理的 golang 库，可以很好的帮助我们解决上述痛点，让我们将精力更多聚焦于开发具体业务逻辑。

1、简介

go-parallel 是一个用于管理并发任务的 golang 库，基于这个工具，您可以将精力聚焦于开发具体业务逻辑，而不必担心并发任务的处理及超时。 go-parallel基于生产消费者模型实现并发任务的计算结果收集，支持并发流程的整体超时机制的设置。

例如：您需要并发的从多个接口拉取数据，并对数据进行汇聚。您可以创建一个并发任务，并添加多个function或者Process(如果需要参数传入的话)，并配置一个全局的超时时间。那么您在超时时间内会收到全部接口的结果，或者除超时之外的接口结果。

2、安装

go get github.com/sjy3/go-parallel

3、使用

对于大多数情况来说。通过 AppendFunc 来添加自己并发任务的逻辑，具体入下：

func main() {
	po := go_parallel.NewParallelObject()

	po.AppendFunc("", func() interface{} {
		time.Sleep(10 * time.Millisecond)
		return "func 1"
	})
	po.AppendFunc("", func() interface{} {
		time.Sleep(10 * time.Millisecond)
		return "func 2"
	})

	data, isTimeout := po.Run()
	var result []string
	for _,d := range data{
		result = append(result, d.Data.(string))
	}
	fmt.Println("isTimeout = ", isTimeout)
	fmt.Println(result)
	fmt.Println(len(result))
}

默认情况下，整个并发处理流程会等全部的子流程执行结束退出，isTimeout 恒为 false。此外，go-parallal 还支持设置超时时间，只需要在po.Run()之前设置参数, 如果存在超时子任务的话，isTimeout = true

	po.SetTimeout(5000 * time.Millisecond)

您也可以通过 context 上下文信息来处理超时，context 方式的好处是可以继承父任务的超时时间。

	ctx, cancel = context.WithTimeout(p.ctx, timeout)
    po.SetContext(ctx)

4、参数传入

当子任务需要依赖外部传入参数时，需要使用 AppendProcess 模式，具体来说，需要首先声明传入参数的执行对象，并实现 ParallelProcessor 接口，具体例子如下：

type process struct {
	param []int
}

func (p process) Do() interface{} {
	var result int
	for _, i := range p.param {
		result += i
	}
	return result
}

func main() {

	po := go_parallel.NewParallelObject()
	po.AppendProcess("", process{
		[]int{1,2,3,4,5},
	})
	po.AppendProcess("", process{
		[]int{5,6,7,8,9},
	})
	po.SetTimeout(5000 * time.Millisecond)

	data, isTimeout := po.Run()

	var result []int
	for _,d := range data{
		result = append(result, d.Data.(int))
	}

	fmt.Println("isTimeout = ", isTimeout)
	fmt.Println(result)
}

5、子任务区分

如果不同子任务的返回结果处理方式需要区分处理的情况，AppendFunc 和 AppendProcess 都支持子任务名称的设置（默认可以传空字符串），返回结果中会回传，可用于处理流程的区分。

func main() {
	po := go_parallel.NewParallelObject()
	po.AppendFunc("getName", func() interface{} {
		time.Sleep(10 * time.Millisecond)
		return "func 1"
	})
	po.AppendFunc("getId", func() interface{} {
		time.Sleep(10 * time.Millisecond)
		return 10
	})
	po.SetTimeout(5000 * time.Millisecond)
	data, isTimeout := po.Run()
	fmt.Println("isTimeout = ", isTimeout)

	for _,d := range data{
		switch d.Name{
		case "getName":
			fmt.Println("the result = ",d.Data.(string))
		case "getId":
			fmt.Println("the result = ",d.Data.(int))
		}
	}
}

6、注意事项

超时时间的设置，只能保证整体流程的正常退出，对于用户通过 AppendFunc 和 AppendProcess 声明的子任务，需要用户保证流程的正常退出，否则会有协程泄漏的风险存在。