Golang并发编程——goroutine、channel、sync

并发与并行

并发和并行是有区别的，并发不等于并行。

并发

两个或多个事件在同一时间不同时间间隔发生。对应在Go中，就是指多个 goroutine 在单个CPU上的交替运行。

并行

两个或者多个事件在同一时刻发生。对应在Go中，就是指多个 goroutine 在多个CPU上同时运行。

goroutine

介绍

goroutine 是 Go 中一种轻量级线程。也称为用户态线程。由 Go 的 runtime 进行管理。Go 的程序会智能地将 goroutine 中的任务合理地分配给每个 CPU。

在程序中，我们只要使用 go 关键字，就可以轻易开启一个 goroutine

建议

在使用 goroutine 时，以下两个建议可以有效避免 goroutine 泄露。

1. 调用者清楚 goroutine 什么时候结束
2. 调用者可以控制 goroutine 的生命周期

来看一个泄露的例子

func leak() {         ch := make(chan int) 	go func() { 		<-ch//leak 函数阻塞在接受 ch  		fmt.Println("receive a value") 	}() } func main() { 	leak(ch)//函数返回, }  

这个channel将无法被关闭，leak 函数里开启的 goroutine 也永远无法返回，当然，这个例子中 leak 函数返回了，main 函数结束，leak 函数里开启的 goroutine 也就返回了。

1.调用者不清楚什么时候结束，也无法控制 goroutine 的生命周期。只能被动等待 channel 接受信号，然后执行函数逻辑，如你所见，造成的后果便是容易产生 goroutine 泄露。

来看下面一个例子

type Worker struct { 	wg sync.WaitGroup }  func (w *Worker) Do() { 	w.wg.Add(1)  	go func() { 		defer w.wg.Done() 		//do someting 		time.Sleep(800 * time.Millisecond) 		fmt.Println("finish") 	}()  }  func (w *Worker) Shutdown(ctx context.Context) error { 	ch := make(chan struct{}) 	go func() { 		w.wg.Wait() 		close(ch) 	}()  	select { 	case <-ch: 		return nil 	case <-ctx.Done(): 		// time out 		// close(ch) 		return errors.New("time out") 	} }  func main() { 	worker := &Worker{ 		wg: sync.WaitGroup{}, 	} 	ctx, cancel := context.WithDeadline(context.Background(), time.Now().Add(1000*time.Millisecond)) 	defer cancel() 	worker.Do() 	if err := worker.Shutdown(ctx); err != nil { 		fmt.Println(err) 	} }  

有一个 worker 对象，这个对象会做一些耗时操作。我们在 Do() 方法中使用 goroutine 来处理具体逻辑，在开启goroutine 之前调用 wg.Add(1), 然后在 goroutine 的 defer 函数中 调用 wg.Done(),在 Shutdown() 方法中使用 wg.Wait() 来等待 Do() 方法执行结束。在 Shutdown() 方法中，如果 goroutine 执行结束了，就会往 ch channel 中发送消息，底下 select {} 中收到 ch channel 消息后，Shutdown 方法就可以正常返回，函数到此执行结束。如果 Do() 方法执行太长超出了 ctx 的最长时间。Shutdown 会返回 "time out" 异常。返回之前可以进行资源的处理。

在这个例子中调用者可以通过控制上下文控制来控制 Worker 对象的生命周期。

sync.Mutex、_s_ync.RWMutex

Go 的 sync 包提供了 mutex、RwMutex,分别是互斥锁与读写锁。

在需要共享内存的地方，如果有多个对象同时对这个地方进行读写操作，就会产生竞态条件。我们需要使用程序语言提供的同步原语对读写操作进行保护。互斥锁就是同一时刻一段代码只能被一个线程/协程运行。Mutex 在大量并发的情况下，会造成锁等待，对性能的影响比较大。在读多写少的场景下可以使用读写锁。读写锁主要遵循以下原则：

1. 读写锁的读锁可以重入，在已经有读锁的情况下，可以继续加读锁。
2. 在读锁没有全部解锁时，写操作会阻塞直到所有读锁解锁。
3. 在写锁没有解锁时，其他协程的读写操作都会被阻塞，直到写锁解锁。

下面是一个互斥锁简单示例。在需要访问共享资源的地方使用 Lock 和 Unlock 方法。表示这部分操作属于“原子操作”。使用时需要注意锁粒度。我们要尽可能的减小锁粒度。锁粒度小了，锁竞争就少。对程序的性能影响就小。

var l sync.Mutex var a string  func f() { 	a = "hello, world" 	l.Unlock() }  func main() { 	l.Lock() 	go f() 	l.Lock() 	print(a) }  

sync/atomic

sync/atomic 提供了用于实现同步算法的底层原子内存原语

copy-on-write 思路在微服务降级或者 local cache 经常使用。我们可以使用 atomic 来实现。atmic 依赖于原子 CPU 指令而不是依赖外部锁，性能不俗。

type NumberArray struct { 	array []int }  func main() { 	var atomic atomic.Value  	go func() { 		var i int 		for { 			i++ 			numArray := &NumberArray{ 				array: []int{i, i + 1, i + 2, i + 3}, 			} 			atomic.Store(numArray) 			time.Sleep(100 * time.Millisecond) 		} 	}()  	time.Sleep(500 * time.Millisecond) //先让数据更新  	var wg sync.WaitGroup 	for n := 0; n < 100000; n++ { 		wg.Add(1) 		time.Sleep(100 * time.Millisecond) 		go func() { 			numArray := atomic.Load() 			fmt.Println(numArray) 			wg.Done() 		}() 	} 	wg.Wait() }  

errgroup

errgroup 为处理公共任务的子任务的 goroutine 组提供同步、错误传播和上下文取消。

https://github.com/go-kratos/kratos/blob/main/app.go  func (a *App) Run() error { 	instance, err := a.buildInstance() 	if err != nil { 		return err 	} 	eg, ctx := errgroup.WithContext(NewContext(a.ctx, a)) 	wg := sync.WaitGroup{} 	for _, srv := range a.opts.servers { 		srv := srv 		eg.Go(func() error { 			<-ctx.Done() // wait for stop signal 			stopCtx, cancel := context.WithTimeout(NewContext(a.opts.ctx, a), a.opts.stopTimeout) 			defer cancel() 			return srv.Stop(stopCtx) 		}) 		wg.Add(1) 		eg.Go(func() error { 			wg.Done() 			return srv.Start(NewContext(a.opts.ctx, a)) 		}) 	} 	wg.Wait() 	if a.opts.registrar != nil { 		rctx, rcancel := context.WithTimeout(ctx, a.opts.registrarTimeout) 		defer rcancel() 		if err := a.opts.registrar.Register(rctx, instance); err != nil { 			return err 		} 		a.lk.Lock() 		a.instance = instance 		a.lk.Unlock() 	} 	c := make(chan os.Signal, 1) 	signal.Notify(c, a.opts.sigs...) 	eg.Go(func() error { 		for { 			select { 			case <-ctx.Done(): 				return ctx.Err() 			case <-c: 				if err := a.Stop(); err != nil { 					a.opts.logger.Errorf("failed to stop app: %v", err) 					return err 				} 			} 		} 	}) 	if err := eg.Wait(); err != nil && !errors.Is(err, context.Canceled) { 		return err 	} 	return nil }  

channels

channel 是 Go 语言中一种类型安全的消息队列，充当两个 goroutine 之间的通道，通过它可以进行任意资源的的交换。同时通过 channel 实现 Go 的同步机制。

无缓冲通道

当创建的 channel 没有缓冲时，称为无缓冲通道。无缓冲管道必须读写同时操作才会有效果，如果只进行读或者只进行写那么会被阻塞，等待另外一方的操作。

缓冲通道

创建的 channel 具有缓冲时，称为缓冲通道。缓冲通道是固定容量的先进先出（FIFO）队列。容量在队列创建的时候就已经固定，运行是无法更改。消费者从队列中取出元素并处理它们。如果队列为空并且消费者无事可做，就会发生阻塞，直到生产者放入一个元素。如果队列已满，并且消费者未开始消费，则会发生阻塞，知道消费者消费一个元素。

不论是无缓冲通道还是缓冲通道，都不能往一个已关闭的 channel 发送消息，否则程序会直接 panic ,因此，最好是由发送端进行关闭 channel。

func main() { 	ch := make(chan int) 	close(ch) 	fmt.Println(<-ch)//0 	//close(ch)  //panic: close of closed channel 	//ch <- 2  //panic: send on closed channel  	chs := make(chan int, 2) 	chs <- 1 	chs <- 3 	close(chs) 	fmt.Println(<-chs) 	fmt.Println(<-chs) 	fmt.Println(<-chs)//0 	// chs <- 2 //panic: send on closed channel }  

关于channel 还可以查看这篇文章 polarisxu：无缓冲和有缓冲通道