golang使用泛型实现mapreduce操作

1.使用面向对象的方式写

package streamimport ("fmt""log""reflect""sort""strconv""strings"
)type Stream[T any] struct {data      []TkeyBy     stringsortByNum stringsortByStr []string
}func FromElement[T any](data []T) *Stream[T] {return &Stream[T]{data: data,}
}// 过滤算子
type filterfunc[F any] func(F) boolfunc (s *Stream[T]) Filter(filterFun filterfunc[T]) *Stream[T] {var new []Tfor _, item := range s.data {isfiltered := filterFun(item)if isfiltered {continue}new = append(new, item)}s.data = newreturn s
}// 单行处理
type mapfunc[F any] func(F) Ffunc (s *Stream[T]) Map(mapFun mapfunc[T]) *Stream[T] {for idx, item := range s.data {ret := mapFun(item)s.data[idx] = ret}return s
}// 排序
func (s *Stream[T]) SortByNum(key string) *Stream[T] {s.sortByNum = keyif len(s.sortByStr) > 0 {s.sortByStr = nil}return s
}// 每次排序只能使用一种排
func (s *Stream[T]) SortByStr(keys ...string) *Stream[T] {s.sortByStr = keysif s.sortByNum != "" {s.sortByNum = ""}return s
}func (s *Stream[T]) Sort(esc bool) *Stream[T] {if s.sortByNum == "" && len(s.sortByStr) == 0 {log.Println("please call SortBy() before sort()")return s}if s.sortByNum != "" {sort.Slice(s.data, func(i, j int) bool {v := reflect.ValueOf(s.data[i]).Elem()field := v.FieldByName(s.sortByNum)if !field.IsValid() {log.Panicf("field=%s not valid", s.sortByNum)}idata := fmt.Sprintf("%v", field.Interface())num, err := strconv.ParseInt(idata, 10, 64)if err != nil {log.Panic("please use num when use sortByNum", idata)}v1 := reflect.ValueOf(s.data[j]).Elem()field1 := v1.FieldByName(s.sortByNum)if !field1.IsValid() {log.Panicf("field=%s not valid", s.sortByNum)}jdata := fmt.Sprintf("%v", field1.Interface())num1, err := strconv.ParseInt(jdata, 10, 64)if err != nil {log.Panic("please use num when use sortByNum")}if esc {return num < num1} else {return num > num1}})}if len(s.sortByStr) > 0 {sort.Slice(s.data, func(i, j int) bool {var ifinalv, jfinalv stringfor _, key := range s.sortByStr {v := reflect.ValueOf(s.data[i]).Elem()field := v.FieldByName(key)if !field.IsValid() {log.Panicf("field=%s not valid", key)}idata := fmt.Sprintf("%v", field.Interface())ifinalv = ifinalv + idata}for _, key := range s.sortByStr {v := reflect.ValueOf(s.data[j]).Elem()field := v.FieldByName(key)if !field.IsValid() {log.Panicf("field=%s not valid", key)}jdata := fmt.Sprintf("%v", field.Interface())jfinalv = jfinalv + jdata}// i 大于j的话 返回1 所以正序需要返回falseret := strings.Compare(ifinalv, jfinalv)if esc {return ret < 0}return ret >= 0})}return s
}// 设置聚合的key
func (s *Stream[T]) KeyBy(key string) *Stream[T] {s.keyBy = keyreturn s
}// reduce
// 暂时木有办法改变输出的结构
type reducefunc[F any] func([]F) Ffunc (s *Stream[T]) Reduce(reduceFun reducefunc[T]) *Stream[T] {if s.keyBy == "" {log.Fatal("please call keyby() before reduce()")return nil}var cache = make(map[string][]T)defer func() {cache = nil}()for _, item := range s.data {v := reflect.ValueOf(item).Elem()field := v.FieldByName(s.keyBy)key := field.String()lis, ok := cache[key]if !ok {lis = make([]T, 0)}lis = append(lis, item)cache[key] = lis}var new []Tfor _, lis := range cache {ret := reduceFun(lis)new = append(new, ret)}s.data = newreturn s
}// 返回个数
func (s *Stream[T]) Limit(n int) []T {if n > len(s.data) {n = len(s.data)}return s.data[0:n]
}func (s *Stream[T]) Print() {for idx, item := range s.data {log.Printf("idx=%d val=%v", idx, item)}
}func (s *Stream[T]) Result() []T {return s.data
}

测试例子

func TestTostream(t *testing.T) {FromElement([]*Student{&Student{"xyf", "数学", 101},&Student{"xyf", "语文", 108},&Student{"xyf", "外语", 101},}).Map(func(st *Student) *Student {st.Score = st.Score + 10return st}).Filter(func(st *Student) bool {return st.Name == "xyf"}).// SortByStr("Name", "Subject").SortByNum("Score").Sort(false).KeyBy("Name").Reduce(func(st []*Student) *Student {var ret = &Student{Name:    st[0].Name,Subject: "all",}for _, item := range st {ret.Score = ret.Score + item.Score}return ret}).Print()
}

 缺点：golang有点挫的在于不能在方法里面返回新的泛型类型，比如从student返回一个int类型。虽然能通过在struct定义俩个类型 但是万一要生成第三种类型就无能为力了，不可能一直往后加类型吧（这会导致定义类型超级长 写起来超级丑）。

2.通过函数的方式实现（简单举个例子）

type StreamV2[T any] struct {data []T
}func (s StreamV2[T]) Print() {for i, item := range s.data {log.Println("idx=", i, " value=", item)}
}func FromElementV2[T any](data []T) Stream[T] {return Stream[T]{data: data,}
}func Map[T any, K any](source Stream[T], mapfunc func(data T) K) StreamV2[K] {var ret []Kfor _, item := range source.data {ret1 := mapfunc(item)ret = append(ret, ret1)}return StreamV2[K]{data: ret,}
}

测试

func TestTostreamv2(t *testing.T) {stream1 := FromElementV2([]*Student{&Student{"xyf", "数学", 101},&Student{"xyf", "语文", 108},})stream2 := Map(stream1, func(f *Student) int {return f.Score})stream2.Print()
}

优缺点：这种方式能够将一种容器类型转化为另一种。缺点就是写过java的会吐血（因为搞大数据的朋友都喜欢使用类似builder模式的写法）