Go通用的 MapReduce 工具函数详解

func TestMap(t *testing.T) {
	type args struct {
		ctx         context.Context
		inputs      []string
		concurrency int
		mapper      MapperFunc[string, int]
	}
	tests := []struct {
		name     string
		args     args
		want     []int
		wantErrs []error
		opType   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 := 10000
	chInput := 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 := 0
	for outItem := range chOutput {
		mapResultCount++
		flog.Debugf("outItem=%v", outItem)
	}
	du.GoAssertEqual(t, inItemCount, mapResultCount, "inItemCount")
}