命令记不住

sync 七种使用方式

• sync 包有以下几个内容：

sync.Pool 		临时对象池
sync.Mutex 		互斥锁
sync.RWMutex 		读写互斥锁
sync.WaitGroup 	        组等待
sync.Cond 		条件等待
sync.Once 		单次执行
sync.Map 		并发安全字典结构

1.临时对象池 (sync.Pool)

Pool 是用于存储那些被分配了但是没有被使用，而未来可能会使用的值，以减小垃圾回收的压力。

Pool 是协程安全的，应该用于管理协程共享的变量，不推荐用于非协程间的对象管理。

调动 New 函数，将使用函数创建一个新对象返回。

从 Pool 中取出对象时，如果 Pool 中没有对象，将执行 New()，如果没有对 New 进行赋值，则返回 Nil。

先进后出存储原则，和栈类似 Pool 一个比较好的例子是 fmt 包，fmt 包总是需要使用一些[]byte 之类的对象，Go 建立了一个临时对象池，存放着这些对象，如果需要使用一个[]byte，就去 Pool 里面拿，如果拿不到就分配一份。这比起不停生成新的 []byte，用完了再等待 gc 回收来要高效得多。

package main

import (
	"fmt"
	"sync"
)

// 定义函数值给pool.New
func Demo() interface{} {
	return 3.14
}
func main() {
	var pool sync.Pool
	//Put 将数据存储在临时对象池
	pool.Put(1)
	pool.Put("hello")
	pool.Put(true)
	pool.Put(3.14)

	//Get 将对象从临时对象池取出
	value := pool.Get()
	fmt.Println(value)
	//临时对象池第一个数据在最前，后续的数据采用先进后出规则
	value = pool.Get()
	fmt.Println(value)
	value = pool.Get()
	fmt.Println(value)
	value = pool.Get()
	fmt.Println(value)
	//循环获取所有对象内容
	for {
		value := pool.Get()
		if value == nil {
			break
		}
		fmt.Println(value)
	}
	//需要赋值函数类型变量
	pool.New = Demo
	//当Get返回nil时，New可选地指定一个函数来生成一个值。不能在调用Get时同时更改它
}

• 资源池实例

利用 goroutine 和 channel 进行 go 的并发模式，实现一个资源池实例，资源池可以存储一定数量的资源，用户程序从资源池获取资源进行使用，使用完成将资源释放回资源池。

• 安全的资源池

//一个安全的资源池，被管理的资源必须都实现io.Close接口
type Pool struct {
	m       sync.Mutex
	res     chan io.Closer
	factory func() (io.Closer, error)
	closed  bool
}

这个结构体 Pool 有四个字段

m 是一个互斥锁，这主要是用来保证在多个 goroutine 访问资源时，池内的值是安全的

res 字段是一个有缓冲的通道，用来保存共享的资源，这个通道的大小，在初始化 Pool 的时候就指定的。注意这个通道的类型是 io.Closer 接口，所以实现了这个 io.Closer 接口的类型都可以作为资源，交给我们的资源池管理。

factory 这个是一个函数类型，它的作用就是当需要一个新的资源时，可以通过这个函数创建，也就是说它是生成新资源的，至于如何生成、生成什么资源，是由使用者决定的，所以这也是这个资源池灵活的设计的地方。

closed 字 段表示资源池是否被关闭，如果被关闭的话，再访问是会有错误的。

• 资源池 pool.go 和 main.go

// 代码可以分为两个文件pool.go和main.go
package main

import (
	"errors"
	"fmt"
	"io"
	"math/rand"
	"sync"
	"sync/atomic"
	"time"
)

// pool.go
type Pool struct {
	m   sync.Mutex
	res chan io.Closer
	//创建资源的方法，由用户程序自己生成传入
	factory func() (io.Closer, error)
	closed  bool
	//资源池获取资源超时时间
	timeout <-chan time.Time
}

// 资源池关闭标志
var ErrPoolClosed = errors.New("资源池已经关闭")

// 超时标志
var ErrTimeout = errors.New("获取资源超时")

