Go语言常见的并发模式

Posted 2021-09-03

Go语言常见的并发模式

并发不是并行 并发关注的是程序的设计层面,并发的程序也可以顺序执行，在多核CPU上才可能真正同时的运行，并行关注的是程序的运行层面，如GPU中对图像处理都会有大量的并行运算。

Go语言将其并发编程哲学Slogan:不要通过共享内存来通信，而应通过通信来共享内存。

sync.Mutex / channel /sync.WaitGroup 实现同步

func main() {
	var (
		mu sync.Mutex
		i  = 0
	)
	mu.Lock()
	go func() {
		for x := 0; x < 10; x++ {
			i++
		}
	}()
	mu.Unlock()
}


func main() {
	done := make(chan int, 1)

	go func() {
		fmt.Println("hello world")
		done <- 1
	}()

	<-done
}

对于带缓冲的Channel，对于Channel的第K个接收完成操作发生在第K+C个发送操作完成之前，其中C是Channel的缓存大小。虽然管道是带缓存的


func main() {
	done := make(chan int, 10) //缓存为 0

	for i := 0; i < cap(done); i++ {
		go func(x int) {
			fmt.Println("x = ", x)
			done <- 1
		}(i)
	}

	//等待 10 个goroutine
	for i := 0; i < cap(done); i++ {
		<-done
	}
}

func main() {
	var wg sync.WaitGroup

	for i := 0; i < 10; i++ {
		wg.Add(1) //在开启goroutine 前

		go func(x int) {
			fmt.Println("x = ", x)
			wg.Done()
		}(i)
	}

	wg.Wait()
}

生产者消费者模型

生产消费是异步的两个过程，生产者生产数据，然后放到成果队列中，同时消费者从成果队列中来消费这些数据，当成果队列中没有数据时，消费者就进入饥饿的等待中；而当成果队列中数据已满时，生产者则需要等待消费者消费数据。

// 生产者
func Producer(out chan<- int) {
	for i := 0; ; i++ {
		out <- i * i
	}
}

// 消费者
func Consumer(in <-chan int) {
	for v := range in {
		fmt.Println(v)
	}
}
func main() {
	ch := make(chan int, 10)
	go Producer(ch)
	go Consumer(ch)
	time.Sleep(2 * time.Second)
}

发布订阅模型

发布订阅模型通常被简写为pub/sub模型。在这个模型中，消息生产者成为发布者（publisher），而消息消费者则成为订阅者（subscriber），生产者和消费者是M:N的关系。在传统生产者和消费者模型中，是将消息发送到一个队列中，而发布订阅模型则是将消息发布给一个主题

控制并发数

可以通过带缓存Channel的使用量和最大容量比例来判断程序运行的并发率

func main() {
	control := make(chan interface{}, 3)
	done := make(chan struct{}, 10)
	for i := 1; i <= 10; i++ {
		control <- i
		go func(j int) {
			fmt.Printf("i: %d, time: %d\\n", j, time.Now().Unix())
			time.Sleep(time.Second)
			<-control

			done <- struct{}{}
		}(i)

	}
	for i := 1; i <= 10; i++ {
		<-done

	}
}

并发的安全退出

Go语言中不同Goroutine之间主要依靠管道进行通信和同步。要同时处理多个管道的发送或接收操作，我们需要使用select关键字。当select有多个分支时，会随机选择一个可用的管道分支，如果没有可用的管道分支则选择default分支，否则会一直保存阻塞状态

select {
case v := <-in:
    fmt.Println(v)
case <-time.After(time.Second):
	return // 超时
default:
    return // 没有数据
}

context

标准库context包，用来简化对于处理单个请求的多个Goroutine之间与请求域的数据、超时和退出等操作

func main() {
	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	var wg sync.WaitGroup
	for i := 0; i < 10; i++ {
		wg.Add(1)
		go worker(ctx, &wg)
	}

	time.Sleep(time.Second)
	cancel()

	wg.Wait()
}

func worker(ctx context.Context, wg *sync.WaitGroup) error {
	defer wg.Done()
	for {
		select {
		default:
			fmt.Println("hello")
		case <-ctx.Done():
			fmt.Println("ctx done")
			return ctx.Err()
		}
	}
}