gostream

README

gostream

go Map Reduce Sorted Filter Limit stream , a simple library

进度 已实现

• Limit 支持流截断
• Combine 支持流之间的拼接
• Filter 支持流过滤
• Sorted 支持流排序
• Reduce 支持流 Reduce
• Map 支持流 Map
• MultipartMap 支持并发流
• Copy 流拷贝
• Distinct 排除重复元素 拟定适用技术 bloom 过滤器 / Cuckoo Filter 过滤器

进度 待实现

• Manager 流的管理器，管理多个流，做到简单，快捷操作多个流

Documentation here: https://godoc.org/github.com/felix0080/gostream

Usage

package main
import (
   "fmt"
   "strconv"
   . "github.com/felix0080/gostream"
)
type IntSlice []interface{}

func (s IntSlice) Len() int { return len(s) }
func (s IntSlice) Swap(i, j int){ s[i], s[j] = s[j], s[i] }
func (s IntSlice) Less(i, j int) bool {
   return s[i].(int) < s[j].(int)
} 
type Item int
func (s Item) HashCode() []byte {
   return []byte(strconv.Itoa(int(s)))
}
func main() {
  //调用BuildStream 方法，传入实现了sort 接口的slice，若不需要排序，则无需实现sort
   //定义一个slice.填充元素
   a:=[]interface{}{4,3,2,1}
   //使用实现了sort的type 包装，并调用调用BuildStream
   s,err:=BuildStream(IntSlice(a))
   if err != nil {
       fmt.Println(err)
       return
   }
   //定义一个slice.填充元素
   a1:=[]interface{}{6,5,8,7}
   //使用实现了sort的type 包装,并调用调用BuildStream
   s1,err:=BuildStream(IntSlice(a1))
   if err != nil {
       fmt.Println(err)
       return
   }
   //将两个流合并为1个，并且以s为主流
   value:=s.Combine(s1).
   //排序，对流的排序，若没有实现sort接口，将不会进行排序
       Sorted().
   //只保留前6个元素
       Limit(6).
   //map 将函数作用到每一个流元素上
       Map(func(i interface{}) interface{} {
           return i.(int)+1
       }).
   //Filter 过滤函数返回值为true的元素，形成新流
       Filter(func(i interface{}) bool {
           return i.(int)==7
       }).
   //Reduce 将前两个元素计算后的结果和第三个元素计算，以此类推
   Reduce(func(i interface{}, i2 interface{}) interface{} {
           return i.(int)+i2.(int)
       })//==20
   //s.Limit(10)
   fmt.Println("after arr: ", s.Collect())
   fmt.Println("after value: ", value)
  
   s2:=s.Copy()
   s2.Map(func(i interface{}) interface{} {
       return i.(int)+1
   }).Sorted()
   fmt.Println(s2.Collect())
   fmt.Println(s.Collect())
   a3 := []interface{}{Item(1), Item(2), Item(4), Item(4)}
   s3, err := BuildStream(IntSlice(a3))
   if err != nil {
       fmt.Println(err)
       return
   }
   s3.Distinct()
   fmt.Println(s3.Collect())
}
  

Documentation

Overview

Package gostream implements a stream slice.

Usage

调用BuildStream 方法，传入实现了sort 接口的slice，若不需要排序，则无需实现sort

//定义一个slice.填充元素
a:=[]interface{}{4,3,2,1}
//使用实现了sort的type 包装，并调用调用BuildStream
s,err:=BuildStream(IntSlice(a))
if err != nil {
	fmt.Println(err)
	return
}
//定义一个slice.填充元素
a1:=[]interface{}{6,5,8,7}
//使用实现了sort的type 包装,并调用调用BuildStream
s1,err:=BuildStream(IntSlice(a1))
if err != nil {
	fmt.Println(err)
	return
}
//将两个流合并为1个，并且以s为主流
value:=s.Combine(s1).
//排序，对流的排序，若没有实现sort接口，将不会进行排序
	Sorted().
//只保留前6个元素
	Limit(6).
//map 将函数作用到每一个流元素上
	Map(func(i interface{}) interface{} {
		return i.(int)+1
	}).
//Filter 过滤函数返回值为true的元素，形成新流
	Filter(func(i interface{}) bool {
		return i.(int)==7
	}).
//Reduce 将前两个元素计算后的结果和第三个元素计算，以此类推
Reduce(func(i interface{}, i2 interface{}) interface{} {
		return i.(int)+i2.(int)
	})//==20
//s.Limit(10)
fmt.Println("after arr: ", s.Collect())
fmt.Println("after value: ", value)

可使用copy,将流深度拷贝成新的Stream

a:=[]interface{}{4,3,2,1}
s,err:=BuildStream(IntSlice(a))
if err != nil {
	fmt.Println(err)
	return
}
s1:=s.Copy()
s1.Map(func(i interface{}) interface{} {
	return i.(int)+1
}).Sorted()
fmt.Println(s1.Collect())
fmt.Println(s.Collect())

