Link multiple channels into one channel.
+Link multiple channels into one channel until cancel the context.
Signature: @@ -114,7 +114,7 @@ func main() { ### FanIn -merge multiple channels into one channel.
+merge multiple channels into one channel until cancel the context.
Signature: @@ -277,7 +277,7 @@ func main() { ### OrDone -Read a channel into one channel, will close until cancel context.
+Read a channel into another channel, will close until cancel context.
Signature: diff --git a/docs/concurrency_zh-CN.md b/docs/concurrency_zh-CN.md new file mode 100644 index 0000000..98cd5e8 --- /dev/null +++ b/docs/concurrency_zh-CN.md @@ -0,0 +1,390 @@ +# Concurrency +并发包包含一些支持并发编程的功能。例如：goroutine, channel, async等。 + + + +## 源码: + +- [https://github.com/duke-git/lancet/blob/main/concurrency/channel.go](https://github.com/duke-git/lancet/blob/main/concurrency/channel.go) + + + +## 用法: +```go +import ( + "github.com/duke-git/lancet/v2/concurrency" +) +``` + + + +## 目录 +### Channel +- [NewChannel](#NewChannel) +- [Bridge](#Bridge) +- [FanIn](#FanIn) +- [Generate](#Generate) +- [Or](#Or) +- [OrDone](#OrDone) +- [Repeat](#Repeat) +- [RepeatFn](#RepeatFn) +- [Take](#Take) +- [Tee](#Tee) + + + +## 文档 + + +### Channel +### NewChannel +返回一个 Channel 指针实例
+ +函数签名: + +```go +type Channel struct {} +func NewChannel() *Channel +``` +例子: + +```go +package main + +import ( + "fmt" + "github.com/duke-git/lancet/v2/concurrency" +) + +func main() { + c := concurrency.NewChannel() +} +``` + + + +### Bridge + +将多个通道链接到一个通道，直到取消上下文。
+ +函数签名: + +```go +func (c *Channel) Bridge(ctx context.Context, chanStream <-chan <-chan any) <-chan any +``` +例子: + +```go +package main + +import ( + "context" + "fmt" + "github.com/duke-git/lancet/v2/concurrency" +) + +func main() { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + c := concurrency.NewChannel() + genVals := func() <-chan <-chan any { + chanStream := make(chan (<-chan any)) + go func() { + defer close(chanStream) + for i := 0; i < 10; i++ { + stream := make(chan any, 1) + stream <- i + close(stream) + chanStream <- stream + } + }() + return chanStream + } + + index := 0 + for val := range c.Bridge(ctx, genVals()) { + fmt.Printf("%v ", val) //0 1 2 3 4 5 6 7 8 9 + } +} +``` + + + + +### FanIn + +将多个通道合并为一个通道，直到取消上下文
+ +函数签名: + +```go +func (c *Channel) FanIn(ctx context.Context, channels ...<-chan any) <-chan any +``` +例子: + +```go +package main + +import ( + "context" + "fmt" + "github.com/duke-git/lancet/v2/concurrency" +) + +func main() { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + c := concurrency.NewChannel() + channels := make([]<-chan any, 3) + + for i := 0; i < 3; i++ { + channels[i] = c.Take(ctx, c.Repeat(ctx, i), 3) + } + + mergedChannel := c.FanIn(ctx, channels...) + + for val := range mergedChannel { + fmt.Println("\t%d\n", val) //1,2,1,0,0,1,0,2,2 (order not for sure) + } +} +``` + + +### Repeat + +返回一个chan，将参数`values`重复放入chan，直到取消上下文。
+ +函数签名: + +```go +func (c *Channel) Repeat(ctx context.Context, values ...any) <-chan any +``` +例子: + +```go +package main + +import ( + "context" + "fmt" + "github.com/duke-git/lancet/v2/concurrency" +) + +func main() { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + c := concurrency.NewChannel() + intStream := c.Take(ctx, c.Repeat(ctx, 1, 2), 5) + + for v := range intStream { + fmt.Println(v) //1, 2, 1, 2, 1 + } +} +``` + + + + +### RepeatFn + +返回一个chan，重复执行函数fn，并将结果放入返回的chan，直到取消上下文。
+ +函数签名: + +```go +func (c *Channel) RepeatFn(ctx context.Context, fn func() any) <-chan any +``` +例子: + +```go +package main + +import ( + "context" + "fmt" + "github.com/duke-git/lancet/v2/concurrency" +) + +func main() { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + fn := func() any { + s := "a" + return s + } + c := concurrency.NewChannel() + dataStream := c.Take(ctx, c.RepeatFn(ctx, fn), 3) + + for v := range dataStream { + fmt.Println(v) //a, a, a + } +} +``` + + + +### Or + +将一个或多个通道读取到一个通道中，当任何读取通道关闭时将结束读取。
+ +函数签名: + +```go +func (c *Channel) Or(channels ...<-chan any) <-chan any +``` +例子: + +```go +package main + +import ( + "context" + "fmt" + "github.com/duke-git/lancet/v2/concurrency" +) + +func main() { + sig := func(after time.Duration) <-chan any { + c := make(chan interface{}) + go func() { + defer close(c) + time.Sleep(after) + }() + return c + } + + start := time.Now() + + c := concurrency.NewChannel() + <-c.Or( + sig(1*time.Second), + sig(2*time.Second), + sig(3*time.Second), + sig(4*time.Second), + sig(5*time.Second), + ) + + fmt.Println("done after %v", time.Since(start)) //1.003s +} +``` + + + + +### OrDone + +将一个通道读入另一个通道，直到取消上下文。
+ +函数签名: + +```go +func (c *Channel) OrDone(ctx context.Context, channel <-chan any) <-chan any +``` +例子: + +```go +package main + +import ( + "context" + "fmt" + "github.com/duke-git/lancet/v2/concurrency" +) + +func main() { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + c := concurrency.NewChannel() + intStream := c.Take(ctx, c.Repeat(ctx, 1), 3) + + for val := range c.OrDone(ctx, intStream) { + fmt.Println(val) //1 + } +} +``` + + + + +### Take + +返回一个chan，其值从另一个chan获取，直到取消上下文。
+ +函数签名: + +```go +func (c *Channel) Take(ctx context.Context, valueStream <-chan any, number int) <-chan any +``` +例子: + +```go +package main + +import ( + "context" + "fmt" + "github.com/duke-git/lancet/v2/concurrency" +) + +func main() { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + numbers := make(chan any, 5) + numbers <- 1 + numbers <- 2 + numbers <- 3 + numbers <- 4 + numbers <- 5 + defer close(numbers) + + c := concurrency.NewChannel() + intStream := c.Take(ctx, numbers, 3) + + for val := range intStream { + fmt.Println(val) //1, 2, 3 + } +} +``` + + + +### Tee + +将一个通道分成两个通道，直到取消上下文。
+ +函数签名: + +```go +func (c *Channel) Tee(ctx context.Context, in <-chan any) (<-chan any, <-chan any) +``` +例子: + +```go +package main + +import ( + "context" + "fmt" + "github.com/duke-git/lancet/v2/concurrency" +) + +func main() { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + c := concurrency.NewChannel() + inStream := c.Take(ctx, c.Repeat(ctx, 1, 2), 4) + + out1, out2 := c.Tee(ctx, inStream) + for val := range out1 { + fmt.Println(val) //1 + fmt.Println(<-out2) //1 + } +} +``` \ No newline at end of file