golang学习之go并发编程内置库

go

go 关键字用来开启一个协程

package main

import "time"

type myDate struct {
	count int
}

func (d *myDate) add() {
	d.count++
}

func (d *myDate) read() {
	println(d.count)
}

func main() {
	d := myDate{}
	for i := 0; i < 1000; i++ {
		// 开启1000个协程分别对count加1
		go d.add()
	}
	// 主协程等待3秒，让1000个子协程执行完
	time.Sleep(3 * time.Second)
	d.read()//打印结果986与预期不一致
}

sync.WaitGroup

让主协程睡眠等待有所不妥，可以使用sync.WaitGroup。sync.WaitGroup有三个方法： Add()设置等待值、 Done()等待值减一、 Wait()检查等待值是否为0，为0立即返回，否则阻塞当前协程。

package main

import (
	"sync"
)

type myDate struct {
	count int
}

// 传入指针类型否则将是一个新的wg
func (d *myDate) add(wg *sync.WaitGroup) {
	d.count++
	// 等待值减一
	wg.Done()
}

func (d *myDate) read() {
	println(d.count)
}

func main() {
	var wg sync.WaitGroup
	// 设置等待值1000
	wg.Add(1000)
	d := myDate{}
	for i := 0; i < 1000; i++ {
		// 开启1000个协程分别对count加1
		go d.add(&wg)
	}
	// 主协程检查等待值是否为0（1000个子协程执行完），如果是0则继续往下执行，否则阻塞当前协程
	wg.Wait()
	d.read() //打印结果986与预期不一致
}

sync.Mutex

互斥锁的零值表示未上锁。为防止出现异常现象上锁、解锁需成对出现。互斥锁的重复上锁操作会阻塞当前协程。互斥锁的重复解锁会出现恐慌。解锁一个未上锁的写锁会引起恐慌，解锁一个未上锁的读锁不会引起恐慌。

package main

import (
	"sync"
)

type myDate struct {
	count int
	sync.Mutex
}

func (d *myDate) add(wg *sync.WaitGroup) {
	d.Lock()         //上锁
	defer d.Unlock() //解锁
	d.count++
	wg.Done()
}

func (d *myDate) read() {
	d.Lock()         //上锁
	defer d.Unlock() //解锁
	println(d.count)
}

func main() {
	var wg sync.WaitGroup
	// 设置等待值1000
	wg.Add(1000)
	d := myDate{}
	for i := 0; i < 1000; i++ {
		// 开启1000个协程分别对count加1
		go d.add(&wg)
	}
	// 主协程检查等待值是否为0（1000个子协程执行完），如果是0则继续往下执行，否则阻塞当前协程
	wg.Wait()
	d.read() //打印1000
}

sync.RWMutex

读写锁可以分别对读、写操作进行上锁解锁。被读锁锁定的资源允许同时有多个读操作，但同时只能有一个写操作。

package main

import (
	"sync"
)

type myDate struct {
	count int
	sync.RWMutex
}

func (d *myDate) add(wg *sync.WaitGroup) {
	d.Lock()         //写操作上锁
	defer d.Unlock() //写操作解锁
	d.count++
	wg.Done()// 等待值减一
}

func (d *myDate) read(wg *sync.WaitGroup) {
	d.RLock()         //读操作上锁
	defer d.RUnlock() //读操作解锁
	println(d.count)
	wg.Done()// 等待值减一
}

func main() {
	var wg sync.WaitGroup
	// 设置等待值1010
	wg.Add(1010)
	d := myDate{}
	// 开启1000个协程分别对count加1
	for i := 0; i < 1000; i++ {
		go d.add(&wg)
	}
	for i := 0; i < 10; i++ {
		go d.read(&wg)
	}
	// 主协程检查等待值是否为0（1000个子协程执行完），如果是0则继续往下执行，否则阻塞当前协程
	wg.Wait()
}

sync/atomic

原子操作是由一个独立的CPU指令完成的，因此可以在并发环境下保证操作数据的安全。

package main

import (
	"sync"
	"sync/atomic"
)

type myDate struct {
	count int64
}

func (d *myDate) add(wg *sync.WaitGroup) {
	//d.count 原子加1
	atomic.AddInt64(&d.count, 1)
	// 等待值减一
	wg.Done()
}

func (d *myDate) read() {
	println(d.count)
}

func main() {
	var wg sync.WaitGroup
	wg.Add(1000)
	d := myDate{}
	for i := 0; i < 1000; i++ {
		go d.add(&wg)
	}
	wg.Wait()
	d.read()打印1000
}

sync.Once

sync.Once只有一个Do方法，该方法只会被执行一次

package main

import (
	"sync"
)

