并发型

Worker Pool

固定数量的 goroutine 处理任务队列，控制并发度。

问题场景

需要并发处理大量任务，但要限制同时运行的 goroutine 数量。

解决方案

创建固定数量的 worker goroutine，从 channel 中获取任务。

Go 实现

worker-pool.go

1type Task struct {
2    ID      string
3    Payload interface{}
4}
5 
6type Result struct {
7    TaskID string
8    Data   interface{}
9    Err    error
10}
11 
12type WorkerPool struct {
13    tasks   chan Task
14    results chan Result
15    done    chan struct{}
16}
17 
18func NewWorkerPool(workers, queueSize int) *WorkerPool {
19    pool := &WorkerPool{
20        tasks:   make(chan Task, queueSize),
21        results: make(chan Result, queueSize),
22        done:    make(chan struct{}),
23    }
24 
25    for i := 0; i < workers; i++ {
26        go pool.worker(i)
27    }
28 
29    return pool
30}
31 
32func (p *WorkerPool) worker(id int) {
33    for task := range p.tasks {
34        result := p.process(task)
35        p.results <- result
36    }
37}
38 
39func (p *WorkerPool) Submit(task Task) {
40    p.tasks <- task
41}
42 
43func (p *WorkerPool) Close() {
44    close(p.tasks)
45}
46 
47// Web3 应用：批量查询代币余额
48func (p *WorkerPool) process(task Task) Result {
49    address := task.Payload.(string)
50    balance, err := getTokenBalance(address)
51    return Result{TaskID: task.ID, Data: balance, Err: err}
52}

Web3 应用场景

批量查询链上数据、并发处理交易事件。

优点

+控制资源使用
+避免 goroutine 爆炸
+背压机制

缺点

-需要合理配置 worker 数量

策略模式 Pipeline 模式

Go 实现

worker-pool.go

1type Task struct {
2    ID      string
3    Payload interface{}
4}
5 
6type Result struct {
7    TaskID string
8    Data   interface{}
9    Err    error
10}
11 
12type WorkerPool struct {
13    tasks   chan Task
14    results chan Result
15    done    chan struct{}
16}
17 
18func NewWorkerPool(workers, queueSize int) *WorkerPool {
19    pool := &WorkerPool{
20        tasks:   make(chan Task, queueSize),
21        results: make(chan Result, queueSize),
22        done:    make(chan struct{}),
23    }
24 
25    for i := 0; i < workers; i++ {
26        go pool.worker(i)
27    }
28 
29    return pool
30}
31 
32func (p *WorkerPool) worker(id int) {
33    for task := range p.tasks {
34        result := p.process(task)
35        p.results <- result
36    }
37}
38 
39func (p *WorkerPool) Submit(task Task) {
40    p.tasks <- task
41}
42 
43func (p *WorkerPool) Close() {
44    close(p.tasks)
45}
46 
47// Web3 应用：批量查询代币余额
48func (p *WorkerPool) process(task Task) Result {
49    address := task.Payload.(string)
50    balance, err := getTokenBalance(address)
51    return Result{TaskID: task.ID, Data: balance, Err: err}
52}