concurrent

Documentation

Index

• func FanInRec(channels ...<-chan interface{}) <-chan interface{}
• func FanOut(in <-chan interface{}, out []chan interface{}, async bool)
• func MapChan(in <-chan interface{}, fn func(interface{}) interface{}) <-chan interface{}
• func MergeChannel(a, b <-chan interface{}) <-chan interface{}
• func OrDone(channels ...<-chan interface{}) <-chan interface{}
• func Orderly(tasks []*OrderlyTask)
• func Pipeline(in chan interface{}) <-chan interface{}
• func ReduceChan(in <-chan interface{}, fn func(r, v interface{}) interface{}) interface{}
• func SkipFn(ctx context.Context, valueStream <-chan interface{}, ...) <-chan interface{}
• func SkipN(ctx context.Context, valueStream <-chan interface{}, num int) <-chan interface{}
• func SkipWhile(ctx context.Context, valueStream <-chan interface{}, ...) <-chan interface{}
• func Stream(ctx context.Context, values ...interface{}) <-chan interface{}
• func TaskFn(ctx context.Context, valueStream <-chan interface{}, ...) <-chan interface{}
• func TaskN(ctx context.Context, valueStream <-chan interface{}, num int) <-chan interface{}
• func TaskWhile(ctx context.Context, valueStream <-chan interface{}, ...) <-chan interface{}
• type OrderlyTask
  • func NewOrderTask(fn func()) *OrderlyTask
  • func (o *OrderlyTask) Do()
• type WaitGroup

Functions

func FanInRec

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

FanInRec 扇入模式

func FanOut

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

FanOut 扇出模式

func MapChan

func MapChan(in <-chan interface{}, fn func(interface{}) interface{}) <-chan interface{}

func MergeChannel

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

MergeChannel 合并两个channel

func OrDone

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

OrDone 任意channel完成后返回

func Orderly

func Orderly(tasks []*OrderlyTask)

Orderly 顺序执行

func Pipeline

func Pipeline(in chan interface{}) <-chan interface{}

Pipeline 串联模式

func ReduceChan

func ReduceChan(in <-chan interface{}, fn func(r, v interface{}) interface{}) interface{}

func SkipFn

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

SkipFn 跳过满足条件的数据

func SkipN

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

SkipN 跳过流中的前N个数据

func SkipWhile

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

SkipWhile 跳过满足条件的数据,一旦不满足,当前这个元素以后的元素都会输出

func Stream

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

func TaskFn

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

TaskFn 筛选流中的数据,只保留满足条件的数据

func TaskN

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

TaskN 只取流中的前N个数据

func TaskWhile

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

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

Types

type OrderlyTask

type OrderlyTask struct {
	sync.WaitGroup
	// contains filtered or unexported fields
}

func NewOrderTask

func NewOrderTask(fn func()) *OrderlyTask

NewOrderTask 初始化任务

func (*OrderlyTask) Do

func (o *OrderlyTask) Do()

Do 执行任务

type WaitGroup

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