syncs

Documentation

Overview

Package syncs 异步控制.

包括 RoutineGroup、LeakyBucket.

Index

• Variables
• func Consume[E any](cacheSize, sliceMax, routine int, minTime time.Duration, ...) int64
• func ConsumeFunc[E any](cacheSize, sliceMax, routine int, minTime time.Duration, ...) int64
• type LeakyBucket
  • func NewLeakyBucket(capacity uint, leakInterval time.Duration) *LeakyBucket
  • func (p *LeakyBucket) Consume(drop uint)
  • func (p *LeakyBucket) TryConsume(drop uint) bool
• type Limit
  • func NewLimit(qps uint) *Limit
  • func (p *Limit) Try() error
  • func (p *Limit) Wait()
• type ProducerAndConsumer
• type RoutineGroup
  • func NewRoutineGroup(size uint) *RoutineGroup
  • func (p *RoutineGroup) Add(delta uint)
  • func (p *RoutineGroup) Done()
  • func (p *RoutineGroup) Wait()

Examples

• ConsumeFunc

Variables

var (
	ErrSizeLessZero = errors.New("size less than 0")
	ErrLimit        = errors.New("limit time")
)

Functions

func Consume

func Consume[E any](cacheSize, sliceMax, routine int, minTime time.Duration,
	pac ProducerAndConsumer[E],
) int64

Consume 消费. cacheSize 缓存数量. sliceMax 每次消费最大数量. routine 消费协程数量. minTime 预计的最小消费时间. producer 生产者. consumer 消费者.

func ConsumeFunc

func ConsumeFunc[E any](cacheSize, sliceMax, routine int, minTime time.Duration,
	producFunc func(produceChan chan E),
	consumFunc func(num int, elems []E),
) int64

ConsumeFunc 消费. cacheSize 缓存数量. sliceMax 每次消费最大数量. routine 消费协程数量. minTime 预计的最小消费时间. producer 生产者. consumer 消费者. nolint

Example

package main

import (
	"fmt"
	"time"

	"github.com/xuender/oils/syncs"
)

func main() {
	syncs.ConsumeFunc(10, 3, 3, time.Millisecond, func(produceChan chan int) {
		for i := 0; i < 100; i++ {
			produceChan <- i
		}
	}, func(num int, elems []int) {
		time.Sleep(time.Microsecond)
		fmt.Println(num, elems)
	})
}

Types

type LeakyBucket

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

LeakyBucket 漏桶算法.

func NewLeakyBucket

func NewLeakyBucket(capacity uint, leakInterval time.Duration) *LeakyBucket

NewLeakyBucket 新建漏桶.

func (*LeakyBucket) Consume

func (p *LeakyBucket) Consume(drop uint)

Consume 消费，如果无法消费，则 Sleep 直到可以消费.

func (*LeakyBucket) TryConsume

func (p *LeakyBucket) TryConsume(drop uint) bool

TryConsume 尝试消费，消费失败返回 false.

type Limit

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

Limit 频率限制.

func NewLimit

func NewLimit(qps uint) *Limit

NewLimit 频率限制，qps每秒限制次数.

func (*Limit) Try

func (p *Limit) Try() error

Try 尝试执行.

func (*Limit) Wait

func (p *Limit) Wait()

Wait 等待执行.

type ProducerAndConsumer

type ProducerAndConsumer[E any] interface {
	Produc(produceChan chan E)
	Consum(num int, elems []E)
}

type RoutineGroup

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

RoutineGroup 协程组，是sync.WaitGroup的增强版本.

func NewRoutineGroup

func NewRoutineGroup(size uint) *RoutineGroup

NewRoutineGroup 协程组，控制协程总数量.

func (*RoutineGroup) Add

func (p *RoutineGroup) Add(delta uint)

Add 增加协程.

func (*RoutineGroup) Done

func (p *RoutineGroup) Done()

Done 协程完成.

func (*RoutineGroup) Wait

func (p *RoutineGroup) Wait()

Wait 等待全部完成.