使用Golang实现开发中常用的【并发设计模式】

使用Golang实现开发中常用的【并发设计模式】

设计模式是解决常见问题的模板，可以帮助我们提升思维能力，编写更高效、可维护性更强的代码

屏障模式

未来模式

管道模式

协程池模式

发布订阅模式

下面是使用 Go 语言实现屏障模式、未来模式、管道模式、协程池模式和发布订阅模式的示例代码。

1. 屏障模式

屏障模式（Barrier Pattern）用于同步多个 goroutine，在所有 goroutine 都到达某个点之前，任何 goroutine 都不能继续执行。

package mainimport ("fmt""sync"
)func main() {const numWorkers = 5var wg sync.WaitGroupwg.Add(numWorkers)// 创建一个 Barrierbarrier := make(chan struct{}, numWorkers)for i := 0; i < numWorkers; i++ {go func(id int) {defer wg.Done()// 模拟工作fmt.Printf("Worker %d is working...\n", id)// 到达屏障点barrier <- struct{}{}// 等待所有 goroutine 到达屏障点<-barrier// 继续工作fmt.Printf("Worker %d continues after barrier.\n", id)}(i)}wg.Wait()close(barrier)
}

2. 未来模式

未来模式（Future Pattern）用于异步获取结果，通常用于长时间运行的任务。

package mainimport ("fmt""time"
)type Future interface {Get() (string, error)
}type futureImpl struct {resultChan chan stringerrChan    chan error
}func NewFuture() Future {f := &futureImpl{resultChan: make(chan string, 1),errChan:    make(chan error, 1),}go f.run()return f
}func (f *futureImpl) run() {// 模拟长时间运行的任务time.Sleep(2 * time.Second)select {case f.resultChan <- "Result":case f.errChan <- nil:}
}func (f *futureImpl) Get() (string, error) {select {case result := <-f.resultChan:return result, nilcase err := <-f.errChan:return "", err}
}func main() {future := NewFuture()fmt.Println("Doing other work...")result, err := future.Get()if err != nil {fmt.Println("Error:", err)} else {fmt.Println("Result:", result)}
}

3. 管道模式

管道模式（Pipeline Pattern）用于将多个 goroutine 串联起来，形成一个数据处理流水线。

package mainimport ("fmt""time"
)func produce(in chan<- int) {for i := 0; i < 10; i++ {in <- itime.Sleep(100 * time.Millisecond)}close(in)
}func process(out <-chan int, in chan<- int) {for v := range out {v *= 2in <- v}close(in)
}func consume(in <-chan int) {for v := range in {fmt.Println("Consumed:", v)}
}func main() {ch1 := make(chan int)ch2 := make(chan int)go produce(ch1)go process(ch1, ch2)go consume(ch2)time.Sleep(2 * time.Second)
}

4. 协程池模式

协程池模式（Coroutine Pool Pattern）用于管理和重用一组固定数量的 goroutine。

package mainimport ("fmt""sync""time"
)type Task func(int)type GoroutinePool struct {size    inttasks   chan Taskwg      sync.WaitGroupstop    boolstopCh  chan struct{}doneCh  chan struct{}
}func NewGoroutinePool(size int) *GoroutinePool {return &GoroutinePool{size:   size,tasks:  make(chan Task),stopCh: make(chan struct{}),doneCh: make(chan struct{}),}
}func (p *GoroutinePool) Start() {for i := 0; i < p.size; i++ {p.wg.Add(1)go p.worker(i)}
}func (p *GoroutinePool) worker(id int) {defer p.wg.Done()for {select {case task := <-p.tasks:task(id)case <-p.stopCh:p.doneCh <- struct{}{}return}}
}func (p *GoroutinePool) Submit(task Task) {p.tasks <- task
}func (p *GoroutinePool) Stop() {p.stop = trueclose(p.stopCh)p.wg.Wait()close(p.doneCh)
}func main() {pool := NewGoroutinePool(5)pool.Start()for i := 0; i < 20; i++ {task := func(id int) {fmt.Printf("Task %d handled by worker %d\n", i, id)time.Sleep(100 * time.Millisecond)}pool.Submit(task)}time.Sleep(1 * time.Second)pool.Stop()
}

5. 发布订阅模式

发布订阅模式（Publish-Subscribe Pattern）用于解耦消息的发送者和接收者。

package mainimport ("fmt""sync"
)type Event stringtype Subscriber func(Event)type Publisher struct {mu         sync.Mutexsubscribers map[Subscriber]struct{}
}func NewPublisher() *Publisher {return &Publisher{subscribers: make(map[Subscriber]struct{}),}
}func (p *Publisher) Subscribe(subscriber Subscriber) {p.mu.Lock()defer p.mu.Unlock()p.subscribers[subscriber] = struct{}{}
}func (p *Publisher) Unsubscribe(subscriber Subscriber) {p.mu.Lock()defer p.mu.Unlock()delete(p.subscribers, subscriber)
}func (p *Publisher) Publish(event Event) {p.mu.Lock()defer p.mu.Unlock()for sub := range p.subscribers {go sub(event)}
}func main() {pub := NewPublisher()subscriber1 := func(event Event) {fmt.Printf("Subscriber 1 received: %s\n", event)}subscriber2 := func(event Event) {fmt.Printf("Subscriber 2 received: %s\n", event)}pub.Subscribe(subscriber1)pub.Subscribe(subscriber2)pub.Publish("Event 1")pub.Publish("Event 2")pub.Unsubscribe(subscriber1)pub.Publish("Event 3")
}

总结

1. 屏障模式：使用通道和 sync.WaitGroup 同步多个 goroutine。
2. 未来模式：使用通道异步获取任务结果。
3. 管道模式：使用通道连接多个 goroutine 形成数据处理流水线。
4. 协程池模式：管理一组固定数量的 goroutine，重用这些 goroutine 处理任务。
5. 发布订阅模式：解耦消息的发送者和接收者，使用通道广播消息。