concurrent

Documentation

Index

• func Add(ctx context.Context, valueStream <-chan int, delta int) <-chan int
• func FanInRec(channels ...<-chan interface{}) <-chan interface{}
• func FanOut(in <-chan interface{}, n int, async bool) []chan interface{}
• func MergeChannel(a, b <-chan interface{}) <-chan interface{}
• func Multiply(ctx context.Context, valueStream <-chan int, factor int) <-chan int
• func Or(channels ...<-chan interface{}) <-chan interface{}
• func OrDone(channels ...<-chan interface{}) <-chan interface{}
• func SkipFn(ctx context.Context, valueStream <-chan interface{}, fn func(interface{}) bool) <-chan interface{}
• func SkipN(ctx context.Context, valueStream <-chan interface{}, n int) <-chan interface{}
• func SkipWhile(ctx context.Context, valueStream <-chan interface{}, fn func(interface{}) bool) <-chan interface{}
• func Stream(ctx context.Context, values ...interface{}) <-chan interface{}
• func TaskFn(ctx context.Context, valueStream <-chan interface{}, fn func(interface{}) bool) <-chan interface{}
• func TaskN(ctx context.Context, valueStream <-chan interface{}, n int) <-chan interface{}
• func TaskWhile(ctx context.Context, valueStream <-chan interface{}, fn func(interface{}) bool) <-chan interface{}
• type OrderlyTask
• type WaitGroup

Functions

func Add

func Add(ctx context.Context, valueStream <-chan int, delta int) <-chan int

func FanInRec

func FanInRec(channels ...<-chan interface{}) <-chan interface{}

FanInRec 扇入模式

func FanOut

func FanOut(in <-chan interface{}, n int, async bool) []chan interface{}

FanOut 扇出模式

func MergeChannel

func MergeChannel(a, b <-chan interface{}) <-chan interface{}

MergeChannel 合并两个channel

func Multiply

func Multiply(ctx context.Context, valueStream <-chan int, factor int) <-chan int

Multiply factor 乘法流水线中需要执行的数据操作

func Or

func Or(channels ...<-chan interface{}) <-chan interface{}

Or 复合channel 当任意一个channel关闭时，返回

func OrDone

func OrDone(channels ...<-chan interface{}) <-chan interface{}

OrDone 任意chan关闭时，返回 相比于Or来说减少性能消耗，性能高

func SkipFn

func SkipFn(ctx context.Context, valueStream <-chan interface{}, fn func(interface{}) bool) <-chan interface{}

SkipFn 跳过满足条件的数据 从输入的数据流中跳过满足给定条件的元素，然后将剩余的元素发送到输出通道 对于每个从 valueStream 接收到的元素，如果 fn 函数返回 true，则跳过该元素； 如果 fn 函数返回 false，则将该元素发送到输出通道 out 当上下文被取消或输入流关闭时，函数会停止并关闭输出通道

func SkipN

func SkipN(ctx context.Context, valueStream <-chan interface{}, n int) <-chan interface{}

SkipN 跳过前n个数据 从输入的数据流中跳过前n个元素，然后将剩余的元素发送到输出通道 先从输入流中读取并丢弃前n个元素，然后将后续所有元素转发到输出通道 当上下文被取消或输入流关闭时，函数会停止并关闭输出通道

func SkipWhile

func SkipWhile(ctx context.Context, valueStream <-chan interface{}, fn func(interface{}) bool) <-chan interface{}

SkipWhile 跳过满足条件的数据 跳过输入流中连续满足条件的元素，一旦遇到不满足条件的元素， 则从该元素开始（包括该元素），将后续所有元素都发送到输出通道 这类似于"跳过前缀"的操作，一旦条件不满足，后续所有元素都会被转发 当上下文被取消或输入流关闭时，函数会停止并关闭输出通道

func Stream

func Stream(ctx context.Context, values ...interface{}) <-chan interface{}

Stream 将离散数据转化为数据流 该函数接收一个上下文和一组值，将这些值发送到返回的通道中 当上下文被取消或值发送完毕时，通道会被关闭

func TaskFn

func TaskFn(ctx context.Context, valueStream <-chan interface{}, fn func(interface{}) bool) <-chan interface{}

TaskFn 只取满足条件的数据 从输入的数据流中过滤出满足给定条件的元素 函数会持续从输入流读取数据，只有当元素满足fn函数条件时才会发送到输出通道 当上下文被取消或输入流关闭时，函数会停止并关闭输出通道

func TaskN

func TaskN(ctx context.Context, valueStream <-chan interface{}, n int) <-chan interface{}

TaskN 只取前n个数据 从输入的数据流中获取前n个元素，发送到返回的通道中 当获取到n个元素、上下文被取消或输入流关闭时，函数会停止并关闭输出通道

func TaskWhile

func TaskWhile(ctx context.Context, valueStream <-chan interface{}, fn func(interface{}) bool) <-chan interface{}

TaskWhile 只取满足条件的数据,一旦不满足就不再取

Types

type OrderlyTask

type OrderlyTask struct {
	sync.WaitGroup
}

type WaitGroup

type WaitGroup interface {
	Add(int)
	Wait()
	Done()
	Do()
}