intitle:网站建设,企业微信网站怎么做,做网站公司排行,自适应科技公司网站模板make 和 new 的区别
make 和 new 都是用来分配内存
make 只能对 slice map channel 进行初始化结构体实例。new 可以对任意类型进行初始化make 用于分配数据对象的具体实例#xff0c;new 用于分配数据类型的默认值#xff0c;并返回该数据的指针。
new 出来的 slice 、ma…make 和 new 的区别
make 和 new 都是用来分配内存
make 只能对 slice map channel 进行初始化结构体实例。new 可以对任意类型进行初始化make 用于分配数据对象的具体实例new 用于分配数据类型的默认值并返回该数据的指针。
new 出来的 slice 、map、channel 是无法使用的是一个 nil 值。这三个值是不会分配地址空间的因为其默认值就是nil。而map 会返回非nil 值。
Slice
slice 切片具体概念如下
data 存了什么元素len 存了多少个元素cap 总共可以存多少个元素 cap 指的是底层数组的容量也就是从slice 的 第一个元素开始底层数组能够容纳的元素个数。slice 的 cap 决定了 slice 能扩展的空间。当 len 超过了 cap就需要扩容。
slice 的元素会存在于一段连续的内存中底层实现是数组。如果初始化的 切片是 var ints []int则data nil、len、cap 均为0没有底层数据结构。而make 的方式初始化slice会按照make的传参来初始化 slice 的底层数据机构。new 方法创建的 slice 同样没有底层数组只有通过 append 才能为其创建底层数组 并追加元素。
数组转切片
func main() {arr : [10]int{1, 2, 3, 4, 5, 6, 7, 8, 9, 10}var (s1 []int arr[1:5]s2 []int arr[7:])fmt.Printf(%v,%v,%v\n, s1, len(s1), cap(s1)) //cap 9fmt.Printf(%v,%v,%v, s2, len(s2), cap(s2)) //cap 3
}上面这段代码就是利用 数组的底层逻辑来划分切片。
首先数组的底层是 0~9 一共 10 位。s1 利用 cap 从数组的 1位开始1位对应的是数字1那么len 5-1 cap 10 - 1s2 利用 cap 从数组的 7位开始7位对应的数字7那么 len 10-7cap 10 - 7
重点注意slice 的 cap 是一个非常关键的字段cap 是从数组的第一个元素开始计算而不是从最后一个元素开始计算。比如[1:10] 和 [2:5]只会从 1 / 2 开始计算数组长度而不会从 10 /5 计算。
浅拷贝 和 深拷贝
概念
浅拷贝 只 复制对象的指针两个对象会用同一块内存空间修改其中一个会影响另一个。深拷贝 是 复制对象的数据而不复制对象的指针两个对象拥有各自的内存空间修改其中一个对象不会影响另一个。
通过实例来理解深拷贝和浅拷贝
假设我们有一个结构体类型 Person它有两个字段Name 和 Age。我们可以用下面的代码来创建一个 Person 的实例
p1 : Person{Name: Alice, Age: 20}如果我们想要复制 p1 的值我们可以用下面的代码来实现深拷贝
p2 : p1 // 深拷贝p2 和 p1 拥有各自的内存空间
p2.Name Bob // 修改 p2 的 Name 字段
fmt.Println(p1.Name) // Alicep1 的 Name 字段不受影响
fmt.Println(p2.Name) // Bobp2 的 Name 字段已经改变但是如果我们想要复制 p1 的指针我们可以用下面的代码来实现浅拷贝
p3 : p1 // 浅拷贝p3 和 p1 共享同一块内存空间
p3.Name Charlie // 修改 p3 的 Name 字段
fmt.Println(p1.Name) // Charliep1 的 Name 字段也被修改了
fmt.Println(p3.Name) // Charliep3 的 Name 字段已经改变注意浅拷贝复制后如果使用append 就会触发新的底层数组也就变成了两个 slice。修改其中一个也就不会导致第二个一并修改了。
渐进式扩容 和 翻倍扩容
slice 默认扩容规则翻倍预估具体规则如下一般的默认规则是在当前基础上 * 2如果 *2 之后的容量可以满足要求则 *2的容量。然后将原有数据复制到新的内存地址中。这种翻倍扩容的方式会重新分配底层数组需要消耗一定的时间 和 内存开销。
渐进式扩容渐进式扩容无需进行翻倍而是根据原来的 cap 和 len 来计算新的 cap。下面是示例代码
func main() {// 创建一个初始 cap 为 10 的 slices : make([]int, 0, 10)// 打印初始的 len 和 capfmt.Printf(len: %d, cap: %d\n, len(s), cap(s))// 循环添加元素for i : 0; i 100; i {s SliceAppend(s, i)// 每添加 10 个元素打印一次 len 和 capif i%10 0 {fmt.Printf(len: %d, cap: %d\n, len(s), cap(s))}}
}func SliceAppend(old []int, i int) []int {var newCaps int//判断当前的长度是否超过了容量if len(old) cap(old) {//则需要计算新的容量如果原来的容量 《 1024 则直接翻倍if cap(old) 1024 {newCaps cap(old) * 2} else {newCaps cap(old) * 5 / 4}} else {old append(old, i)return old}//创建一个新的 slice, 并将容量 新的容量newSlice : make([]int, len(old), newCaps)//复制原来的元素copy(newSlice, old)newSlice append(newSlice, i)return newSlice
}这种渐进式扩容的方式通过cap 和 len 来计算新的 cap而不是简单的翻倍避免系统过度分配内存减少内存浪费和垃圾回收压力。渐进式扩容可以提高内存利用率减少内存碎片和重新分配的次数因为它可以适应不同大小和增长的 slice提高灵活性。不依赖于 append 函数的内部实现更清晰的控制 slice 的 cap自定义扩容策略。
Benchmark 测试代码如下
package testimport (fmttesting
)func growSlice(old []int, n int) []int {if old nil {return nil}var newCaps intif len(old) cap(old) {if cap(old) 1024 {newCaps cap(old) * 2} else {newCaps cap(old) * 5 / 4}} else {old append(old, n)return old}newSlice : make([]int, len(old), newCaps)copy(newSlice, old)newSlice append(newSlice, n)return newSlice
}func BenchmarkGrowSlice(b *testing.B) {s1 : make([]int, 0, 10)b.StartTimer()for i : 0; i b.N; i {//使用渐进式添加s1 growSlice(s1, i)}b.StopTimer()fmt.Printf(s1 len: %d, cap: %d\n, len(s1), cap(s1)) // 打印 s1 的 len 和 cap
}func BenchmarkAppendSlice(b *testing.B) {s1 : make([]int, 0, 10)b.StartTimer()for i : 0; i b.N; i {//使用渐进式添加s1 growSlice(s1, i)}b.StopTimer()fmt.Printf(s2 len: %d, cap: %d\n, len(s1), cap(s1)) // 打印 s1 的 len 和 cap
} Map
map 用来存储 key-value 集合的数据每个key 只会出现一次。Map的主要实现有两种方式“哈希表hash table”和 “搜索树search tree”。golang 和 java 基于哈希表实现c的Map基于 平衡二叉树红黑实现。
map 的key 都允许哪些类型
map的 key-value 中key 允许所有的可以比较的数据类型如下
boolstringint、float所有的指针类型数组类型但是要求数组类型的元素是可以比较的类型。结构体类型结构体中的字段都是可以比较的类型。
不建议使用slice、map、struct、channel 作为map的key因为难以维护。如果非要使用建议将这些类型的数据做成 md5的编码为key值。
map删除条目会触发资源回收吗
首先map中删除一个key在一定条件下会释放资源。要满足条件如下
该map 的key没有任何引用关系没有任何其他引用指向这个key。该map 的key在垃圾回收检测时访问不到这个内存空间就会被释放。
如果需要立即释放要手动调用gc来回收资源。或者显示的清空map。
map 的某些key被删除后map本身的底层数组并不会缩小占有内存空间也不会减少。因为删除key-value只是从分配好的 哈希表中将这个值删除并不会改变底层哈希表的容量大小。
虽然底层哈希表的数组容量不变但是会有GC在满足条件情况下释放资源。
哈希表散列表
哈希表是由 数组 哈希桶链表组合而成其中的几个概念是必须要提前了解的
哈希函数哈希冲突哈希桶哈希桶又称哈希槽每个哈希桶都是一个链表链表中记录着k-v的数据对象。负载因子负载因子是表示哈希表元素的填满程度。负载因子越大元素越多冲突概率就越高。负载因子是用来决定哈希表是否要进扩容的关键指标参数。
哈希函数
哈希函数又称为散列函数用于将任意大小的数据映射到固定大小值得函数常见得包括MD5\SHA等。一个优秀的 hash 函数应该包含以下特性
均匀性一个好的哈希函数应该在其输出范围内尽可能均匀地映射也就是说应以大致相同的概率生成输出范围内的每个哈希值。效率高哈希效率要高即使很长的输入参数也能快速计算出哈希值。可确定性哈希过程必须是确定性的这意味着对于给定的输入值它必须始终生成相同的哈希值。雪崩效应微小的输入值变化也会让输出值发生巨大的变化。不可逆从哈希函数的输出值不可反向推导出原始的数据。
哈希冲突
哈希冲突是随着 kv数据增多哈希函数的结果可能会出现冲突的情况当一个哈希桶中出现两个相同的 k-v就会出现冲突。解决方法如下
链式存储
链式存储的思想就是将映射到一个哈希桶中的所有元素都使用链表串起来。如果出现冲突了就将其追加到链表的后面使用链表组织起来。这样做在查询时需要遍历整个链表。如果头节点的key不相同就要向链表的下级节点一致遍历。
开放寻址
开放寻址是不断进行查找空余的哈希桶如果最开始得到的哈希桶已经存在数据则继续寻找下一个哈希桶直到找到一个空闲的哈希桶存储key-value。开放寻址法有多种不同的探测方式包括线性探测、二次探测、双重散列等。但是其核心思想都是不断寻找下一个位置直到找到一个空槽。
Go Map - hmap 底层实现
golang 基于 哈希表来实现采用链式存储方法解决哈希冲突问题。map 的底层结构是一个 hmap 的结构体关键字段 buckets 是指向 bmap 数组的指针以及其它字段分别指向旧桶扩容进度溢出桶等信息。 // A header for a Go map.
type hmap struct {count int // 代表哈希表中的元素个数调用len(map)时返回的就是该字段值。flags uint8 // 状态标志下文常量中会解释四种状态位含义。B uint8 // buckets桶的对数log_2哈希表元素数量最大可达到装载因子*2^Bnoverflow uint16 // 溢出桶的大概数量。hash0 uint32 // 哈希种子。buckets unsafe.Pointer // 指向buckets数组的指针数组大小为2^B如果元素个数为0它为nil。oldbuckets unsafe.Pointer // 如果发生扩容oldbuckets是指向老的buckets数组的指针老的buckets数组大小是新的buckets的1/2。非扩容状态下它为nil。nevacuate uintptr // 表示扩容进度小于此地址的buckets代表已搬迁完成。extra *mapextra // 这个字段是为了优化GC扫描而设计的。当key和value均不包含指针并且都可以inline时使用。extra是指向mapextra类型的指针。
最重要的是 buckets 字段指向哈希桶数组数组的元素是 bmap。bmap 是一个哈希桶哈希桶用来存储key-value 数据在源代码阶段是如下
// A bucket for a Go map.
type bmap struct {tophash [bucketCnt]uint8 //存储每个key哈希值的 高8位以方便查找
}桶外确定数组位置hmap 中 buckets 数组长度的对数 是指桶数组的长度用来计算数组长度的公式2的B次方。例如 B 3 那么桶数组的长度是 2^3 8。方便通过位与运算来计算桶的位置。hash-key 值的低 B 位是指将 hash 后的按照 二进制表示取最低的 B位。如下所示 如果 hash 值是 0x12345678那么它的二进制表示是 0001 0010 0011 0100 0101 0110 0111 1000 如果 B 4那么它的低 B 位就是 1000 如果 B 8那么它的低 B 位就是 0111 1000 如果 B 12那么它的低 B 位就是 0101 0111 1000 hash 值得低B位绝对了key-value 落入那个 buckets 数组的 bmap。
桶内寻找key-value位置tophash 是一个数组类型数组长度是8。用来存储每个key哈希值的高8位通过高8位来寻找哈希桶内位置判断key-value是否冲突。每个哈希桶最多存储8个键值对。
为了让哈希桶存放数据更加的紧凑8个key和8个value会排列在一起存储。tophash根据哈希值的高8位来确定存放在哈希桶的位置。 举例说明 哈希值 与 tophash 的关系 假设一个桶中有4个key它们的哈希值分别是 0x12345678 0x87654321 0xABCDEF01 0x10FEDCBA 那么它们的高 8 位分别是 0x12 0x87 0xAB 0x10 那么 tophash 数组就是 [0x12, 0x87, 0xAB, 0x10, 0, 0, 0, 0] 注意在tophash中未使用的位置会填充0表示空闲。 当存储key-value时在tophash 数组中寻找 key哈希值 是否在数组中存在相同的高8位如果有则代表找到了对应的key如果没有说明不在这个桶中。具体的操作还要继续判断。
判断 key-value 是否冲突找到对应的key之后还要进一步比较完整的哈希值 和 key是否相等。如果完整的hash值和key都相等说明已经存在map中直接进行覆盖更新替换value。如果完整的哈希值和key并不相等说明key 的高8位发生了哈希冲突高8位虽然相同但是实际上是不同的key。那么就需要在桶内寻找下一个空闲的位置将 key-value 存储在空闲处。如果桶内没有空闲的位置就需要创建一个新的桶用 overflow指针指向这个新的桶并在新的溢出桶中存储这个 key和value。
注意
overflow 溢出桶就是一个 bmap新的bmap 内部结构完全一样。一个bmap 可能会关联多个溢出桶最终形成一个链表。 例如假设有以下几个 key 0x12345678 0x22345678 0x32345678 0x42345678 0x52345678 0x62345678 0x72345678 0x82345678 如果 B 8那么它们的低 B 位都是 0111 1000也就是说它们都会被映射到同一个 bmap 中。但是一个 bmap 只能存储 8 个 key-value 对所以当第九个 key 来的时候就需要创建一个新的溢出桶并将 overflow 指针指向它。如果还有第十个、第十一个、第十二个…这样的 key 来的时候就需要不断地创建新的溢出桶并将它们串成一个链表。 在编译阶段期间 bmap会动态添加字段完整的bmap 如下
type bmap struct { topbits [8]uint8 //tophash keys [8]keytype //key 存储的连续keyvalues [8]valuetype //value 存储的连续valuepad uintptr //用于对齐的字段填充overflow uintptr //overflow 指向下一个桶的指针
}这样编译阶段补全的方式是为了让bmap结构更灵活根据不同的key 和 value 类型来生成不同的 bmap 结构。
总结map实现
map的实现依赖底层 hmap 数据结构该数据结构中包含了一个 buckets 数组数组的每一个元素是一个哈希桶key和value 就存储在这个 哈希桶中。buckets 数组元素是 bmap 数据结构一个 bmap 最多存储 8对 key-value。如果对key-value指向增删改操作时会在bmap中寻找匹配的key如果找不到则判断是否存在溢出桶如果有则去溢出桶寻找溢出桶如果还是没有则返回 value 的默认值。
map 存储key-value 的过程
通过哈希函数得到一个存储 kv的哈希值。根据哈希值的低B位确定选择数组中的哪一个桶B对应 hmap 中的一个字段表示桶数组长度的对数。根据哈希值的高8位确定哈希桶内部的位置使用tophash数组来存储每一个key的高8位。如果桶内 或者 关联的溢出桶 能找到相同的key则覆盖value如果桶内已经满了 或者 出现冲突则创建一个新的桶通过 overflow 字段指向溢出桶。如果 map 的负载因子超过了阈值则触发扩容操作将桶的数组长度扩大一倍并重新分配所有的key-value 键值对。
map 遍历过程为什么无序
map遍历使用for range 语句来实现for range 会调用 map的迭代器迭代器会返回 map 中的key 和 value。map 的迭代器 是一个 hiter 结构体包含了遍历map 所需要的信息比如当前桶索引、桶中元素索引、溢出桶索引等等。map 的迭代器会按照以下步骤遍历 map 中的元素
检查 map 是否为空或者正在扩容如果是直接返回。然后从第一个桶开始遍历每一个 bmap 的 cell如果 cell 不为空则返回 cell 的key-value。如果 bmap 的 overflow 不为空包含溢出桶则继续遍历溢出桶的 cell。直到遍历完所有的桶为止。
map 无序的原因
map 的元素首先存储时就是无序的是按照哈希函数进行分布的。map 在每次遍历的时候 都会随机生成序号 bucket再从其中随机的 cell 进行遍历。具体随机代码如下
// 生成随机数
r : uintptr(fastrand())
if h.B 31-bucketCntBits {r uintptr(fastrand()) 31
}
// 决定了从哪个随机的bucket开始
it.startBucket r bucketMask(h.B)
// 决定了每个bucket中随机的cell的位置
it.offset uint8(r h.B (bucketCnt - 1))注意如果需要特定顺序遍历需要先取出key 进行排序然后按照key 的顺序取出map的value
map 扩容
触发扩容条件
负载因子是否超过 6.5 判断是否有过多的溢出桶 总结map扩容触发条件一种是map的负载因子超过了6.5一种是溢出桶太多了。
如何扩容 map 扩容的核心函数如下
hashGrow负责初始化新的桶。groWork迁移数据每次调用最多迁移两个桶的数据。调用 evacuate。evacuate真正干活的迁移函数负责迁移指定桶中的数据。advanceEvacuationMark收尾工作增加 nevacuate确定所有的 oldbuckets 都迁移完了。
扩容迁移具体步骤
开始扩容准备新的哈希桶数组
需要扩容时就要创建更多的新桶然后把旧桶存储的 key - value 都迁移到新桶里如果哈希表存储的key-value 数据较多一次性迁移所有桶花费的时间会比较显著所以通常在分配哈希表进行扩容时会先分配足够多的新桶然后用 oldbuckets 字段记录旧桶的位置再加一个字段记录旧桶迁移的进度进度下一个要迁移的旧桶编号。
数据迁移
扩容完成后需要做数据的迁移。数据的迁移并不是一次完成的是使用时才会做对应 bucket 的迁移每次做多迁移两个哈希桶数据直到逐步完成迁移。
从第一个hash桶开始检查是否已经被搬迁到新的 哈希桶中 如果是则跳过如果不是则将该桶中的元素重新计算哈希值和位置并复制到新的桶中。 如果该桶存在溢出桶也按照同样的方式重新计算哈希值和位置再搬迁溢出桶中的元素并释放溢出桶内存。最后如果该桶中没有溢出桶则继续搬迁下一个桶直到搬迁完所有桶位置。
map 如何分辨读写数据在新桶还是旧桶
map 会通过 evacuatedX 和 evacuatedY 两个标志位来分辨要读写的数据是在旧桶还是在新桶。 当map 进行扩容时会将原来旧桶赋值给 oldbuckets 字段然后创建一个新的 哈希桶数组赋值给 buckets。新桶的长度是原来的二倍。 然后会从第一个旧桶开始将数据迁移到新桶中迁移时会根据哈希值和新桶的数量重新进行计算元素位置然后分配到新桶中完成迁移。 迁移完成后会将旧桶中的 tophash 数组每个元素都设置 evacuatedX 或者 evacuatedY。表示该旧桶已经被迁移。
evacuatedX表示迁移后的位置在原来位置上 ~ 原来位置 旧桶数量 之间如原来是3旧桶是8则迁移后的位置区间是 3 ~11。
evacuatedY表示迁移后的位置在原来位置上 旧桶数量 ~ 原来位置 2倍旧桶数量之间如果原来是3旧桶数量是8则迁移后的位置在 11~19之间。
当map读写时会先根据哈希值和桶数量计算出桶的位置并检查该桶是否已经被迁移如果是则会根据 evacuatedX 和 evacuatedY 的标志位从新桶中获取元素如果不是则从原来的旧桶中获取元素。
map 迁移时如何保护读写不受影响
读取时
map遍历时根据 evacuatedX / evacuatedY标志位进行判断判断k-v 目前是在新桶中还是在旧桶中。通过 hmap.nevacuate 字段的值可以判断已经搬迁了多少个旧桶如果某个旧桶的索引小于该值则说明该旧桶已经完全被搬迁可以直接读新桶。
写入时
在map写入时会先检查是否被迁移过如果已经完成了迁移则直接到新桶中进行处理。如果没有完成迁移则会调用 growWork 函数该函数会根据迁移的进度找到还未迁移的哈希桶将其数据复制到新的哈希桶中在进行写入更新。
map 遍历、读写时都会根据哈希值和桶数量计算出桶的位置并检查该桶是否已经被迁移如果是则会从新的桶数组中获取元素如果不是则会从原来的桶数组中获取元素。
Mutex 锁
Mutex 锁的使用
package mainimport (fmtsynctime
)var mu sync.Mutex
var wg sync.WaitGroupfunc main() {// 模拟正常模式fmt.Println(Normal mode:)wg.Add(3)go worker(1, 100*time.Millisecond) // 持有锁100msgo worker(2, 0) // 立即释放锁go worker(3, 0) // 立即释放锁wg.Wait()fmt.Println()// 模拟饥饿模式fmt.Println(Starvation mode:)wg.Add(3)go worker(1, 0) // 立即释放锁go worker(2, 0) // 立即释放锁go worker(3, 2*time.Second) // 持有锁2swg.Wait()fmt.Println()
}func worker(id int, duration time.Duration) {defer wg.Done()fmt.Printf(worker %d: trying to lock\n, id)mu.Lock() //上锁fmt.Printf(worker %d: locked\n, id)time.Sleep(duration) // 模拟占用锁的时间mu.Unlock() //解锁fmt.Printf(worker %d: unlocked\n, id)
}
Mutex 实现原理
Mutex 内置了一个状态值state 标识锁的状态以及一个 sema 信号量。
type Mutex struct {state int32 //表示互斥锁的状态sema uint32 //信号量变量用来控制等待 goroutinee 的阻塞休眠和唤醒
}const (mutexLocked 1 iota //第一位是否持有互斥锁mutexWoken //第二位是否有被唤醒的等待者mutexStarving //第三位互斥锁是否处于饥饿模式mutexWaiterShift iota //阻塞等待的数量starvationThresholdNs 1e6 //饥饿阈值 1ms
)state 是int32类型通过二进制位来表示锁的状态
第1位 Locked表示锁是否被持有1 加锁0 解锁。第2位 Woken表示是否有被唤醒的等待者1 有0 没有。第3位 Starving表示是否处于饥饿模式1 饥饿模式0 正常工作模式。第4~n位等待偏移量包含一些排队上锁的
由Lock 和 UnLock 实现加解锁具体的逻辑工作函数由 LockSlow 和 UnLockSlow 来实现。
mutex 锁升级
针对 goroutines 竞争锁资源golang有两种处理方法也是锁升级的一个过程。以下是处理 锁竞争时的两种处理方法 阻塞 / 唤醒 阻塞当goroutine 试图加锁失败出现阻塞后当前 goroutine 会被直接挂起。唤醒当goroutine 等待锁资源释放并成功抢到锁资源后再将goroutine 重新唤醒继续执行。 自旋CAS: 基于自选结合 CAS 的方式重复校验锁的状态并尝试获取到锁始终处于循环状态。
自旋CAS 无序阻塞唤醒goroutine但是会长时间for循环占用CPU资源适合并发情况较低的场景。阻塞唤醒需要频繁的挂起协程进行上下文切换操作较重适合并发场景高的场景。
golang 基于以上两种 场景设计了一个锁升级机制针对并发环境由乐观正常模式逐渐转向悲观饥饿模式的过程进行了区分实现。
mutex 如何判断锁竞争是否激烈
通过 state 和 seam 信号量来判断。
首先是保持正常模式正常模式下的 goroutine 会自旋CAS 的方式竞争锁自旋。当mutex 的 state 字段表示mutex锁的状态如果 state 值经常发生变化或者有很多的 goroutine 处于排队状态就说明锁的竞争比较激烈。seam 信号量字段用于阻塞和唤醒等待的 goroutine。如果 seam 信号量频繁发生变化就说明有很多阻塞 和 唤醒的goroutine操作说明锁竞争很激烈。
重点原因
mutex 的自旋次数和时间是源代码中模式切换的阈值 自旋时间超过 1ms 没有获取到锁自旋次数超过 4次 没有获取到锁
CAS 自旋
CAS 是 CPU 中的一个指令自旋的实现依赖 CAS。当一个 Goroutine 在获取锁的时候如果锁资源已经被其它 Goroutine 持有那么当前 Goroutine 会一致循环等待监听 mutex不断的判断是否能够成功获取到锁直到获取到锁才会退出循环。
mutex 的正常模式和饥饿模式
正常模式过程详解
正常状态的 state 1在尝试加锁时 goroutine 会自旋4次尝试通过原子性操作获取到锁。若无法获取锁则通过seam信号量进入阻塞状态进入排队等待状态。所有的goroutine 都会按照先入先出 FIFO 的顺序等待排队。当一个持有锁的 Goroutine 调用 UnLock 方法时会检查当前 mutex 的 state字段判断是否有等待的 goroutine并且还没有被唤醒则会选择一个阻塞等待的 goroutine 唤醒。当 Goroutine 被唤醒后并不会直接拥有锁而是需要跟后来者竞争这些后来者就是处于自旋阶段且尚未排队的goroutine。如果没有竞争到锁资源就会重新排队插入到队列的头部等待下一次的唤醒。为了避免这种频繁唤醒却拿不到锁资源的情况当一个 goroutine 加锁等待时间超过 1ms还是不能获取到锁就会将mutex 转变为饥饿模式。
饥饿模式过程详解
饥饿模式是指 goroutine 等待锁的时间过长会切换正常模式到互斥模式。在饥饿模型下锁的所有权会直接从解锁的 goroutine 传递给等待队列中的第一个 Goroutine后来的 goroutine 不会再进行自旋或者尝试获取。保证了公平性。
饥饿模式如何转换回正常模式
队列首个 goroutine 的等待时间已经小于 1ms队列中没有 goroutine 排队获取到锁之后就会将 mutex state 状态从饥饿模式转变为正常模式。
正常模式和饥饿模式哪个更好
正常模式的优点是能够提高吞吐量因为频繁的挂起和唤醒 goroutine 会带来较多的开销。缺点是可能会出现队列尾端排队的 goroutine 迟迟抢不到锁。
饥饿模式的优点是能够保证公平性所有 goroutine 都需要排队严格的 FIFO。缺点是性能会相对差一些因为锁的所有权需要传递不能直接被来的 goroutine 抢占。
mutex 的 midPath 和 osq Lock
mid path是一种介于 fast path 和 slow path 之间的代码路径也叫 optimistic spinning乐观锁它是指当 mutex 被其他协程持有时当前协程不进入等待队列而是自旋等待锁的释放希望能够快速获取锁。这种方式可以避免上下文切换的开销但是也会消耗 CPU 资源。
osq lock是一种用于实现 mid path 的锁机制它是一种结合了 MCS lock 和 qspinlock 的实现又根据 mutex 的语义量身裁剪过的锁。它可以保证处于 mid path 的协程之间的公平性和高效性避免过分乐观的自旋等待。osq lock 使用一个双向链表来组织处于 mid path 的协程每个协程都有一个 optimistic_spin_node 结构体来记录自己的状态和位置。osq lock 还使用一个原子变量 tail 来记录链表的尾部节点方便新来的协程加入链表。当持有 mutex 的协程释放锁时它会检查 osq lock 队列是否为空如果不为空就会释放 osq lock并让 spinner 们竞争锁。
Lock() 加锁过程
fast path 快速路径
在 Lock() 中通过原子性操作实现处理理想的锁状态而哪些非预期的状态放在了 LockSlow 方法中处理。
func (m *Mutex) Lock() { //atomic判断当前锁状态是不是0如果是0则设置成1表示加锁if atomic.CompareAndSwapInt32(m.state, 0, mutexLocked) {if race.Enabled {race.Acquire(unsafe.Pointer(m))}return} m.lockSlow()
}Lock方法中会通过原子性操作判断互斥锁当前是否处于解锁状态如果是则直接加锁理想状态下只需要一个自旋操作就能够获取到锁如果一个CAS 没有获取到锁就会进入 LockSlow方法处理。
lockslow goroutine 队列饥饿模式声明一些局部变量用来存储状态
func (m *Mutex) Lock() { // 等待时间var waitStartTime int64// 饥饿标记starving : false// 唤醒标记awoke : false// 自旋次数iter : 0// 当前的锁的状态old : m.state...
}方法进来之后会先创建几个局部变量
waitStartTime记录goroutine 进入阻塞时的时间点starving是否开启饥饿模式awoke是否有尝试唤醒的 goroutine 在等锁iter标记当前 goroutine 参与自旋的次数old临时存储锁的 state 状态。 自旋CAS 空转
func (m *Mutex) Lock() { ...
for { // 锁是非饥饿状态锁还没被释放尝试自旋if old(mutexLocked|mutexStarving) mutexLocked runtime_canSpin(iter) {//将当前goroutine设置成唤醒状态worken 1 if !awoke oldmutexWoken 0 oldmutexWaiterShift ! 0 atomic.CompareAndSwapInt32(m.state, old, old|mutexWoken) {awoke true}// 自旋runtime_doSpin()// 自旋次数加1iter// 设置当前锁的状态old m.statecontinue}...
}进入 for 循环对锁当前的状态进行判断满足三个条件Ⅰ加锁状态、Ⅱ锁处于正常模式、Ⅲruntime_canSpin 还允许继续执行自旋。则进入自旋状态检查是否满足自旋状态进入自旋模式后 执行自旋 runtime_doSpin()iter 自旋次数重新赋值当前 mutex 状态
自旋后预写状态新值
如果自旋的条件不满足则需要对新的状态代码进行更新具体如下
new : old
if oldmutexStarving 0 {// 如果当前不是饥饿模式那么将mutexLocked状态位设置1表示加锁new | mutexLocked
}
if old(mutexLocked|mutexStarving) ! 0 {// 如果当前被锁定或者处于饥饿模式则waiter加一表示等待一个等待计数new 1 mutexWaiterShift
}
// 如果是饥饿状态并且已经上锁了那么mutexStarving状态位设置为1设置为饥饿状态
if starving oldmutexLocked ! 0 {new | mutexStarving
}
// awoke为true则表明当前线程在上面自旋的时候修改mutexWoken状态成功
if awoke { if newmutexWoken 0 {throw(sync: inconsistent mutex state)}// 清除唤醒标志位new ^ mutexWoken
}第一个判断如果当前加了锁但是没有处于饥饿状态会重新加一次锁属于一种保底策略保证当前 goroutine 一定加锁成功。第二个判断如果锁处于饥饿模式或者已经上锁了则在队列中要加上一个等待的 goroutine。第三个判断如果锁处于饥饿模式并且已经上锁了则将mutex 状态改成饥饿模式第四个判断如果在自旋时设置成功那么这里要消除 Woken 标志位。因为后续流程中很有可能当前线程会被挂起需要等待其它释放锁的 gouroutine来唤醒。如果 unlock 时发现 mutexWoken 位置不是0就不会去唤醒了。
继续判断是否需要加锁
if atomic.CompareAndSwapInt32(m.state, old, new) {// 1.如果原来状态没有上锁也没有饥饿那么直接返回表示获取到锁if old(mutexLocked|mutexStarving) 0 {break // locked the mutex with CAS}// 2.到这里是没有获取到锁判断一下等待时长是否不为0// 如果不为0那么加入到队列头部queueLifo : waitStartTime ! 0// 3.如果等待时间为0那么初始化等待时间if waitStartTime 0 {waitStartTime runtime_nanotime()}// 4.阻塞等待runtime_SemacquireMutex(m.sema, queueLifo, 1)// 5.唤醒之后检查锁是否应该处于饥饿状态starving starving || runtime_nanotime()-waitStartTime starvationThresholdNsold m.state// 6.判断是否已经处于饥饿状态if oldmutexStarving ! 0 { if old(mutexLocked|mutexWoken) ! 0 || oldmutexWaiterShift 0 {throw(sync: inconsistent mutex state)}// 7.加锁并且将waiter数减1delta : int32(mutexLocked - 1mutexWaiterShift)if !starving || oldmutexWaiterShift 1 { // 8.如果当前goroutine不是饥饿状态就从饥饿模式切换会正常模式delta - mutexStarving}// 9.设置状态atomic.AddInt32(m.state, delta)break}awoke trueiter 0
} else {old m.state
}到这里之后首先会CAS设置新的状态如果成功则往下走否则返回之后循环设置状态。
首先会判断old状态如果没有饥饿也没有获取到锁那么直接返回因为这种情况在进入到这段代码之前会将new状态设置为mutexLocked表示已经获取到锁。这里还判断了一下old状态不能为饥饿状态否则也不能获取到锁判断waitStartTime是否已经初始化过了如果是新的goroutine来抢占锁那么queueLifo会返回false如果不是新的goroutine来抢占锁那么加入到等待队列头部这样等待最久的 goroutine 优先能够获取到锁如果等待时间为0那么初始化等待时间阻塞等待当前goroutine进行休眠唤醒之后检查锁是否应该处于饥饿状态并设置starving变量值判断是否已经处于饥饿状态如果不处于饥饿状态那么这里直接进入到下一个for循环中获取锁加锁并且将waiter数减1这里我看了一会没用懂什么意思其实需要分两步来理解相当于statemutexLocked然后state再将waiter部分的数减一如果当前goroutine不是饥饿状态或者waiter只有一个就从饥饿模式切换会正常模式设置状态 加锁流程面试总结
lock() 方法加锁时分为两种状态
fastPath 理想状态互斥锁处于解锁状态直接通过CAS获取锁。lockSlow goroutine队列 饥饿模式通过 goroutine 排队的方式来获取锁。
具体流程如下
首先通过 CAS 原子性操作尝试将 mutex 加锁。如果成功表示加锁成功。CAS 失败后表示当前 mutex 已经被其它 goroutine 持有会进入 slowPath 。在 slow Path 中继续判断是否可以进行自旋等待。如果自旋次数 达到4次还没有获取到锁就会强制进入饥饿模式。在饥饿模式中会排队等待锁的释放。 将当前 goroutine 加入到队首。当持有 mutex 的 goroutine 调用Unlock 时会检查 wait list 队列是否为空并唤醒排队队首的 goroutine。将锁的所有权传递给它避免饥饿。 如果当前 goroutine 就是 mutex 阻塞队列的最后一个则将 state 设置为正常模式。
UnLock() 解锁过程
fast pathslow path
具体流程如下 首先检查 mutex 的 state 是否 1如果不为1则表示没有被锁定属于运行时错误导致panic。 fast Path 如果 state 1则表示当前mutex 被goroutine 持有尝试通过原子操作将state 设置为0如果设置成功则表示解锁成功。 slow path 如果设置 state 0 失败就表示 mutex 还有其它协程在等待或者自旋进入 slow path。在slow path 中会判断是否有等待的 Goroutine如果有则会唤醒队首的 Goroutine并将锁的所有权传递避免饥饿。直到 Goroutine 阻塞队列为空时并且没有自旋的 Goroutine 在抢锁后就会将 state 0并退出临界区。
RWMutex 读写锁
RWMutex 可以理解为一把读锁一把写锁写锁
具有严格的排他性互斥读、写当其被占用时其它视图获取读写锁的 goroutine 将会被阻塞。
读锁
具有共享性互斥写锁兼容读锁。视图获取写锁时会进入阻塞视图读取操作可以与其它goroutine 共享 读锁资源。
RWMutex 适合读多写少的场景最理想化的情况是所有操作均可以使用读锁避免了程序出现阻塞最坏的情况是所有操作均使用写锁退化成普通的 Mutex。
底层数据结构实现
RWMutex 读写锁通过 count 计数器 和 Sem 信号量的方式记录读锁数量和写锁的数量。
type RWMutex struct {w Mutex // 互斥锁用于保护内部字段writerSem uint32 // 写信号量用于唤醒阻塞的写者readerSem uint32 // 读信号量用于唤醒阻塞的读者readerCount int32 // 读者计数器记录当前活跃的读者数量readerWait int32 // 等待计数器记录当前等待的读者数量
}关键字段作用如下w互斥锁用户保护内部字段以及实现写优先逻辑。
当有写操作时会先尝试获取 w互斥锁如果成功则可以进行写入操作。如果失败则表示有其它写操作 或者 读操作正在进行需要等待 w 锁的释放。
writerSem写信号量用于唤醒阻塞的写 Goroutine。
当有写操作被阻塞时会在 writeSem上等待。当 w锁 被释放时会检查所有等待的 Goroutine如果有就会通过 writeSem 唤醒并将锁的所有权传递。
readerSem读信号量用于唤醒阻塞的读 Goroutine。
当有读操作被阻塞时会在 readerSem 上等待。当 w锁 被释放时会检查所有读的 Goroutine如果有就通过 readerSem 唤醒所有的 Goroutines并让它们可以并发读取。
readerCount读计数器
记录当前活跃的读者数量。当有读操作开始时会原子性的增加 readerCount的值。当有读操作结束时会原子性的减少 readerCount的值。当readerCount 0时代表没有读操作可以写入。
readerWait等待计数器
记录当前等待的读者数量。当有读操作被阻塞时会原子性的增加 readerWait的值当前有读操作被唤醒则原子性减少 readerWait 的值。当 readerWait 0 代表没有等待的读者可以释放 w锁。
读锁操作
RLock() 加锁执行过程
func (rw *RWMutex) RLock() {// 如果启用了 race检测,就先禁用它if race.Enabled {_ rw.w.staterace.Disable()}//原子性的增加 readerCount 的值if atomic.AddInt32(rw.readerCount, 1) 0 {// A writer is pending, wait for it.//如果readerCount 小于0表示有writer 在等待或者执行写操作阻塞在readerSem上。runtime_SemacquireMutex(rw.readerSem, false, 0)}// 如果启用了 race 检测就重新启用它并获取 readerSem 的所有权。if race.Enabled {race.Enable()race.Acquire(unsafe.Pointer(rw.readerSem))}
}调用RLock方法检查是否有写锁 或者 阻塞等待获取写锁 的 Goroutine。使用atomic.AddInt32 原子性的增加 readerCount 的值表示有一个 reader 获取了读锁。接着判断 当前的 readerCount是否小于0。 如果 readerCount 的值小于 0 为负数就会进入等待状态说明此时有写入操作读操作被阻塞在 readerSem上表示 写 操作完成后唤醒。
为什么ReaderCount 会出现小于 0 的情况
判断 readerCount 是否小于0是为了判断是否有 writer 写入操作在队列 A 中等待或者正在执行写操作。如果有 write 在队列A中等待那么ReaderCount 会减去 MaxReaders使 ReaderCount 变成负数。变成负数后新来的 reader 读操作就会知道此时有 writer 写入操作正在执行需要在信号量中排队等待。当writer 完成后会把 readerCount 加回那个很大的数使其变回正数。这样后面的 reader 就可以获取读锁了。
RUnlock() 解锁执行过程
func (rw *RWMutex) RUnlock() {if race.Enabled {_ rw.w.staterace.ReleaseMerge(unsafe.Pointer(rw.writerSem))race.Disable()}if r : atomic.AddInt32(rw.readerCount, -1); r 0 {// Outlined slow-path to allow the fast-path to be inlinedrw.rUnlockSlow(r)}if race.Enabled {race.Enable()}
}func (rw *RWMutex) rUnlockSlow(r int32) {if r1 0 || r1 -rwmutexMaxReaders {race.Enable()fatal(sync: RUnlock of unlocked RWMutex)}// A writer is pending.if atomic.AddInt32(rw.readerWait, -1) 0 {// The last reader unblocks the writer.runtime_Semrelease(rw.writerSem, false, 1)}
}首先使用atmoic.AddInt32 原子性的将 readerCount - 1表示放当前Goroutine 的读锁。接着判断 readerCount 是否为负数如果是负数表示 write 在队列中等待执行此时需要调用 rUnlockSlow 方法进行慢速路径进行处理。处理完成后结束函数表示释放了读锁。
rUnlockSlow 方法
首先判断 readerCount 是否 0 或者 -MaxReaders如果是表示没有 reader 持有读锁或者读锁已经溢出此时发生 fatal 错误。使用 atomic.Addint32 原子性的减少 readerWait的值表示有一个 reader 从队列中退出等待。判断 readerWait 是否为零值如果是零则表示队列中已经没有了 reader此时需要唤醒 write 协程并将 writerSem 信号量设置为 1。
总结
解锁时有 fast path 和 slow path 两种解锁方式第一种是直接 减去 readerCount 信号量表示释放了一个读锁。第二种是需要从 readerWait 阻塞队列中释放一个读锁。因为readerCount 0 表示当前有写入操作正在执行有写操作进行时读操作会都进入阻塞等待状态。
写锁操作
lock 加写锁
// Lock locks rw for writing.
// If the lock is already locked for reading or writing,
// Lock blocks until the lock is available.
func (rw *RWMutex) Lock() {if race.Enabled {_ rw.w.staterace.Disable()}// First, resolve competition with other writers.rw.w.Lock()// Announce to readers there is a pending writer.r : atomic.AddInt32(rw.readerCount, -rwmutexMaxReaders) rwmutexMaxReaders// Wait for active readers.if r ! 0 atomic.AddInt32(rw.readerWait, r) ! 0 {runtime_SemacquireMutex(rw.writerSem, false, 0)}if race.Enabled {race.Enable()race.Acquire(unsafe.Pointer(rw.readerSem))race.Acquire(unsafe.Pointer(rw.writerSem))}
}调用 w.Lock() 互斥锁加锁的方法该方法会使用互斥锁来保证只有一个写入操作。接着使用 atomic.AddInt32 原子性的减少 readerCount 的值并加上 rwmutexMaxReaders表示有一个 writer 在队列A中等待或者执行写操作。如果 readerCount 不为零表示还有 reader 持有读锁此时需要使用 atomic.AddInt32 原子地增加 readerWait 的值并等待 writerSem 的信号量。
Unlock释放写锁
func (rw *RWMutex) Unlock() {if race.Enabled {_ rw.w.staterace.Release(unsafe.Pointer(rw.readerSem))race.Disable()}// Announce to readers there is no active writer.r : atomic.AddInt32(rw.readerCount, rwmutexMaxReaders)if r rwmutexMaxReaders {race.Enable()fatal(sync: Unlock of unlocked RWMutex)}// Unblock blocked readers, if any.for i : 0; i int(r); i {runtime_Semrelease(rw.readerSem, false, 0)}// Allow other writers to proceed.rw.w.Unlock()if race.Enabled {race.Enable()}
}使用 atomic.AddInt32 元子地增加 readCount 的值并加上 MaxReaders 表示释放了写锁。如果 readerCount MaxReaders 表示没有 reader 持有读锁或者读锁溢出触发fatal。根据 readerCount 的值唤醒相应数量的 reader Goroutine并将ReaderSem 信号量设置为 1。调用 w.Unlock 方法该方法使用互斥锁来释放写协程的怕他性。
channel
前置知识 CSP
CSP 是并发编程的一种编程规范允许多个任务在同一时间执行从而提高程序的性能和效率。CSP 是一种思想用于描述两个独立的并发实体通过共享的通信管道进行通信的方式。“不要以共享内存的方式来通信相反要用通信来共享内存”这样可以避免内存带来的复杂性和不确定性只需要关注通信的逻辑和顺序。
golang 是支持CSP思想的语言使用 channel 作为通信的原语以goroutine 作为并发实体。
channel 特性
goroutine 用于并发执行任务。而 channel 可以同时被多个 goroutine 访问用于在多个goroutines间并发安全的存储和传递数据。
chan T // 声明一个双向通道
chan- T // 声明一个只能用于发送的通道
-chan T // 声明一个只能用于接收的通道有缓冲channel 的特性
buffer channel 带 size 缓冲区属于异步模式只要有剩余容量生产和消费双方就不会阻塞。否则写入方将会被阻塞直到出现接收方才会被唤醒。
从 hchan 数据结构来分析
dataqsize 不为0buf 缓冲区不为空sendx 和 recvx 两个字段用来维护缓冲区中下一个要发送或者接收的元素索引。sendq 和 recvq 两个字段来维护发送或者接收数据的 goroutine 队列实现阻塞和唤醒逻辑。
问题一如果有缓冲channel 写满了还继续写会发生什么答channel 写满了之后会出现死锁阻塞必须有接收方读取才能继续写入。
问题二有缓冲channel 中有数据但是channel 关闭了还能读吗答channel 被关闭后如果当中存在数据可以正常被接收者读取。但是如果channel 关闭后还继续写入则会出现 panic 报错。
无缓冲channel 的特性
unbuffer channel 不带size 缓冲区属于同步模式只有生产和消费双方同步就绪数据才能够进行传递。无缓冲channel 的数据传输是一次性的不会在channel中存储任何元素。
从 hchan 数据结构来分析
qcount 肯定0因为无缓冲区不会在 channel 中存储任何元素所以 qcount 始终0。dataqsize 肯定0表示无传冲区channelrecvq 和 sendq
问题一无缓冲channel 没有接收者会panic吗答无缓冲的channel 如果没有接收者是不会报错的并且可以正常写入数据但是会一直阻塞直到有消费者消费。
channel 数据结构
hchan 数据结构通过make 创建channel 的底层数据结构。
type hchan struct {qcount uint // 元素数量dataqsiz uint // 环形队列的数组长度buf unsafe.Pointer // 缓冲区指向datasize元素类型大小的数组elemsize uint16 // 元素大小closed uint32 // channel 是否关闭elemtype *_type // 元素类型sendx uint // 已发送元素在循环队列中的索引recvx uint // 已接收元素在循环队列中的索引recvq waitq // 等待接收队列sendq waitq // 等待发送队列lock mutex //保护hchan的互斥锁
}qcount
表示channel中当前有效的元素个数在元素入队和出队时做出改变。qcount 永远不会大于 dataqsize。只有 buffer channel 中hchan 才会存储部分元素数据。
dataqsiz
表示channel缓冲区的大小在初始化时赋值之后不会再改变。只有 buffer channel 中才会有缓冲区大小。
buf
表示channel 的缓冲区地址指向一个循环数组用于存放channel中的所有元素。只有 buffer channel 中才会有 buffer 循环数组。
elemsize表示channel中元素的大小根据元素类型而定。elemtype表示channel中元素的类型是一个 *_type 类型用于反射和垃圾回收。closed表示channel是否已经关闭用于修改和读取。sendx指向buf 循环队列表示channel中下一个要发送的元素索引位置用于循环缓冲区。recvx指向buf 循环队列表示channel中下一个要接收的元素索引位置用于循环缓冲区。sendq表示阻塞等待发送数据的 goroutine 队列。指向 waitqrecvq表示阻塞等待接收数据的 goroutine 队列。
waitq 数据结构
type waitq struct {first *sudoglast *sudog
}waitq 是一个链表结构保存在channel中的 goroutine 信息。
first 指向头指针last 指向尾指针
waitq 作用当一个 goroutine 因为channel 操作而被阻塞时就会创建一个sudog 结构体封装当前阻塞的 goroutine并将其加入到 waitq 链表中并等待被唤醒。当一个goroutine 因为channel 操作而被唤醒时会从 waitq中移除对应的 sudog结构体并恢复 goroutine的执行。
recvq 和 sendqchannel 中有两个waitq 链表分别是 recvq 和 sendq。 recvq用于保存等待接收数据的 goroutine 信息。 当channel 的缓冲区为空时接收方会被阻塞并加入到 recvq 中等待执行。如果有发送的数据了会优先唤醒 recvq 中goroutine。 sendq用于保存等待发送数据的 goroutine 信息。 当channel 的缓冲区慢时发送方会阻塞并加入的 sendq 中等待发送。如果有接收者了会优先唤醒 sendq 中的 goroutine。
sudog 数据结构
type sudog struct {g *g //指向被阻塞或唤醒的 goroutine 的指针next *sudog //指向waitq中下一个sudog 的指针实现链表操作prev *sudog //指向waitq中上一个sudog 的指针实现链表操作elem unsafe.Pointer // 指向goroutine 要发送或者接收的数据元素的指针acquiretime int64 //记录goroutine被阻塞的时间用于调试和统计。releasetime int64 //记录goroutine被唤醒的时间用于调试和统计。ticket uint32 //记录goroutine 被阻塞或唤醒时的时钟周期用于调试和统计isSelect bool //标记 goroutine 正在执行select 语句success bool //标记goroutine 是否成功发送或者接收数据parent *sudog //指向select 语句中的sudog结构体用于实现select逻辑waitlink *sudog //指向select 语句中的下一个sudog结构体用于实现select逻辑。waittail *sudog //指向select 语句中最后一个sudog结构体用于实现select逻辑c *hchan //指向goroutine所属的channel结构体。
}作用
sudog用于封装goroutine 的状态和数据。并将其加入到 recvq 和 sendq 中等待被唤醒。当一个 goroutine 因为 channel 操作而被唤醒时会从channel 的 recvq 和 sendq 链表中移除 sudog并恢复 goroutine 继续执行。可以实现select 语句的逻辑多个channel 上进行非阻塞或者随机的选择。
有缓冲channel 的收发过程
初始状态下channel缓冲区为空读写下标sendx 和 recvx 都指向下标0的位置。等待队列 recvq 和 sendq 也都为空。
举例说明假设 ch : make(chan int, 5) 创建一个5缓冲区的channel收发过程如下
首个 goroutine 向channel 中发送5次消息数据因为此时还没有开启消费者所以会不断向 buf 缓冲区中填入5次数据。 每个填入的数据会封装成 sudog 结构体以链表的形式存储在 sendq 中。并且sendx 会随着填入的数据不断递增当写满5个后会回到循环数组的起始位置。此时发现属于 0索引位置的 数据还没有被接收就会进入阻塞状态。
接着创建一个接收方Goroutine不断接收来自channel 中的数据。
此时的recvx 指向的是下一个要消费的索引位置也就是 0 。所以接收开始后第一个接收的数据就是 recvx 0 的数据。此时会通过 sudog 中的元素指针指向内存空间中的数据内容并将其拷贝到channel 中发送给消费者。然后从 recvq 链表中删除 goroutine 的 sudog。并唤醒等待阻塞的 sendq。1将sendq 队列中排队阻塞的6 写入到缓冲区数组中填充到 sendx 0 的循环数组首位更新索引位置最终实现写入。 无缓冲channel 的收发过程
无缓冲cahnnel 的底层数据结构是 hcahn 结构体主要用到了 互斥锁sendq 和 recvq 发送队列和接收队列。这两个队列都是由 sudog 结构体组成的链表sudog 结构体封装了 goroutine 的状态和数据。
向无缓冲channel 发送数据过程
先获取channel 的互斥锁检查channel是否已经关闭了如果关闭了会抛出 panic异常。如果没有关闭则检查接收队列是否有等待的 goroutine 如果有将数据拷贝到对方的 sudog 中。如果没有则将自己封装成一个 sudog加入到发送队列中并阻塞自己等待唤醒。
接收数据过程
会先获取channel 的互斥锁然后检查channel 是否已经被关闭了如果关闭并且发送队列为空则返回 默认值和false。如果 channel 没有关闭则检查发送队列中是否有等待执行的 goroutine。 如果有消费者会将发送方 sudog 中的 elem 拷贝到自己的变量中并将发送发从队列中移除并唤醒发送者。如果没有则将自己封装成一个 sudog 加入到 sendq 队列中并且阻塞自己等待被唤醒。 无缓冲区 与 有缓冲区的最大区别
无缓冲区的数据传递是指针拷贝
无缓冲channel 没有缓冲区来存储数据而缓冲channel 有一定的容量。这意味着当一个 goroutine 像无缓冲区发送数据时数据是不会存存储在缓冲区中的而是直接拷贝到接收者的 sudog 中。无缓冲区通过sudog 来管理等待发送和接收的goroutinesudog 包含 goroutine 的信息。当有goroutine 向无缓冲 channel 发送或者 接收数据时会检查是否有匹配的 goroutine 在等待如果有则直接交换数据并唤醒对方。
有缓冲区的数据传递是从buf 的 sendq 和 recvq 中读取
select
select 执行过程大致如下
首先创建一个 selectgo 数据结构是一个数组数组的每一个元素就是一个 scase 结构体表示select 语句中的所有分支。selectgo 函数会按照一定的顺序加锁所有涉及到的channel以避免出现死锁和竞态条件。selectgo 函数会生成一个随机的轮询顺序用于遍历所有的 case并尝试从channel 中发送或者接收标志并唤醒对应的 goroutine。如果没有任何的 case 成功执行了数据交换则判断是否有 default case如果有default case则返回default case的索引和接收标志。如果没有则继续等待阻塞。selectgo 函数会按照相反的顺序解锁所有涉及到的channel并返回被 case 选择的索引和接收标志。
scase 结构体
scase 使用来实现 select 的一种数据结构属于case分支的信息每一个case分支包含了channel 的指针操作类型发送/接收数据元素的指针等等。核心思想是实现多路复用同时等待多个通道上的数据并执行i相应的代码块提高代码的性能。
type scase struct {c *hchan // chanelem unsafe.Pointer // data element
}sync 包
sync.Once
sync.Map
sync.WaitGroup
sync.pool
