[Go]通用的 MapReduce 工具函数

前言

最近在测试学习 aws s3 sdk 中的 Multi Part Upload 功能，其基本步骤就是 CreateMultipartUpload 后， 串行或并行地 UploadPart ，最后 CompleteMultipartUpload 或 AbortMultipartUpload 收尾。为了最高效率地完成整个传输，中间的 UploadPart 部分使用多个 goroutine 并发地上传是最快地。因此尝试着写了一下，并完美地实现。

扩展

虽然已经完成对应功能的开发和测试，但仔细分析一下，发现有大量的模式代码，比如：

• 1. 创建指定个数的 goroutine, 并使用 sync.WaitGroup 管理和同步.
• 2. 使用 chan 提供待处理数据,并接受处理结果
• 3. 看起来整个处理流程就是典型的 map-reduce 结构 或者 说是 Java Stream/ParallelStream 中的 Map, Reduce.

网上搜索一下, 发现很多人也有这个需求,也写了不少库，但实测了一下，发现根本不好用。于是决定自己再造一个“自己觉得比较好的”轮子，因此有了 mapreduce 和本篇文章。

主要功能函数

• func Map[T any, R any](ctx context.Context, inputs []T, mapper MapperFunc[T, R]) ResultsMap[T, R]
  这是最简单的同步 Map, 通过泛型的 T 和 R 支持任意类型的数据转换
• func ParallelMap[T any, R any](ctx context.Context, inputs []T, concurrency int, name string, mapper MapperFunc[T, R]) ResultsMap[T, R]
  这是并发的Map,内部回启动最多 2+concurrency 个 goroutine, 并发的处理完 inputs 中的所有数据. 并且结果可以按照输入的顺序排序。
• func StreamMap[T any, R any](ctx context.Context, concurrency int, name string, startIndex int64, chInput chan T, mapper MapperFunc[T, R]) chan *OutputItem[T, R]
  类似纤程池的形态, 可以无限地处理 chInput 中的数据,并将结果写入 OutputItem.

额外说明

错误处理

作为一个并发处理框架，对于错误情况也应该能很好的支持，有的时候, 一项元素处理失败了不影响整体的流程处理, 但有的时候其中一项失败, 就不需要继续进行(比如 S3 的 multi part upload, 如果其中一部分失败, 那其他的部分再上传也没有意义了)。因此代码中定义了 OperationType 类型, 其值分为
Continue 或 Stop , 框架只根据这个值确认是否继续处理, 而不是根据 mapper 函数是否返回 error.

结果返回

并发处理时, 每个 Item 的处理时长/顺序等是不同的，而且有可能因为错误造成部分输入元素尚未处理即结束，因此返回的结果默认情况下不一定能和输入顺序一一对应，因此采用了 Map 的方式保存输入序号 => 结果。

排序

s3 的 multi part upload 在调用 CompleteMultipartUpload 时参数 Parts 需要是排好序的,因此通过 ConvertResult 函数对结果进行排序。

测试代码

注意: 并发处理带错误数据的时候, 由于错误项的处理顺序比较随机, 因此我使用了 concurrency: 1 的方式保证 UT 能顺利判断。如果将 concurrency 更改为大于1的情况, 其 want 不一定能满足. 比如: “error with stop” 时, 如果 concurrency > 1, 结果有可能就不是 [1 2 3 0] 而是 [1 2 3 0 4 5] 了, 这种属于正常现象(自己更改测试一下即可理解 )

func TestMap(t *testing.T) {type args struct {ctx         context.Contextinputs      []stringconcurrency intmapper      MapperFunc[string, int]}tests := []struct {name     stringargs     argswant     []intwantErrs []erroropType   OperationType}{{name: "all successful",args: args{ctx: context.Background(),inputs: []string{"0", "1", "2", "3", "4", "5", "6", "7", "8", "9"}, concurrency: runtime.NumCPU(), mapper: convertStopFunc,},want:     []int{0, 1, 2, 3, 4, 5, 6, 7, 8, 9},wantErrs: []error{nil, nil, nil, nil, nil, nil, nil, nil, nil, nil},opType:   Continue,},{name:     "error with continue",args:     args{ctx: context.Background(), inputs: []string{"1", "not", "3"}, concurrency: 1, mapper: convertContinueFunc},want:     []int{1, 0, 3},wantErrs: []error{nil, numberErrHelper("not"), nil},opType:   Continue, // 出现过错误,但忽略了. 如果采用 Continue 的方式来处理错误, 则只能自己遍历 ResultsMap 的结果集才知道是否有错误},{// 注意: 如果并发度 concurrency > 1, 则结果个数不确定, 但肯定至少有一个错误的name: "error with stop",args: args{ctx: context.Background(), inputs: []string{"1", "2", "3", "not_exist", "4", "5", "6"},concurrency: 1, mapper: convertStopFunc},want:     []int{1, 2, 3, 0},wantErrs: []error{nil, nil, nil, numberErrHelper("not_exist")},opType:   Stop,},{name: "error last", // 最后一个数据出错时,其返回的结果数组长度和输入数组的长度相同. 因此不能依靠数组长度来判断是否有问题.args: args{ctx: context.Background(), inputs: []string{"1", "2", "3", "not_exist"},concurrency: 1, mapper: convertStopFunc},want:     []int{1, 2, 3, 0},wantErrs: []error{nil, nil, nil, numberErrHelper("not_exist")},opType:   Stop,},}for _, tt := range tests {t.Run(tt.name, func(t *testing.T) {if true {//使用 Map 串行转换got := Map(tt.args.ctx, tt.args.inputs, tt.args.mapper)//flog.Infof("Map name=%s, got=%+v", tt.name, got)realResult, errs, opType := got.ConvertResult()du.GoAssertEqual(t, tt.want, realResult, "want")du.GoAssertEqual(t, tt.wantErrs, errs, "wantErrs")du.GoAssertEqual(t, tt.opType, opType, "opType")}if true {//使用 ParallelMap 并行转换got := ParallelMap(tt.args.ctx, tt.args.inputs, tt.args.concurrency, tt.name, tt.args.mapper)//flog.Infof("ParallelMap name=%s, got=%+v", tt.name, got)realResult, errs, opType := got.ConvertResult()du.GoAssertEqual(t, tt.want, realResult, "want")du.GoAssertEqual(t, tt.wantErrs, errs, "wantErrs")du.GoAssertEqual(t, tt.opType, opType, "opType")}})}
}func TestStreamMap(t *testing.T) {ctx := context.Background()inItemCount := 10000chInput := make(chan string)go func() {for i := 0; i < inItemCount; i++ {idx := rand.Intn(100)chInput <- fmt.Sprintf("%d", idx)}close(chInput)}()//启动 100 个 纤程并行处理 inItemCount(10000) 个数据的转换chOutput := StreamMap(ctx, 100, "testStreamMap", 100, chInput, convertContinueFunc)mapResultCount := 0for outItem := range chOutput {mapResultCount++flog.Debugf("outItem=%v", outItem)}du.GoAssertEqual(t, inItemCount, mapResultCount, "inItemCount")
}

##补充信息

• 1. 因为众所周知的原因, 以后 go-library 的代码将只更新 https://gitee.com/fishjam/go-library, 不再更新 github 上的版本.
• 2. S3 的 multi upload 不需要大家自己写, manager.NewUploader 已经提供了完整的实现, 比大多数人实现得更好。