去除重复元素样例

func (s Item) HashCode() []byte {
	return []byte(strconv.Itoa(int(s)))
}
func (s IntSlice) Len() int      { return len(s) }
func (s IntSlice) Swap(i, j int) { s[i], s[j] = s[j], s[i] }
func (s IntSlice) Less(i, j int) bool {
	return s[i].(int) < s[j].(int)
}
func TestStream_Distinct(t *testing.T) {
	a := []interface{}{Item(1), Item(2), Item(4), Item(4)}
	s, err := BuildStream(IntSlice(a))
	if err != nil {
		fmt.Println(err)
		return
	}
	s.Distinct()
	fmt.Println(s.Collect())
}

Index

• type Stream
  • func BuildStream(array interface{}) (*Stream, error)
  • func (stream *Stream) Collect() interface{}
  • func (stream *Stream) Combine(anotherStream *Stream) *Stream
  • func (stream *Stream) Copy() *Stream
  • func (stream *Stream) Distinct() *Stream
  • func (stream *Stream) Filter(f func(interface{}) bool) *Stream
  • func (stream *Stream) Limit(len int) *Stream
  • func (stream *Stream) Map(f func(interface{}) interface{}) *Stream
  • func (stream *Stream) MultipartMap(worknum int, f func(interface{}) interface{}) *Stream
  • func (stream *Stream) Reduce(f func(interface{}, interface{}) interface{}) interface{}
  • func (stream *Stream) Sorted() *Stream
• type Unique

Types

type Stream

type Stream struct {
	// contains filtered or unexported fields
}

定义一个流，内部为 实现了Sort 接口的 Slice(interfalce{}). 若没有实现Sort接口，sort方法将不可使用 使用方法类似于java8 stream 欢迎贡献，大家一起完善此库

func BuildStream

func BuildStream(array interface{}) (*Stream, error)

Example:

type IntSlice []interface{}

func (s IntSlice) Len() int { return len(s) }
func (s IntSlice) Swap(i, j int){ s[i], s[j] = s[j], s[i] }
func (s IntSlice) Less(i, j int) bool {
	return s[i].(int) < s[j].(int)
}
a:=[]interface{}{4,3,2,1}
s,err:=BuildStream(IntSlice(a))

func (*Stream) Collect

func (stream *Stream) Collect() interface{}

收集流，将流的内部数组返回。

func (*Stream) Combine

func (stream *Stream) Combine(anotherStream *Stream) *Stream

组合流，将两个stream 简单组合在一起

func (*Stream) Copy

func (stream *Stream) Copy() *Stream

流拷贝，拷贝一个新流，返回与以前的流相同大小，容量的新流，且将内容拷贝进去

s2:=s.Copy()
s2.Map(func(i interface{}) interface{} {
	return i.(int)+1
}).Sorted()
fmt.Println(s2.Collect())
fmt.Println(s.Collect())

func (*Stream) Distinct

func (stream *Stream) Distinct() *Stream

此方法将需要流中的元素实现 unique接口 hashCode（） 算法采用布谷鸟算法

Example
func (s Item) HashCode() []byte {
	return []byte(strconv.Itoa(int(s)))
}
func (s IntSlice) Len() int      { return len(s) }
func (s IntSlice) Swap(i, j int) { s[i], s[j] = s[j], s[i] }
func (s IntSlice) Less(i, j int) bool {
	return s[i].(int) < s[j].(int)
}
func TestStream_Distinct(t *testing.T) {
	a := []interface{}{Item(1), Item(2), Item(4), Item(4)}
	s, err := BuildStream(IntSlice(a))
	if err != nil {
		fmt.Println(err)
		return
	}
	s.Distinct()
	fmt.Println(s.Collect())
}

func (*Stream) Filter

func (stream *Stream) Filter(f func(interface{}) bool) *Stream

设置条件满足的被过滤

func (*Stream) Limit

func (stream *Stream) Limit(len int) *Stream

用户获取指定数量的流

func (*Stream) Map

func (stream *Stream) Map(f func(interface{}) interface{}) *Stream

用于映射每个元素到对应的结果

func (*Stream) MultipartMap

func (stream *Stream) MultipartMap(worknum int, f func(interface{}) interface{}) *Stream

func (stream *Stream) Copy()*Stream {
	a1:=[]interface{}{6,5,"asd",7}
	a2:=make([]interface{},len(a1))
	fmt.Println(fmt.Sprintf("%p %p",a1,a2))
	copy(a2,a1)
	fmt.Println(a2)
}

用于映射每个元素到对应的结果

func (*Stream) Reduce

func (stream *Stream) Reduce(f func(interface{}, interface{}) interface{}) interface{}

当数组小于2时，返回Nil

func (*Stream) Sorted

func (stream *Stream) Sorted() *Stream

用户对流进行排序 需要自行实现sort接口才可使用此方法

type Unique

type Unique interface {
	HashCode() []byte
}