// 新建资源池
func New(fn func() (io.Closer, error), size int) (*Pool, error) {
	if size <= 0 {
		return nil, errors.New("新建资源池大小太小")
	}
	//新建资源池
	p := Pool{
		factory: fn,
		res:     make(chan io.Closer, size),
	}
	//向资源池循环添加资源，直到池满
	for count := 1; count <= cap(p.res); count++ {
		r, err := fn()
		if err != nil {
			fmt.Println("添加资源失败，创建资源方法返回nil")
			break
		}
		fmt.Println("资源加入资源池")
		p.res <- r
	}
	fmt.Println("资源池已满，返回资源池")
	return &p, nil
}

// 获取资源
func (p *Pool) Acquire(d time.Duration) (io.Closer, error) {
	//设置d时间后超时
	p.timeout = time.After(d)
	select {
	case r, ok := <-p.res:
		fmt.Println("获取", "共享资源")
		if !ok {
			return nil, ErrPoolClosed
		}
		return r, nil
	case <-p.timeout:
		return nil, ErrTimeout
	}
}

// 放回资源池
func (p *Pool) Release(r io.Closer) {
	//上互斥锁，和Close方法对应，不同时操作
	p.m.Lock()
	defer p.m.Unlock()

	if p.closed {
		r.Close()
		return
	}
	//资源放回队列
	select {
	case p.res <- r:
		fmt.Println("资源池放回队列")
	default:
		fmt.Println("资源队列已满，释放资源")
		r.Close()
	}
}

// 关闭资源池
func (p *Pool) Close() {
	//互斥锁，保证同步和Release方法相关，用同一把锁
	p.m.Lock()
	defer p.m.Unlock()

	if p.closed {
		return
	}

	p.closed = true
	//清空通道资源之前，将通道关闭，否则引起死锁
	close(p.res)
	for r := range p.res {
		r.Close()
	}
}

// main.go
const (
	maxGoroutines   = 25
	pooledResources = 2
)

// 实现接口类型 资源类型
type dbConnection struct {
	ID int32
}

// 实现接口方法
func (conn *dbConnection) Close() error {
	fmt.Printf("资源关闭,ID:%d\n", conn.ID)
	return nil
}

// 给每个连接资源给id
var idCounter int32

// 创建新资源
func createConnection() (io.Closer, error) {
	id := atomic.AddInt32(&idCounter, 1)
	fmt.Printf("创建新资源,id:%d\n", id)
	return &dbConnection{ID: id}, nil
}

// 测试资源池
func performQueries(query int, p *Pool) {
	conn, err := p.Acquire(10 * time.Second)
	if err != nil {
		fmt.Println("获取资源超时")
		fmt.Println(err)
		return
	}
	//方法结束后将资源放进资源池
	defer p.Release(conn)
	//模拟使用资源
	time.Sleep(time.Duration(rand.Intn(1000)) * time.Millisecond)
	fmt.Printf("查询goroutine id:%d,资源ID：%d\n", query, conn.(*dbConnection).ID)
}
func main() {
	var wg sync.WaitGroup
	wg.Add(maxGoroutines)
	p, err := New(createConnection, pooledResources)
	if err != nil {
		fmt.Println(err)
	}
	//每个goroutine一个查询，每个查询从资源池中获取资源
	for query := 0; query < maxGoroutines; query++ {
		go func(q int) {
			performQueries(q, p)
			wg.Done()
		}(query)
	}
	//主线程等待
	wg.Wait()
	fmt.Println("程序结束")
	//释放资源
	p.Close()
}

2.互斥锁 (sync.Mutex)

互斥锁用来保证在任一时刻，只能有一个协程访问某对象。

Mutex 的初始值为解锁状态，Mutex 通常作为其它结构体的匿名字段使用，使该结构体具有 Lock 和 Unlock 方法。

Mutex 可以安全的再多个协程中并行使用。

如果对未加锁的进行解锁，则会引发 panic.

• 加锁保证数据能够正确

package main

import (
	"fmt"
	"sync"
)

var num int
var mu sync.Mutex
var wg sync.WaitGroup

func main() {
	wg.Add(10000)
	for i := 0; i < 10000; i++ {
		go Add()
	}
	wg.Wait()
	fmt.Println(num)
}
func Add() {
	mu.Lock() // 加锁
	defer func() {
		mu.Unlock() // 解锁
		wg.Done()
	}()
	num++
}

