mirror of
https://github.com/duke-git/lancet.git
synced 2026-02-17 03:02:28 +08:00
feat: add ForEachConcurrent
This commit is contained in:
@@ -53,6 +53,7 @@ import (
|
|||||||
- [Flatten](#Flatten)
|
- [Flatten](#Flatten)
|
||||||
- [FlattenDeep](#FlattenDeep)
|
- [FlattenDeep](#FlattenDeep)
|
||||||
- [ForEach](#ForEach)
|
- [ForEach](#ForEach)
|
||||||
|
- [ForEachConcurrent](#ForEachConcurrent)
|
||||||
- [ForEachWithBreak](#ForEachWithBreak)
|
- [ForEachWithBreak](#ForEachWithBreak)
|
||||||
- [GroupBy](#GroupBy)
|
- [GroupBy](#GroupBy)
|
||||||
- [GroupWith](#GroupWith)
|
- [GroupWith](#GroupWith)
|
||||||
@@ -1179,6 +1180,43 @@ func main() {
|
|||||||
}
|
}
|
||||||
```
|
```
|
||||||
|
|
||||||
|
### <span id="ForEachConcurrent">ForEachConcurrent</span>
|
||||||
|
|
||||||
|
<p>对slice并发执行foreach操作。</p>
|
||||||
|
|
||||||
|
<b>函数签名:</b>
|
||||||
|
|
||||||
|
```go
|
||||||
|
func ForEachConcurrent[T any](slice []T, iteratee func(index int, item T), numThreads int)
|
||||||
|
```
|
||||||
|
|
||||||
|
<b>示例:<span style="float:right;display:inline-block;">[运行]()</span></b>
|
||||||
|
|
||||||
|
```go
|
||||||
|
import (
|
||||||
|
"fmt"
|
||||||
|
"github.com/duke-git/lancet/v2/slice"
|
||||||
|
)
|
||||||
|
|
||||||
|
func main() {
|
||||||
|
nums := []int{1, 2, 3, 4, 5, 6, 7, 8}
|
||||||
|
|
||||||
|
result := make([]int, len(nums))
|
||||||
|
|
||||||
|
addOne := func(index int, value int) {
|
||||||
|
result[index] = value + 1
|
||||||
|
}
|
||||||
|
|
||||||
|
slice.ForEachConcurrent(nums, addOne, 4)
|
||||||
|
|
||||||
|
fmt.Println(result)
|
||||||
|
|
||||||
|
// Output:
|
||||||
|
// [2 3 4 5 6 7 8 9]
|
||||||
|
}
|
||||||
|
```
|
||||||
|
|
||||||
|
|
||||||
### <span id="ForEachWithBreak">ForEachWithBreak</span>
|
### <span id="ForEachWithBreak">ForEachWithBreak</span>
|
||||||
|
|
||||||
<p>遍历切片的元素并为每个元素调用iteratee函数,当iteratee函数返回false时,终止遍历。</p>
|
<p>遍历切片的元素并为每个元素调用iteratee函数,当iteratee函数返回false时,终止遍历。</p>
|
||||||
|
|||||||
@@ -53,6 +53,7 @@ import (
|
|||||||
- [Flatten](#Flatten)
|
- [Flatten](#Flatten)
|
||||||
- [FlattenDeep](#FlattenDeep)
|
- [FlattenDeep](#FlattenDeep)
|
||||||
- [ForEach](#ForEach)
|
- [ForEach](#ForEach)
|
||||||
|
- [ForEachConcurrent](#ForEachConcurrent)
|
||||||
- [ForEachWithBreak](#ForEachWithBreak)
|
- [ForEachWithBreak](#ForEachWithBreak)
|
||||||
- [GroupBy](#GroupBy)
|
- [GroupBy](#GroupBy)
|
||||||
- [GroupWith](#GroupWith)
|
- [GroupWith](#GroupWith)
|
||||||
@@ -1177,6 +1178,42 @@ func main() {
|
|||||||
}
|
}
|
||||||
```
|
```
|
||||||
|
|
||||||
|
### <span id="ForEachConcurrent">ForEachConcurrent</span>
|
||||||
|
|
||||||
|
<p>Applies the iteratee function to each item in the slice concurrently.</p>
|
||||||
|
|
||||||
|
<b>Signature:</b>
|
||||||
|
|
||||||
|
```go
|
||||||
|
func ForEachConcurrent[T any](slice []T, iteratee func(index int, item T), numThreads int)
|
||||||
|
```
|
||||||
|
|
||||||
|
<b>Example:<span style="float:right;display:inline-block;">[Run]()</span></b>
|
||||||
|
|
||||||
|
```go
|
||||||
|
import (
|
||||||
|
"fmt"
|
||||||
|
"github.com/duke-git/lancet/v2/slice"
|
||||||
|
)
|
||||||
|
|
||||||
|
func main() {
|
||||||
|
nums := []int{1, 2, 3, 4, 5, 6, 7, 8}
|
||||||
|
|
||||||
|
result := make([]int, len(nums))
|
||||||
|
|
||||||
|
addOne := func(index int, value int) {
|
||||||
|
result[index] = value + 1
|
||||||
|
}
|
||||||
|
|
||||||
|
slice.ForEachConcurrent(nums, addOne, 4)
|
||||||
|
|
||||||
|
fmt.Println(result)
|
||||||
|
|
||||||
|
// Output:
|
||||||
|
// [2 3 4 5 6 7 8 9]
|
||||||
|
}
|
||||||
|
```
|
||||||
|
|
||||||
### <span id="ForEachWithBreak">ForEachWithBreak</span>
|
### <span id="ForEachWithBreak">ForEachWithBreak</span>
|
||||||
|
|
||||||
<p>Iterates over elements of slice and invokes function for each element, when iteratee return false, will break the for each loop.</p>
|
<p>Iterates over elements of slice and invokes function for each element, when iteratee return false, will break the for each loop.</p>
|
||||||
|
|||||||
@@ -8,6 +8,47 @@ import (
|
|||||||
"sync"
|
"sync"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
// ForEachConcurrent applies the iteratee function to each item in the slice concurrently.
|
||||||
|
// Play: todo
|
||||||
|
func ForEachConcurrent[T any](slice []T, iteratee func(index int, item T), numThreads int) {
|
||||||
|
sliceLen := len(slice)
|
||||||
|
if sliceLen == 0 {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
if numThreads <= 0 {
|
||||||
|
numThreads = 1
|
||||||
|
}
|
||||||
|
|
||||||
|
var wg sync.WaitGroup
|
||||||
|
|
||||||
|
chunkSize := (sliceLen + numThreads - 1) / numThreads
|
||||||
|
|
||||||
|
for i := 0; i < numThreads; i++ {
|
||||||
|
start := i * chunkSize
|
||||||
|
end := start + chunkSize
|
||||||
|
|
||||||
|
if start >= sliceLen {
|
||||||
|
break
|
||||||
|
}
|
||||||
|
|
||||||
|
if end > sliceLen {
|
||||||
|
end = sliceLen
|
||||||
|
}
|
||||||
|
|
||||||
|
wg.Add(1)
|
||||||
|
go func(start, end int) {
|
||||||
|
defer wg.Done()
|
||||||
|
|
||||||
|
for j := start; j < end; j++ {
|
||||||
|
iteratee(j, slice[j])
|
||||||
|
}
|
||||||
|
}(start, end)
|
||||||
|
}
|
||||||
|
|
||||||
|
wg.Wait()
|
||||||
|
}
|
||||||
|
|
||||||
// MapConcurrent applies the iteratee function to each item in the slice concurrently.
|
// MapConcurrent applies the iteratee function to each item in the slice concurrently.
|
||||||
// Play: todo
|
// Play: todo
|
||||||
func MapConcurrent[T any, U any](slice []T, iteratee func(index int, item T) U, numThreads int) []U {
|
func MapConcurrent[T any, U any](slice []T, iteratee func(index int, item T) U, numThreads int) []U {
|
||||||
|
|||||||
@@ -429,6 +429,23 @@ func ExampleForEach() {
|
|||||||
// [2 3 4]
|
// [2 3 4]
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func ExampleForEachConcurrent() {
|
||||||
|
nums := []int{1, 2, 3, 4, 5, 6, 7, 8}
|
||||||
|
|
||||||
|
result := make([]int, len(nums))
|
||||||
|
|
||||||
|
addOne := func(index int, value int) {
|
||||||
|
result[index] = value + 1
|
||||||
|
}
|
||||||
|
|
||||||
|
ForEachConcurrent(nums, addOne, 4)
|
||||||
|
|
||||||
|
fmt.Println(result)
|
||||||
|
|
||||||
|
// Output:
|
||||||
|
// [2 3 4 5 6 7 8 9]
|
||||||
|
}
|
||||||
|
|
||||||
func ExampleForEachWithBreak() {
|
func ExampleForEachWithBreak() {
|
||||||
numbers := []int{1, 2, 3, 4, 5}
|
numbers := []int{1, 2, 3, 4, 5}
|
||||||
|
|
||||||
|
|||||||
@@ -410,6 +410,73 @@ func TestForEach(t *testing.T) {
|
|||||||
assert.Equal([]int{3, 4, 5, 6, 7}, result)
|
assert.Equal([]int{3, 4, 5, 6, 7}, result)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestForEachConcurrent(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
assert := internal.NewAssert(t, "TestForEachConcurrent")
|
||||||
|
|
||||||
|
t.Run("single thread", func(t *testing.T) {
|
||||||
|
numbers := []int{1, 2, 3, 4, 5, 6, 7, 8, 9}
|
||||||
|
result := make([]int, len(numbers))
|
||||||
|
|
||||||
|
addOne := func(index int, value int) {
|
||||||
|
result[index] = value + 1
|
||||||
|
}
|
||||||
|
|
||||||
|
ForEachConcurrent(numbers, addOne, 1)
|
||||||
|
|
||||||
|
expected := []int{2, 3, 4, 5, 6, 7, 8, 9, 10}
|
||||||
|
assert.Equal(expected, result)
|
||||||
|
})
|
||||||
|
|
||||||
|
t.Run("normal", func(t *testing.T) {
|
||||||
|
numbers := []int{1, 2, 3, 4, 5, 6, 7, 8, 9}
|
||||||
|
result := make([]int, len(numbers))
|
||||||
|
|
||||||
|
addOne := func(index int, value int) {
|
||||||
|
result[index] = value + 1
|
||||||
|
}
|
||||||
|
|
||||||
|
ForEachConcurrent(numbers, addOne, 4)
|
||||||
|
|
||||||
|
expected := []int{2, 3, 4, 5, 6, 7, 8, 9, 10}
|
||||||
|
assert.Equal(expected, result)
|
||||||
|
})
|
||||||
|
|
||||||
|
t.Run("negative threads", func(t *testing.T) {
|
||||||
|
numbers := []int{1, 2, 3, 4, 5, 6, 7, 8, 9}
|
||||||
|
result := make([]int, len(numbers))
|
||||||
|
|
||||||
|
addOne := func(index int, value int) {
|
||||||
|
result[index] = value + 1
|
||||||
|
}
|
||||||
|
|
||||||
|
ForEachConcurrent(numbers, addOne, -4)
|
||||||
|
|
||||||
|
expected := []int{2, 3, 4, 5, 6, 7, 8, 9, 10}
|
||||||
|
assert.Equal(expected, result)
|
||||||
|
})
|
||||||
|
|
||||||
|
t.Run("high number threads", func(t *testing.T) {
|
||||||
|
numbers := make([]int, 1000)
|
||||||
|
for i := range numbers {
|
||||||
|
numbers[i] = i
|
||||||
|
}
|
||||||
|
|
||||||
|
result := make([]int, len(numbers))
|
||||||
|
|
||||||
|
addOne := func(index int, value int) {
|
||||||
|
result[index] = value + 1
|
||||||
|
}
|
||||||
|
|
||||||
|
ForEachConcurrent(numbers, addOne, 50)
|
||||||
|
|
||||||
|
for i, item := range numbers {
|
||||||
|
assert.Equal(item+1, result[i])
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
func TestForEachWithBreak(t *testing.T) {
|
func TestForEachWithBreak(t *testing.T) {
|
||||||
t.Parallel()
|
t.Parallel()
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user