sync

Documentation

Index

• func AsStream(done <-chan struct{}, values ...interface{}) <-chan interface{}
• func CreateCases(chs ...chan int) []reflect.SelectCase
• func FanInRec(chans ...<-chan interface{}) <-chan interface{}
• func FanInReflect(chans ...<-chan interface{}) <-chan interface{}
• func FanOut(ch <-chan interface{}, out []chan interface{}, async bool)
• func NewSemaphore(capacity int) sync.Locker
• func NewWorker(id int, ch chan Token, nextCh chan Token)
• func Or(channels ...<-chan interface{}) <-chan interface{}
• func OrDone(channels ...<-chan interface{}) <-chan interface{}
• func TakeN(done <-chan struct{}, valueStream <-chan interface{}, num int) <-chan interface{}
• type Mutex
  • func NewMutex() *Mutex
  • func (m *Mutex) IsLocked() bool
  • func (m *Mutex) Lock()
  • func (m *Mutex) LockTimeout(timeout time.Duration) bool
  • func (m *Mutex) TryLock() bool
  • func (m *Mutex) Unlock()
• type Once
  • func (o *Once) Do(f func() error) error
• type RWMap
  • func NewRWMap(n int) *RWMap
  • func (m *RWMap) Delete(k int)
  • func (m *RWMap) Each(f func(k, v int) bool)
  • func (m *RWMap) Get(k int) (int, bool)
  • func (m *RWMap) Len() int
  • func (m *RWMap) Set(k int, v int)
• type RecursiveMutex
  • func (m *RecursiveMutex) Lock()
  • func (m *RecursiveMutex) Unlock()
• type Token
• type TokenRecursiveMutex
  • func (m *TokenRecursiveMutex) Lock(token int64)
  • func (m *TokenRecursiveMutex) Unlock(token int64)

Functions

func AsStream

func AsStream(done <-chan struct{}, values ...interface{}) <-chan interface{}

AsStream 把一个slice转换为Stream

func CreateCases

func CreateCases(chs ...chan int) []reflect.SelectCase

CreateCases 基于反射构建动态select

func FanInRec

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

FanInRec 二分递归加速多个chan往最终的一个chan扇入

func FanInReflect

func FanInReflect(chans ...<-chan interface{}) <-chan interface{}

FanInReflect 多个chan 扇入到一个目标chan

func FanOut

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

FanOut chan的扇出事一个chan的数据往多个目标chan扇出

func NewSemaphore

func NewSemaphore(capacity int) sync.Locker

NewSemaphore 创建一个新的信号量

func NewWorker

func NewWorker(id int, ch chan Token, nextCh chan Token)

NewWorker 基于令牌信号传递goroutine的执行顺序

func Or

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

Or 是使用深度递归的方式实现or chan

func OrDone

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

OrDone 是对Or chan的优化版本

func TakeN

func TakeN(done <-chan struct{}, valueStream <-chan interface{}, num int) <-chan interface{}

TakeN 只取流中的前 n 个数据

Types

type Mutex

type Mutex struct {
	// contains filtered or unexported fields
}

Mutex 使用chan实现互斥锁

func NewMutex

func NewMutex() *Mutex

NewMutex 使用锁需要初始化

func (*Mutex) IsLocked

func (m *Mutex) IsLocked() bool

IsLocked 锁是否已被持有

func (*Mutex) Lock

func (m *Mutex) Lock()

Lock 请求锁，直到获取到

func (*Mutex) LockTimeout

func (m *Mutex) LockTimeout(timeout time.Duration) bool

LockTimeout 加入一个超时的设置

func (*Mutex) TryLock

func (m *Mutex) TryLock() bool

TryLock 尝试获取锁

func (*Mutex) Unlock

func (m *Mutex) Unlock()

Unlock 解锁

type Once

type Once struct {
	// contains filtered or unexported fields
}

Once 无限重试的once

func (*Once) Do

func (o *Once) Do(f func() error) error

Do 传入的函数f有返回值error，如果初始化失败，需要返回失败的error给调用者

type RWMap

type RWMap struct {
	sync.RWMutex // 读写锁保护下面的map字段
	// contains filtered or unexported fields
}

func NewRWMap

func NewRWMap(n int) *RWMap

NewRWMap 新建一个RWMap

func (*RWMap) Delete

func (m *RWMap) Delete(k int)

func (*RWMap) Each

func (m *RWMap) Each(f func(k, v int) bool)

func (*RWMap) Get

func (m *RWMap) Get(k int) (int, bool)

func (*RWMap) Len

func (m *RWMap) Len() int

func (*RWMap) Set

func (m *RWMap) Set(k int, v int)

type RecursiveMutex

type RecursiveMutex struct {
	sync.Mutex
	// contains filtered or unexported fields
}

RecursiveMutex 包装一个Mutex,实现可重入

func (*RecursiveMutex) Lock

func (m *RecursiveMutex) Lock()

func (*RecursiveMutex) Unlock

func (m *RecursiveMutex) Unlock()

type Token

type Token struct{}

type TokenRecursiveMutex

type TokenRecursiveMutex struct {
	sync.Mutex
	// contains filtered or unexported fields
}

TokenRecursiveMutex Token方式的递归锁（可重入）

func (*TokenRecursiveMutex) Lock

func (m *TokenRecursiveMutex) Lock(token int64)

Lock 请求锁，需要传入token

func (*TokenRecursiveMutex) Unlock

func (m *TokenRecursiveMutex) Unlock(token int64)

Unlock 释放锁