并发没有加锁时，会发现结果不等于 10000，其原因是因为出现这样的情况，就是其中一些协程刚好读取了 num 的值，此时该协程刚好时间片结束，被挂起，没有完成加 1

然后其他协程进行加 1，然后当调度回到原来那个协程，num = 原来的那个 num(而不是最新的 num) + 1，导致 num 数据出错。

package main

import (
	"fmt"
	"sync"
)

//不建议全局变量在协程中使用，如果使用需要加锁

var num = 0
var wg sync.WaitGroup
var mu sync.Mutex

func Count() {
	mu.Lock()
	defer mu.Unlock()
	num++
	wg.Done()
}

func main() {
	wg.Add(10000)
	for i := 0; i < 10000; i++ {
		go Count()
	}
	wg.Wait()
	fmt.Println(num)
}

3.读写锁(sync.RWMutex)

• RWMutex比Mutex多了一个“写锁定” 和 “读锁定”，可以让多个协程同时读取某对象。

• RWMutex的初始值为解锁状态。RWMutex通常作为其它结构体的匿名字段使用。
• RWMutex可以安全的在多个协程中并行使用。

// Lock 将 rw 设置为写状态，禁止其他协程读取或写入
func (rw *RWMutex) Lock()

// Unlock 解除 rw 的写锁定状态，如果rw未被锁定，则该操作会引发 panic。
func (rw *RWMutex) Unlock()

// RLock 将 rw 设置为锁定状态，禁止其他协程写入，但可以读取。
func (rw *RWMutex) RLock()

// Runlock 解除 rw 设置为读锁定状态，如果rw未被锁定，则该操作会引发 panic。
func (rw *RWMutex) RUnLock()

// RLocker 返回一个互斥锁，将 rw.RLock 和 rw.RUnlock 封装成一个 Locker 接口。
func (rw *RWMutex) RLocker() Locker

允许多读和单写示例：

package main

import (
	"fmt"
	"strconv"
	"sync"
	"time"
)

var rwLock sync.RWMutex

func main() {

	// 获取读锁
	for i := 0; i < 5; i++ {
		go func(i int) {
			rwLock.RLocker()
			defer rwLock.RLocker()
			fmt.Println("read func " + strconv.Itoa(i) + " get rlock at " + time.Now().String())
			time.Sleep(time.Second)
		}(i)
	}

	time.Sleep(time.Second / 10)
	// 获取写锁
	for i := 0; i < 5; i++ {
		go func(i int) {
			rwLock.Lock()
			defer rwLock.Unlock()
			fmt.Println("write func " + strconv.Itoa(i) + " get wlock at " + time.Now().String())
			time.Sleep(time.Second)
		}(i)
	}

	// 保证所有的 goroutine 执行结束
	time.Sleep(time.Second * 10)
}

.

4.等待组(sync.WaitGroup)

WaitGroup用于等待一组线程的结束。父线程调用Add方法来设定应等待的线程的数量，每一次执行Add都会增加线程组的数量。每个被等待的线程在结束时应调用Done方法。同时，主线程里可以调用Wait方法阻塞至所有线程结束。

如果你有多个goroutine并发执行某项操作，但是下方的代码需要等待这些并发的操作执行完成之后才能够执行，那么你可以使用sync.WaitGroup，因为它比Sleep更加有效，并且稳定

• WaitGroup 用于等待一组协程的结束。

• 主协程创建每个子协程的时候先调用Add增加等待计数，每个子协程在结束时调用Done减少协程计数。
• 主协程通过 Wait 方法开始等待，直到计数器归零才继续执行。

func (*WaitGroup) Add(delta int)

Add方法向内部计数加上delta，delta可以是负数；如果内部计数器变为0，Wait方法阻塞等待的所有线程都会释放，如果计数器小于0，方法panic。注意Add加上正数的调用应在Wait之前，否则Wait可能只会等待很少的线程。一般来说本方法应在创建新的线程或者其他应等待的事件之前调用。

func (*WaitGroup) Done

Done方法减少WaitGroup计数器的值，应在线程的最后执行，Done的执行应标志着一个goroutine的结束

func (*WaitGroup) Wait

Wait方法阻塞直到WaitGroup计数器减为0。如果WaitGroup不为0，那么程序就会一直阻塞在Wait函数这里

