ForEach function

main/main.go

@@ -1,9 +1,10 @@
 package main
 
 import (
+	"fmt"
+
 	. "github.com/go-func/stream"
 	"github.com/go-func/stream/collectors"
-	"fmt"
 )
 
 func main() {
@@ -47,6 +48,9 @@ func main() {
 		.Collect(&collectors.SummarizingFloat64{Mapper: func(item interface{}) float64 {
 			return float64(item.(int))
 		}})
+		.ForEach(func(item interface{}) {
+			fmt.Println(item)
+		})
 	*/
 	fmt.Println(tmp)
 }

stream/forEach.go

@@ -0,0 +1,37 @@
+package stream
+
+import "runtime"
+
+//ForEach 对任意数组操作，针对每个元素执行一遍指定的行为
+func (s *Stream) ForEach(action func(interface{})) {
+	if s.isParallel {
+		count := runtime.NumCPU()
+		ch1 := make(chan interface{}, count)
+		ch2 := make(chan interface{})
+		for i := 0; i < count; i++ {
+			go func() {
+				for true {
+					if value, ok := <-ch1; !ok {
+						break
+					} else {
+						action(value)
+					}
+				}
+				ch2 <- 0
+			}()
+		}
+		go func() {
+			for _, v := range s.list {
+				ch1 <- v
+			}
+			close(ch1)
+		}()
+		for i := 0; i < count; i++ {
+			<-ch2
+		}
+	} else {
+		for _, v := range s.list {
+			action(v)
+		}
+	}
+}