本文基于golang 1.13版本分析。
一、前言
1.1 slice 結構
slice實際就是一個struct，在runtime/slice.go中的定義如下：
type slice struct {
array unsafe.Pointer
len int
cap int
}
// An notInHeapSlice is a slice backed by go:notinheap memory.
type notInHeapSlice struct {
array *notInHeap
len int
cap int
}
由定義可以看出slice底層是基于數(shù)組，本質(zhì)是對數(shù)組的封裝。由三部分組成：
-
指針，指向第一個slice元素對應的底層數(shù)組元素地址。 -
長度，表示切片可用元素的個數(shù)，也就是說使用下標對 slice 的元素進行訪問時，下標不能超過 slice 的長度； -
容量，底層數(shù)組的元素個數(shù)，容量 >= 長度。在底層數(shù)組不進行擴容的情況下，容量也是 slice 可以擴張的最大限度。
slice
內(nèi)置函數(shù)len和cap，分別返回slice的長度和容量。slice使用下標不能超過len，向后擴展不能超過cap。多個不同slice之間可以共享底層的數(shù)據(jù)，起始地址、長度都可以不同，所以slice第一個元素未必是數(shù)組的第一個元素。
1.2 使用中的坑
介紹完slice的數(shù)據(jù)結構，作者想把可能遇到的坑寫在最前，避免后來者持續(xù)踩坑。
1.2.1 slice作為函數(shù)入?yún)?/h5>
函數(shù)內(nèi)append slice，修改的僅僅是函數(shù)中slice的len，外面的slice len不會發(fā)生變化。
函數(shù)內(nèi)append slice，修改的僅僅是函數(shù)中slice的len，外面的slice len不會發(fā)生變化。
對切片進行append的時候，如果底層空間足夠就使用原來的空間，如果底層空間不夠，那么就會申請新的空間。函數(shù)傳遞切片時，是值傳遞，不是引用傳遞，傳遞的是slice結構體那三個字段的值，所以不會復制slice的實際內(nèi)容，在函數(shù)內(nèi)append時，修改的僅僅是函數(shù)中slice的len／cap，外面的slice len／cap不會發(fā)生變化。
1.2.2 slice初始化make與append使用
使用make([]int,len(m))初始化時，已經(jīng)賦值零值，萬萬不能配合append使用。
func main() {
m := map[int]int{1:10,2:20}
//=======錯誤的做法=============
a := make([]int,len(m))
for _,v := range m{
a = append(a,v)
}
// [0 0 10 20]
fmt.Println(a)
//=======正確的做法=============
b := make([]int,0,len(m))
for _,v := range m{
b = append(b,v)
}
//[10 20]
fmt.Println(b)
c := make([]int,len(m))
i := 0
for _,v := range m{
c[i] = v
i++
}
//[10 20]
fmt.Println(c)
var d []int
for _,v := range m{
d = append(d,v)
}
//[10 20]
fmt.Println(d)
}
1.2.3 slice nil值與空值
slice nil值與空值不相等
var s []int //nil值
var t = []int{} //空值
var u = make([]int, 3)[3:] //空值
fmt.Printf("value of s: %#v\n", s) // value of s: []int(nil)
fmt.Printf("value of t: %#v\n", t) // value of t: []int{}
fmt.Printf("value of u: %#v\n", u) //value of u: []int{}
fmt.Printf("s is nil? %v\n", s == nil) //true
fmt.Printf("t is nil? %v\n", t == nil) //false
fmt.Printf("u is nil? %v\n", u == nil) //false
區(qū)別是，nil slice的底層數(shù)組指針是nil，empty slice底層數(shù)組指針指向一個長度為0的數(shù)組。
所以測試一個slice是否有數(shù)據(jù)，使用len(s) == 0來判斷，而不應用s == nil來判斷。
一般的用法是nil slice表示數(shù)組不存在，empty slice表示集合為空。序列化json的時候，nil slice會變成null，empty是[]
二、源碼解析
2.1 slice 創(chuàng)建
創(chuàng)建 slice 的方式有以下幾種：
2.1.1 make
func makeslice(et *_type, len, cap int) unsafe.Pointer {
mem, overflow := math.MulUintptr(et.size, uintptr(cap))
if overflow || mem > maxAlloc || len < 0 || len > cap {
// NOTE: Produce a 'len out of range' error instead of a
// 'cap out of range' error when someone does make([]T, bignumber).
// 'cap out of range' is true too, but since the cap is only being
// supplied implicitly, saying len is clearer.
// See golang.org/issue/4085.
mem, overflow := math.MulUintptr(et.size, uintptr(len))
if overflow || mem > maxAlloc || len < 0 {
panicmakeslicelen()
}
panicmakeslicecap()
}
return mallocgc(mem, et, true)
}
先來一小段玩具代碼，使用 make 關鍵字創(chuàng)建 slice：
package main
import "fmt"
func main() {
slice := make([]int, 5, 10) // 長度為5，容量為10
slice[2] = 2 // 索引為2的元素賦值為2
fmt.Println(slice)
}
執(zhí)行如下命令，得到 Go 匯編代碼：
go tool compile -S main.go
我們只關注main函數(shù)：
0x0000 00000 (main.go:5)TEXT "".main(SB), $96-0
0x0000 00000 (main.go:5)MOVQ (TLS), CX
0x0009 00009 (main.go:5)CMPQ SP, 16(CX)
0x000d 00013 (main.go:5)JLS 228
0x0013 00019 (main.go:5)SUBQ $96, SP
0x0017 00023 (main.go:5)MOVQ BP, 88(SP)
0x001c 00028 (main.go:5)LEAQ 88(SP), BP
0x0021 00033 (main.go:5)FUNCDATA $0, gclocals·69c1753bd5f81501d95132d08af04464(SB)
0x0021 00033 (main.go:5)FUNCDATA $1, gclocals·57cc5e9a024203768cbab1c731570886(SB)
0x0021 00033 (main.go:5)LEAQ type.int(SB), AX
0x0028 00040 (main.go:6)MOVQ AX, (SP)
0x002c 00044 (main.go:6)MOVQ $5, 8(SP)
0x0035 00053 (main.go:6)MOVQ $10, 16(SP)
0x003e 00062 (main.go:6)PCDATA $0, $0
0x003e 00062 (main.go:6)CALL runtime.makeslice(SB)
0x0043 00067 (main.go:6)MOVQ 24(SP), AX
0x0048 00072 (main.go:6)MOVQ 32(SP), CX
0x004d 00077 (main.go:6)MOVQ 40(SP), DX
0x0052 00082 (main.go:7)CMPQ CX, $2
0x0056 00086 (main.go:7)JLS 221
0x005c 00092 (main.go:7)MOVQ $2, 16(AX)
0x0064 00100 (main.go:8)MOVQ AX, ""..autotmp_2+64(SP)
0x0069 00105 (main.go:8)MOVQ CX, ""..autotmp_2+72(SP)
0x006e 00110 (main.go:8)MOVQ DX, ""..autotmp_2+80(SP)
0x0073 00115 (main.go:8)MOVQ $0, ""..autotmp_1+48(SP)
0x007c 00124 (main.go:8)MOVQ $0, ""..autotmp_1+56(SP)
0x0085 00133 (main.go:8)LEAQ type.[]int(SB), AX
0x008c 00140 (main.go:8)MOVQ AX, (SP)
0x0090 00144 (main.go:8)LEAQ ""..autotmp_2+64(SP), AX
0x0095 00149 (main.go:8)MOVQ AX, 8(SP)
0x009a 00154 (main.go:8)PCDATA $0, $1
0x009a 00154 (main.go:8)CALL runtime.convT2Eslice(SB)
0x009f 00159 (main.go:8)MOVQ 16(SP), AX
0x00a4 00164 (main.go:8)MOVQ 24(SP), CX
0x00a9 00169 (main.go:8)MOVQ AX, ""..autotmp_1+48(SP)
0x00ae 00174 (main.go:8)MOVQ CX, ""..autotmp_1+56(SP)
0x00b3 00179 (main.go:8)LEAQ ""..autotmp_1+48(SP), AX
0x00b8 00184 (main.go:8)MOVQ AX, (SP)
0x00bc 00188 (main.go:8)MOVQ $1, 8(SP)
0x00c5 00197 (main.go:8)MOVQ $1, 16(SP)
0x00ce 00206 (main.go:8)PCDATA $0, $1
0x00ce 00206 (main.go:8)CALL fmt.Println(SB)
0x00d3 00211 (main.go:9)MOVQ 88(SP), BP
0x00d8 00216 (main.go:9)ADDQ $96, SP
0x00dc 00220 (main.go:9)RET
0x00dd 00221 (main.go:7)PCDATA $0, $0
0x00dd 00221 (main.go:7)CALL runtime.panicindex(SB)
0x00e2 00226 (main.go:7)UNDEF
0x00e4 00228 (main.go:7)NOP
0x00e4 00228 (main.go:5)PCDATA $0, $-1
0x00e4 00228 (main.go:5)CALL runtime.morestack_noctxt(SB)
0x00e9 00233 (main.go:5)JMP 0
我們先從上到下掃一眼，看到幾個關鍵函數(shù)：
CALL runtime.makeslice(SB)
CALL runtime.convT2Eslice(SB)
CALL fmt.Println(SB)
CALL runtime.morestack_noctxt(SB)
- 1是創(chuàng)建 slice 相關的；
- 2是類型轉(zhuǎn)換；調(diào)用 fmt.Println需要將 slice 作一個轉(zhuǎn)換；
- 3是打印語句；
- 4是?？臻g擴容函數(shù)，在函數(shù)開始處，會檢查當前?？臻g是否足夠，不夠的話需要調(diào)用它來進行擴容。
接下來，我們詳細分析這整個過程。
左邊是棧上的數(shù)據(jù)，右邊是堆上的數(shù)據(jù)。array 指向 slice 的底層數(shù)據(jù)，被分配到堆上了。注意，棧上的地址是從高向低增長；堆則從低向高增長。棧左邊的數(shù)字表示對應的匯編代碼的行數(shù)，棧右邊箭頭則表示棧地址。（48）SP、（56）SP 表示的內(nèi)容接著往下看。
注意，在圖中，棧地址是從下往上增長，所以 SP 表示的是圖中 *_type 所在的位置，其它的依此類推
convT2Eslice 的函數(shù)聲明如下：
func convT2Eslice(t *_type, elem unsafe.Pointer) (e eface)
第一個參數(shù)是指針 *_type，_type是一個表示類型的結構體，這里傳入的就是 slice的類型 []int；第二個參數(shù)則是元素的指針，這里傳入的就是 slice 底層數(shù)組的首地址。
返回值 eface 的結構體定義如下：
type eface struct {
_type *_type
data unsafe.Pointer
}
由于我們會調(diào)用 fmt.Println(slice)，看下函數(shù)原型：
func Println(a ...interface{}) (n int, err error)
Println 接收 interface 類型，因此我們需要將 slice 轉(zhuǎn)換成 interface 類型。由于 slice 沒有方法，是個“空 interface”。因此會調(diào)用 convT2Eslice 完成這一轉(zhuǎn)換過程。
convT2Eslice 函數(shù)返回的是類型指針和數(shù)據(jù)地址。源碼就不貼了，大體流程是：調(diào)用 mallocgc 分配一塊內(nèi)存，把數(shù)據(jù) copy 進到新的內(nèi)存，然后返回這塊內(nèi)存的地址，*_type 則直接返回傳入的參數(shù)。
32(SP) 和 40(SP) 其實是 makeslice 函數(shù)的返回值，這里可以忽略。
還剩 fmt.Println(slice) 最后一個函數(shù)調(diào)用了，我們繼續(xù)。
所以調(diào)用 fmt.Println(slice) 時，實際是傳入了一個 slice類型的eface地址。這樣，Println就可以訪問類型中的數(shù)據(jù)，最終給“打印”出來。
最后，我們看下 main 函數(shù)棧幀的開始和收尾部分。
0x0013 00019 (main.go:5)SUBQ $96, SP
0x0017 00023 (main.go:5)MOVQ BP, 88(SP)
0x001c 00028 (main.go:5)LEAQ 88(SP), BP
…………………………
0x00d3 00211 (main.go:9)MOVQ 88(SP), BP
0x00d8 00216 (main.go:9)ADDQ $96, SP
RET
BP可以理解為保存了當前函數(shù)棧幀棧底的地址，SP則保存棧頂?shù)牡刂贰?/p>
初始，BP 和 SP 分別有一個初始狀態(tài)。
main 函數(shù)執(zhí)行的時候，先根據(jù) main 函數(shù)棧幀大小確定 SP 的新指向，使得 main 函數(shù)棧幀大小達到 96B。之后把老的 BP 保存到 main 函數(shù)棧幀的底部，并使 BP 寄存器重新指向新的棧底，也就是 main 函數(shù)棧幀的棧底。
最后，當 main 函數(shù)執(zhí)行完畢，把它棧底的 BP 給回彈回到 BP 寄存器，恢復調(diào)用前的初始狀態(tài)。一切都像是沒有發(fā)生一樣，完美的現(xiàn)場。
2.2 slice 擴容
// growslice handles slice growth during append.
// It is passed the slice element type, the old slice, and the desired new minimum capacity,
// and it returns a new slice with at least that capacity, with the old data
// copied into it.
// The new slice's length is set to the old slice's length,
// NOT to the new requested capacity.
// This is for codegen convenience. The old slice's length is used immediately
// to calculate where to write new values during an append.
// TODO: When the old backend is gone, reconsider this decision.
// The SSA backend might prefer the new length or to return only ptr/cap and save stack space.
func growslice(et *_type, old slice, cap int) slice {
if raceenabled {
callerpc := getcallerpc()
racereadrangepc(old.array, uintptr(old.len*int(et.size)), callerpc, funcPC(growslice))
}
if msanenabled {
msanread(old.array, uintptr(old.len*int(et.size)))
}
if cap < old.cap {
panic(errorString("growslice: cap out of range"))
}
if et.size == 0 {
// append should not create a slice with nil pointer but non-zero len.
// We assume that append doesn't need to preserve old.array in this case.
return slice{unsafe.Pointer(&zerobase), old.len, cap}
}
newcap := old.cap
doublecap := newcap + newcap
if cap > doublecap {
newcap = cap
} else {
if old.len < 1024 {
newcap = doublecap
} else {
// Check 0 < newcap to detect overflow
// and prevent an infinite loop.
for 0 < newcap && newcap < cap {
newcap += newcap / 4
}
// Set newcap to the requested cap when
// the newcap calculation overflowed.
if newcap <= 0 {
newcap = cap
}
}
}
var overflow bool
var lenmem, newlenmem, capmem uintptr
// Specialize for common values of et.size.
// For 1 we don't need any division/multiplication.
// For sys.PtrSize, compiler will optimize division/multiplication into a shift by a constant.
// For powers of 2, use a variable shift.
switch {
case et.size == 1:
lenmem = uintptr(old.len)
newlenmem = uintptr(cap)
capmem = roundupsize(uintptr(newcap))
overflow = uintptr(newcap) > maxAlloc
newcap = int(capmem)
case et.size == sys.PtrSize:
lenmem = uintptr(old.len) * sys.PtrSize
newlenmem = uintptr(cap) * sys.PtrSize
capmem = roundupsize(uintptr(newcap) * sys.PtrSize)
overflow = uintptr(newcap) > maxAlloc/sys.PtrSize
newcap = int(capmem / sys.PtrSize)
case isPowerOfTwo(et.size):
var shift uintptr
if sys.PtrSize == 8 {
// Mask shift for better code generation.
shift = uintptr(sys.Ctz64(uint64(et.size))) & 63
} else {
shift = uintptr(sys.Ctz32(uint32(et.size))) & 31
}
lenmem = uintptr(old.len) << shift
newlenmem = uintptr(cap) << shift
capmem = roundupsize(uintptr(newcap) << shift)
overflow = uintptr(newcap) > (maxAlloc >> shift)
newcap = int(capmem >> shift)
default:
lenmem = uintptr(old.len) * et.size
newlenmem = uintptr(cap) * et.size
capmem, overflow = math.MulUintptr(et.size, uintptr(newcap))
capmem = roundupsize(capmem)
newcap = int(capmem / et.size)
}
// The check of overflow in addition to capmem > maxAlloc is needed
// to prevent an overflow which can be used to trigger a segfault
// on 32bit architectures with this example program:
//
// type T [1<<27 + 1]int64
//
// var d T
// var s []T
//
// func main() {
// s = append(s, d, d, d, d)
// print(len(s), "\n")
// }
if overflow || capmem > maxAlloc {
panic(errorString("growslice: cap out of range"))
}
var p unsafe.Pointer
if et.ptrdata == 0 {
p = mallocgc(capmem, nil, false)
// The append() that calls growslice is going to overwrite from old.len to cap (which will be the new length).
// Only clear the part that will not be overwritten.
memclrNoHeapPointers(add(p, newlenmem), capmem-newlenmem)
} else {
// Note: can't use rawmem (which avoids zeroing of memory), because then GC can scan uninitialized memory.
p = mallocgc(capmem, et, true)
if lenmem > 0 && writeBarrier.enabled {
// Only shade the pointers in old.array since we know the destination slice p
// only contains nil pointers because it has been cleared during alloc.
bulkBarrierPreWriteSrcOnly(uintptr(p), uintptr(old.array), lenmem)
}
}
memmove(p, old.array, lenmem)
return slice{p, old.len, newcap}
}
網(wǎng)上大多數(shù)的文章都是這樣描述的：
當原 slice 容量小于 1024 的時候，新 slice 容量變成原來的 2 倍；原 slice 容量超過 1024，新 slice 容量變成原來的1.25倍。
我在這里先說結論：以上描述是錯誤的。
比如下面的代碼，容量各自是多少呢？
package main
import "fmt"
func main() {
a := []byte{1, 0}
a = append(a, 1, 1, 1)
fmt.Println("cap of a is ",cap(a))
b := []int{23, 51}
b = append(b, 4, 5, 6)
fmt.Println("cap of b is ",cap(b))
c := []int32{1, 23}
c = append(c, 2, 5, 6)
fmt.Println("cap of c is ",cap(c))
type D struct{
age byte
name string
}
d := []D{
{1,"123"},
{2,"234"},
}
d = append(d,D{4,"456"},D{5,"567"},D{6,"678"})
fmt.Println("cap of d is ",cap(d))
}
應該是4個8？基于翻倍的思路，cap從2->4->8。
或者4個5？給4個5的猜測基于以下推測：如果在append多個元素的時候，一次擴容不足以滿足元素的放置，如果我是設計者，我會先預估好需要多少容量才可以放置元素，然后再進行一次擴容，好處就是，不需要頻繁申請新的底層數(shù)組，以及不需要頻繁的數(shù)據(jù)copy。
但是結果有點出人意料。
cap of a is 8
cap of b is 6
cap of c is 8
cap of d is 5
在發(fā)生append那一行代碼打上斷點，然后開始運行程序，為了比較好的說明情況，斷點打到擴容后容量為6的[]int型切片b的append上。
a. 傳進來的cap是5，也就是上文提及到的思路目前來看是正確的，當append多個元素的時候，先預估好容量再進行擴容。
b. slice是一個struct，而struct是值類型。
c. 用capmem進行內(nèi)存分配，然后將newcap作為新的slice的cap，我們來分析這一步capmem = roundupsize(uintptr(newcap) * sys.PtrSize)。
round-up，向上取整，roundupsize，向上取一個size。(uintptr(newcap) * sys.PtrSize)的乘積應該為5*8=40，經(jīng)過向上取整之后得到了新的所需內(nèi)存capmem=48，接著所需內(nèi)存/類型大小int(capmem / sys.PtrSize)，得到了新的容量，也就是6.
2.2.1 slice 內(nèi)存對齊
要明白roundupsize為什么會將40變?yōu)?8，這里需要簡單的引進go的內(nèi)存管理。
對象大小表，大體如下：
// class bytes/obj bytes/span objects tail waste max waste
// 1 8 8192 1024 0 87.50%
// 2 16 8192 512 0 43.75%
// 3 32 8192 256 0 46.88%
// 4 48 8192 170 32 31.52%
// 5 64 8192 128 0 23.44%
// 6 80 8192 102 32 19.07%
// 7 96 8192 85 32 15.95%
// 8 112 8192 73 16 13.56%
// 9 128 8192 64 0 11.72%
// 10 144 8192 56 128 11.82%
// ...
// 65 28672 57344 2 0 4.91%
// 66 32768 32768 1 0 12.50%
最小是8b，最大是32K，還有一類就是超出32K的，共67類（超出32K沒列在這個文件的，66+1=67）?？梢钥吹?，并沒有size為40的類型，于是40向上取整，取到了48，這就是發(fā)生在roundupsize的真相。這里有一個比較專業(yè)的名詞，內(nèi)存對齊。
三、并發(fā)
由于 slice/map 是引用類型，golang函數(shù)是傳值調(diào)用，所用參數(shù)副本依然是原來的 slice， 并發(fā)訪問同一個資源會導致竟態(tài)條件。
3.1 解決方案
3.1.1 方案 1: 加鎖
func main() {
slc := make([]int, 0, 1000)
var wg sync.WaitGroup
var lock sync.Mutex
for i := 0; i < 1000; i++ {
wg.Add(1)
go func(a int) {
defer wg.Done()
// 加鎖
lock.Lock()
defer lock.Unlock()
slc = append(slc, a)
}(i)
}
wg.Wait()
fmt.Println(len(slc))
}
優(yōu)點是比較簡單，適合對性能要求不高的場景。
3.1.2 方案 2: 使用 channel 串行化操作
type ServiceData struct {
ch chan int // 用來 同步的channel
data []int // 存儲數(shù)據(jù)的slice
}
func (s *ServiceData) Schedule() {
// 從 channel 接收數(shù)據(jù)
for i := range s.ch {
s.data = append(s.data, i)
}
}
func (s *ServiceData) Close() {
// 最后關閉 channel
close(s.ch)
}
func (s *ServiceData) AddData(v int) {
s.ch <- v // 發(fā)送數(shù)據(jù)到 channel
}
func NewScheduleJob(size int, done func()) *ServiceData {
s := &ServiceData{
ch: make(chan int, size),
data: make([]int, 0),
}
go func() {
// 并發(fā)地 append 數(shù)據(jù)到 slice
s.Schedule()
done()
}()
return s
}
func main() {
var (
wg sync.WaitGroup
n = 1000
)
c := make(chan struct{})
// new 了這個 job 后，該 job 就開始準備從 channel 接收數(shù)據(jù)了
s := NewScheduleJob(n, func() { c <- struct{}{} })
wg.Add(n)
for i := 0; i < n; i++ {
go func(v int) {
defer wg.Done()
s.AddData(v)
}(i)
}
wg.Wait()
s.Close()
<-c
fmt.Println(len(s.data))
}
實現(xiàn)相對復雜，優(yōu)點是性能很好，利用了channel的優(yōu)勢。
