并发与并行
并发和并行是有区别的，并发不等于并行。
两个或多个事件在同一时间不同时间间隔发生。对应在Go中，就是指多个 goroutine 在单个CPU上的交替运行。
两个或者多个事件在同一时刻发生。对应在Go中，就是指多个 goroutine 在多个CPU上同时运行。
goroutine
goroutine 是 Go 中一种轻量级线程。也称为用户态线程。由 Go 的 runtime 进行管理。Go 的程序会智能地将 goroutine 中的任务合理地分配给每个 CPU。
在程序中，我们只要使用 go 关键字，就可以轻易开启一个 goroutine
在使用 goroutine 时，以下两个建议可以有效避免 goroutine 泄露。
来看一个泄露的例子
func leak() {
ch := make(chan int)
go func() {
<-ch leak 函数阻塞在接受 ch fmt.println("receive a value") }() } func main() { leak(ch) 函数返回, < code></-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) errors.new("time out") } func main() { worker :="&Worker{" wg: sync.waitgroup{}, ctx, cancel time.now().add(1000*time.millisecond)) defer cancel() worker.do() if err !="nil" fmt.println(err) < code></-ch:>
有一个 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 在大量并发的情况下，会造成锁等待，对性能的影响比较大。在读多写少的场景下可以使用读写锁。读写锁主要遵循以下原则：
下面是一个互斥锁简单示例。在需要访问共享资源的地方使用 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() srv.start(newcontext(a.opts.ctx, a)) } wg.wait() if a.opts.registrar !="nil" rctx, rcancel a.opts.registrartimeout) rcancel() err instance); a.lk.lock() a.instance="instance" a.lk.unlock() c os.signal, 1) signal.notify(c, a.opts.sigs...) select case <-ctx.done(): ctx.err() <-c: a.opts.logger.errorf("failed to app: %v", err) && !errors.is(err, context.canceled) nil < code></-ctx.done()>
channels
channel 是 Go 语言中一种类型安全的消息队列，充当两个 goroutine 之间的通道，通过它可以进行任意资源的的交换。同时通过 channel 实现 Go 的同步机制。
当创建的 channel 没有缓冲时，称为无缓冲通道。无缓冲管道必须读写同时操作才会有效果，如果只进行读或者只进行写那么会被阻塞，等待另外一方的操作。
创建的 channel 具有缓冲时，称为缓冲通道。缓冲通道是固定容量的先进先出（FIFO）队列。容量在队列创建的时候就已经固定，运行是无法更改。消费者从队列中取出元素并处理它们。如果队列为空并且消费者无事可做，就会发生阻塞，直到生产者放入一个元素。如果队列已满，并且消费者未开始消费，则会发生阻塞，知道消费者消费一个元素。
不论是无缓冲通道还是缓冲通道，都不能往一个已关闭的 channel 发送消息，否则程序会直接 panic ,因此，最好是由发送端进行关闭 channel。
func main() {
ch := make(chan int)
close(ch)
fmt.Println(<-ch) 0 1 2 3 close(ch) panic: close of closed channel ch <- send on chs :="make(chan" int, 2) close(chs) fmt.println(<-chs) } < code></-ch)>
Original: https://www.cnblogs.com/arvinhuang/p/16437905.html
Author: 平凡键客
Title: Golang并发编程——goroutine、channel、sync
原创文章受到原创版权保护。转载请注明出处：https://www.johngo689.com/684008/
转载文章受原作者版权保护。转载请注明原作者出处！