README
¶
concurrency — 并发控制
提供 Worker Pool、信号量、速率限制器、并发限制器等并发原语，简化 Go 中的并发任务管理。
主要组件
WorkerPool — Worker 池
批量并发执行任务，控制最大并发数：
import "github.com/leeforge/framework/concurrency"
pool := concurrency.NewWorkerPool(10) // 10 个 worker
pool.Start()
// 提交任务
pool.Submit(concurrency.JobFunc(func() error {
// 执行工作
return nil
}))
// 批量提交
pool.SubmitBatch(jobs)
// 等待所有任务完成后停止
pool.Stop()
Semaphore — 信号量
控制同时进入某段代码的 goroutine 数量：
sem := concurrency.NewSemaphore(5) // 同时最多 5 个
sem.WithSemaphore(func() {
// 并发安全的关键操作
callExternalAPI()
})
ConcurrencyLimiter — 并发限制器
limiter := concurrency.NewConcurrencyLimiter(20)
err := limiter.Execute(func() error {
return processItem(item)
})
// 批量并发执行（自动限制并发数）
errs := limiter.ExecuteBatch([]func() error{fn1, fn2, fn3})
RateLimiter — 令牌桶速率限制器
// 每秒最多 100 个请求
rl := concurrency.NewRateLimiter(100)
defer rl.Stop()
if rl.Allow() {
// 允许执行
}
// 或阻塞等待令牌
rl.Wait()
ParallelExecutor — 并行执行器
executor := concurrency.NewParallelExecutor(8) // 最多 8 路并行
errs := executor.Execute([]func() error{
func() error { return processA() },
func() error { return processB() },
func() error { return processC() },
})
Future — 异步结果
executor := concurrency.NewAsyncExecutor(4)
executor.Start()
defer executor.Stop()
future := executor.SubmitAsync(job)
// 阻塞获取结果
result := future.Get()
// 带超时获取
result, err := future.GetWithTimeout(5 * time.Second)
TaskQueue — 任务队列
queue := concurrency.NewTaskQueue(100) // 缓冲区 100
queue.Start(4) // 4 个消费者
queue.Submit(func() {
sendEmail(user)
})
queue.Stop() // 等待队列排空后停止
注意事项
WorkerPool.Stop()有 30 秒超时，若任务长期阻塞请设置合理超时RateLimiter.Wait()使用轮询，精度约 10ms，高精度场景建议使用time.TickerFuture是一次性的，Get 之后不能重复读取
Documentation
¶
Index ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type AsyncExecutor ¶
type AsyncExecutor struct {
// contains filtered or unexported fields
}
AsyncExecutor 异步执行器
func (*AsyncExecutor) SubmitAsync ¶
func (a *AsyncExecutor) SubmitAsync(job Job) *Future
SubmitAsync 提交异步任务
type ConcurrencyLimiter ¶
type ConcurrencyLimiter struct {
// contains filtered or unexported fields
}
ConcurrencyLimiter 并发限制器
func NewConcurrencyLimiter ¶
func NewConcurrencyLimiter(maxConcurrent int) *ConcurrencyLimiter
NewConcurrencyLimiter 创建并发限制器
func (*ConcurrencyLimiter) Execute ¶
func (cl *ConcurrencyLimiter) Execute(fn func() error) error
Execute 在并发限制下执行
func (*ConcurrencyLimiter) ExecuteBatch ¶
func (cl *ConcurrencyLimiter) ExecuteBatch(fns []func() error) []error
ExecuteBatch 批量执行
type ContextPool ¶
type ContextPool struct {
// contains filtered or unexported fields
}
ContextPool 上下文感知的 Worker 池
func (*ContextPool) StopWithContext ¶
func (p *ContextPool) StopWithContext(ctx context.Context) error
StopWithContext 带上下文停止
type ParallelExecutor ¶
type ParallelExecutor struct {
// contains filtered or unexported fields
}
ParallelParallel 并行执行器
func NewParallelExecutor ¶
func NewParallelExecutor(maxWorkers int) *ParallelExecutor
NewParallelExecutor 创建并行执行器
func (*ParallelExecutor) Execute ¶
func (p *ParallelExecutor) Execute(fns []func() error) []error
Execute 并行执行
type PooledExecutor ¶
type PooledExecutor struct {
// contains filtered or unexported fields
}
PooledExecutor 池化执行器
type RateLimiter ¶
type RateLimiter struct {
// contains filtered or unexported fields
}
RateLimiter 速率限制器
type Semaphore ¶
type Semaphore struct {
// contains filtered or unexported fields
}
Semaphore 信号量
func (*Semaphore) WithSemaphore ¶
func (s *Semaphore) WithSemaphore(fn func())
WithSemaphore 在信号量控制下执行
type SimpleResult ¶
type SimpleResult struct {
// contains filtered or unexported fields
}
SimpleResult 简单结果
type WorkerPool ¶
type WorkerPool struct {
// contains filtered or unexported fields
}
WorkerPool Worker 池
func (*WorkerPool) SubmitBatch ¶
func (p *WorkerPool) SubmitBatch(jobs []Job) error
SubmitBatch 批量提交任务
Source Files
Click to show internal directories.
Click to hide internal directories.