// 计数器增加 delta，delte可以负数
func (wg *WaitGroup) Add(delta int)

// 计数器减少1，等价于Add(-1)
func (wg *WaitGroup) Done()

// 等待直到计数器归零。如果计数器小于0，则该操作会引发 panic。
func (wg *WaitGroup) Wait()

func (wg *WaitGroup) Add(delta int):

• Add将可能为负的delta添加到等待组计数器。如果计数器变成零，所有goroutines阻塞的等待被释放。如果计数器变成负数，就增加恐慌。
• 注意，当计数器为0时发生的增量为正的调用必须在等待之前发生。任何时候都可能发生具有负增量的调用，或在计数器大于0时开始具有正增量的调用。通常，这意味着要添加的调用应该在创建goroutine语句或等待的其他事件之前执行。如果一个等待组被重用来等待几个独立的事件集，那么新的添加调用必须在所有之前的等待调用都返回之后发生

等待组和锁示例代码:

package main

import (
	"fmt"
	"sync"
)

func main() {
	var m sync.Mutex
	wg := sync.WaitGroup{}
	count := 0

	for i := 0; i < 100; i++ {
		wg.Add(1)
		go func() {
			m.Lock()        //加锁
			defer wg.Done() //完成
			count++         //模拟数据处理
			m.Unlock()
		}()
	}

	wg.Wait()
	fmt.Println(count)
	fmt.Println("程序继续执行")
}

• 计算后台运行时间

package main

import (
	"fmt"
	"sync"
	"time"
)

// 计算后台运行时间
func dosomething(millisecs time.Duration, wg *sync.WaitGroup) {
	duration := millisecs * time.Millisecond
	time.Sleep(duration)
	fmt.Println("Function in background, duration:", duration)
	wg.Done()
}

func main() {
	var wg sync.WaitGroup // 这里为什么不需要初始化
	wg.Add(4)
	go dosomething(200, &wg)
	go dosomething(1400, &wg)
	go dosomething(150, &wg)
	go dosomething(11600, &wg)

	wg.Wait()
	fmt.Println("Done")
}

5.条件等待(sync.Cond)

条件等待和互斥锁有不同，互斥锁是不同协程公用一个锁，条件等待是不同协程各用一个锁，但是wait()方法调用会等待（阻塞），直到有信号发过来，不同协程是共用信号。

• 信号接收示例代码

package main

import (
	"fmt"
	"sync"
	"time"
)

func main() {
	wg := sync.WaitGroup{}
	cond := sync.NewCond(new(sync.Mutex))

	for i := 0; i < 3; i++ {
		go func(i int) {
			wg.Add(1)
			defer wg.Done()
			fmt.Println("协程启动：", i)
			cond.L.Lock() //加锁
			fmt.Println("协程：", i, "加锁")
			cond.Wait()
			cond.L.Unlock() //解锁
			fmt.Println("协程：", i, "解锁")
		}(i)

	}
	time.Sleep(time.Second * 2)
	//发送信号
	cond.Signal()
	fmt.Println("主协程发送信号")

	time.Sleep(time.Second * 2)
	//发送信号
	cond.Signal()
	fmt.Println("主协程发送信号")

	time.Sleep(time.Second * 2)
	//发送信号
	cond.Signal()
	fmt.Println("主协程发送信号")

	wg.Wait()
}

C:\Users\vSphere\go\src\day05>go run 04.go
协程启动： 2
协程： 2 加锁
协程启动： 0
协程： 0 加锁
协程启动： 1
协程： 1 加锁
主协程发送信号
协程： 2 解锁
主协程发送信号
协程： 0 解锁
主协程发送信号
协程： 1 解锁

6.单次执行(sync.Once)

• Once的作用是多次调用但只执行一次，Once只有一个方法，Once.Do()，向Do传入一个函数，这个函数在第一次执行Once.Do()的时候会被调用
• 以后再执行Once.Do()将没有任何动作，即使传入了其他的函数，也不会被执行，如果要执行其它函数，需要重新创建一个Once对象
• Once可以安全的再多个协程中并行使用，是协程安全的

// 多次调用仅执行一次指定的函数f
func (o *Once) Do(f func())