func main() {

	var one sync.Once
	var wg sync.WaitGroup
	count := 10
	wg.Add(count)

	for i := 0; i < 10; i++ {
		one.Do(func() {
			for j := 0; j < count; j++ {
				go func(i, j int) {
					// i值始终为0
					println(i, j)
					// 等待值减一
					wg.Done()
				}(i, j)
			}
		})
	}

	one.Do(func() {
		println("不打印")
	})
	wg.Wait()
}

sync.Pool

数据存储池，get获取的值是个不确定的值，有可能已被垃圾回收器回收。当获取不到数据时会调用New函数，如果没有New函数返回nil。

package main

import (
	"fmt"
	"sync"
)

func main() {

	var wg sync.WaitGroup
	wgcount := 10
	wg.Add(wgcount)

	var pool sync.Pool

	//向池中加入数据
	pool.Put(1)
	pool.Put(2)
	pool.Put(3)

	//返回并冲pool中删除数据
	fmt.Println(pool.Get())//1
	fmt.Println(pool.Get())//3
	fmt.Println(pool.Get())//2
	fmt.Println(pool.Get())//<nil>
	fmt.Println(pool.Get())//<nil>

	type user struct {
		Name string
		Age  int
	}

	u := user{Name: "tome", Age: 18}
	//向池中加入数据
	pool.Put(&u)
	//没有获取到数据时调用此函数
	pool.New = func() any {
		println("没有获取到数据时调用此函数")
		return &u
	}

	for i := 0; i < wgcount; i++ {
		go func() {
			a := pool.Get()
			if u1, ok := a.(*user); ok {
				u1.Age++
				pool.Put(&u1)
				fmt.Println(u1)
			}
			wg.Done()
		}()

	}
	wg.Wait()
	fmt.Println(u)
}

sync.Map

并发安全的map

package main

import (
	"fmt"
	"sync"
)

func main() {
	var pool sync.Map

	type user struct {
		Name string
		Age  int
	}

	//保存数据
	pool.Store("tom", user{Name: "tome", Age: 18})
	pool.Store("tim", user{Name: "tim", Age: 19})

	// 获取数据
	if u, ok := pool.Load("tom"); ok {
		fmt.Println(u)
	} else {
		println("数据不存在")
	}

	if u, ok := pool.Load("jim"); ok {
		fmt.Println(u)
	} else {
		println("数据不存在")
	}

	// 覆盖存量数据
	pool.Store("tim", user{Name: "tim", Age: 20})
	if u, ok := pool.Load("tim"); ok {
		fmt.Println(u)
	} else {
		println("数据不存在")
	}

	//删除数据
	pool.Delete("tim")
	if u, ok := pool.Load("tim"); ok {
		fmt.Println(u)
	} else {
		println("数据不存在")
	}

	actual, loaded := pool.LoadOrStore("timdd", user{Name: "tiddm", Age: 22})
	fmt.Println(actual, loaded)

	// 遍历pool
	pool.Range(func(key, value any) bool {
		fmt.Println(key, value)
		return false
	})

}

chan

chan通道，用于协程间的数据传送。 golang学习之chan

package main

import (
	"fmt"
)
// 生产者
func Producer(queue chan<- int) {
	for i := 0; i < 10; i++ {
		go func(i int) {
			queue <- i
		}(i)
	}
}

// 消费者
func Consumer(queue <-chan int) {
	for i := 0; i < 10; i++ {
		v := <-queue
		fmt.Println("消费", v)
	}
}
func main() {
	queue := make(chan int)
	Producer(queue)
	Consumer(queue)
}

sync.Cond

通过某个条件变量控制多个协程之间的协作运行。模拟实现sync.WaitGroup 主协程等待子协程完成工作：

package main
import (
	"fmt"
	"sync"
)

func main() {'
	cond := sync.NewCond(&sync.Mutex{})

	count := 10
	adder := 0

	for i := 0; i < count; i++ {
		go func(i int) {
			fmt.Println("子goroutine", i, "do something")
			cond.L.Lock()
			adder++
			cond.L.Unlock()
			cond.Broadcast()

		}(i)
	}

	cond.L.Lock()
	for adder != count {
		fmt.Println("主goroutine waiting")
		cond.Wait()
	}
	cond.L.Unlock()
	fmt.Println("done")
}

context

context可用于父子协程之间务完成情况同步信息。

package main

import (
	"context"
	"fmt"
	"time"
)

func main() {
	t := 2
	// 根context
	ctx := context.Background()
	// 生成一个超时context
	ctx, cancel := context.WithTimeout(ctx, time.Duration(t)*time.Second)
	defer cancel()

	go func() {
		// 模拟子协程工作3秒
		time.Sleep(3 * time.Second)
		cancel()
	}()

	select {
	case <-ctx.Done():// 任务已结束
		fmt.Println("testWTimeout.Done:", ctx.Err())// 任务结束原因
	case e := <-time.After(time.Duration(t) * time.Second):// 任务超时
		fmt.Println("testWTimeout:", e)
	}
}