• 示例代码

// 示例：Once
package main

import (
	"fmt"
	"sync"
)

func main() {
	var once sync.Once
	var wg sync.WaitGroup
	onceFunc := func() {
		fmt.Println("只执行一次")
	}

	for i := 0; i < 10; i++ {
		go func() {
			wg.Add(1)
			defer wg.Done()
			once.Do(onceFunc)
		}()
	}
	wg.Wait()
}

7.并发安全Map(sync.Map)

Go语言中的 map 在并发情况下，只读是线程安全的，同时读写是线程不安全的.

需要并发读写时，一般的做法是加锁，但这样性能并不高，Go语言在 1.9 版本中提供了一种效率较高的并发安全的 sync.Map，sync.Map 和 map 不同，不是以语言原生形态提供，而是在 sync包下的特殊结构

package main

import (
	"fmt"
	"sync"
)

func main() {
	var scene sync.Map
	// 将键值对保存到sync.Map
	scene.Store("x86", 97)
	scene.Store("cloud", 100)
	scene.Store("com", 200)
	// 从sync.Map中根据键取值
	fmt.Println(scene.Load("x86"))
	// 根据键删除对应的键值对
	scene.Delete("x86")
	// 遍历所有sync.Map中的键值对
	scene.Range(func(k, v interface{}) bool {
		fmt.Println("iterate:", k, v)
		return true
	})
}

• sync.Map 有以下特性：

无须初始化，直接声明即可.

sync.Map 不能使用 map 的方式进行取值和设置等操作，而是使用 sync.Map 的方法进行调用，Store 表示存储，Load 表示获取，Delete 表示删除.

使用 Range 配合一个回调函数进行遍历操作，通过回调函数返回内部遍历出来的值，Range参数中回调函数的返回值在需要继续迭代遍历时，返回 true，终止迭代遍历时，返回false.

// 安全字典结构
package main

import (
	"fmt"
	"sync"
)

func Print(key, value interface{}) bool {
	fmt.Println("键", key, "值", value)
	return true
}

//遍历数据（函数回调 将函数作为函数参数）
//smap.Range(Print)

func main() {
	/*err map必须初始化才可以使用make
	var m map[interface{}]interface{}
	m["x86"] = "cloud"
	fmt.Println(m)
	C:\Users\vSphere\go\src\day05>go run 07.go
	panic: assignment to entry in nil map

	goroutine 1 [running]:
	main.main()
	        C:/Users/vSphere/go/src/day05/07.go:12 +0x63
	exit status 2
	*/

	var smap sync.Map
	//将键值对存储在sync.map
	smap.Store("x86", 2018)
	smap.Store("cloud", 2019)
	smap.Store("com.cn", 2020)

	//从map中获取数据
	value, ok := smap.Load("cloud")
	if ok {
		fmt.Println(value)
	} else {
		fmt.Println("未找到数据")
	}

	//从map中删除数据
	smap.Delete("x86")

	//遍历数据（函数回调 将函数作为函数参数）
	smap.Range(Print)
}

使用goroutine并发向sync.Map中填充数据，，每一个goroutine在填充完数据后会使用sync.WaitGroup通知goroutine数据已经填充结束

package main

import (
	"fmt"
	"strconv"
	"sync"
)

var syncMap sync.Map
var waitGroup sync.WaitGroup

func main() {

	routineSize := 5
	// 让主线程等待数据添加完毕
	waitGroup.Add(routineSize)
	// 并发添加数据
	for i := 0; i < routineSize; i++ {
		go addNumber(i * 10)
	}

	// 开始等待
	waitGroup.Wait()
	var size int
	// 统计数量
	syncMap.Range(func(key, value interface{}) bool {
		size++
		//fmt.Println("key-value pair is", key, value, " ")
		return true
	})
	fmt.Println("syncMap current size is " + strconv.Itoa(size))
	// 获取键为 0 的值
	value, ok := syncMap.Load(0)
	if ok {
		fmt.Println("key 0 has value", value, " ")
	}

}

func addNumber(begin int) {
	// 往 syncMap 中放入数据
	for i := begin; i < begin+3; i++ {
		syncMap.Store(i, i)
	}
	// 通知数据已添加完毕
	waitGroup.Done()
}

.