> # 內容 * [內容](http://www.hmoore.net/book/xiaohuamao/go100/edit#_0) * [G的狀態](http://www.hmoore.net/book/xiaohuamao/go100/edit#G_4) * [P的狀態](http://www.hmoore.net/book/xiaohuamao/go100/edit#P_29) * [GMP調度模型](http://www.hmoore.net/book/xiaohuamao/go100/edit#GMP_40) * [GMP數據結構](http://www.hmoore.net/book/xiaohuamao/go100/edit#GMP_57) * [協程的調度不是隨機的](http://www.hmoore.net/book/xiaohuamao/go100/edit#_396) * [CSP 模型](http://www.hmoore.net/book/xiaohuamao/go100/edit#CSP__431) * [協程 (捕獲異常 和 協程池)](http://www.hmoore.net/book/xiaohuamao/go100/edit#____436) * [channel 數據結構](http://www.hmoore.net/book/xiaohuamao/go100/edit#channel__535) * [操作 nil channel, close channel, 正常 channel](http://www.hmoore.net/book/xiaohuamao/go100/edit#_nil_channel_close_channel__channel_556) * [channel 發送和接收數據的本質](http://www.hmoore.net/book/xiaohuamao/go100/edit#channel__563) * [range channel](http://www.hmoore.net/book/xiaohuamao/go100/edit#range_channel_581) * [使用 select 來多路復用 channel](http://www.hmoore.net/book/xiaohuamao/go100/edit#_select__channel_606) * [如何優雅的關閉 channel](http://www.hmoore.net/book/xiaohuamao/go100/edit#_channel_654) * [交替打印](http://www.hmoore.net/book/xiaohuamao/go100/edit#_789) > # G的狀態 ~~~ const ( _Gidle = iota // 0: 剛創建但尚未初始化，未分配棧空間 _Grunnable // 1: 已就緒(創建后/被喚醒時)，等待調度執行，位于運行隊列 _Grunning // 2: 正正在運行，綁定到M和P，其棧被鎖定，無法被GC或其他線程訪問 _Gsyscall // 3: 正在執行系統調用(文件I/O、網絡I/O、線程同步操作等)，處于內核態，棧被鎖定，不在運行隊列 _Gwaiting // 4: 被阻塞，等待資源（如通道、鎖、網絡等），不在運行隊列 _Gmoribund_unused // 5: 未使用狀態（保留，占位符） _Gdead // 6: 已結束生命周期 _Genqueue_unused // 7: 未使用狀態（保留，占位符） _Gcopystack // 8: 棧正在動態擴容/收縮，禁止調度(每G的棧初始大小為2KB,當棧空間不足時會分配更大的棧，并將現有的棧內容復制到新棧中) _Gpreempted // 9: 被調度器搶占暫停，等待重新調度, 在運行隊列 // GC 掃描相關標志 _Gscan = 0x1000 // 標志位，用于疊加在其他狀態上，表示 GC 正在掃描 Goroutine 棧 _Gscanrunnable = _Gscan + _Grunnable // 0x1001: GC 正在掃描運行隊列中的 Goroutine 的棧 _Gscanrunning = _Gscan + _Grunning // 0x1002: GC 正在掃描正在運行的 Goroutine 的棧 _Gscansyscall = _Gscan + _Gsyscall // 0x1003: GC 正在掃描處于系統調用中的 Goroutine 的棧 _Gscanwaiting = _Gscan + _Gwaiting // 0x1004: GC 正在掃描處于等待狀態的 Goroutine 的棧 _Gscanpreempted = _Gscan + _Gpreempted // 0x1009: GC 正在掃描被搶占暫停的 Goroutine 的棧 ) ~~~ > # P的狀態 ~~~ const ( _Pidle = iota // 0: 空閑狀態，未被 M 使用，未運行用戶代碼或調度工作 _Prunning // 1: 正在運行，被 M 持有，運行用戶代碼或調度 Goroutine _Psyscall // 2: 與執行系統調用的 M 關聯，暫不運行用戶代碼 _Pgcstop // 3: 暫停狀態，GC 的 STW 階段被停止 _Pdead // 4: 不可用狀態，通常因 GOMAXPROCS 設置減少 ) ~~~ > # GMP調度模型 * 用戶級線程(G):內核級線程(M) 多對多模型 * 數量 * P的數量：默認情況下P的數量與邏輯CPU核心數相同, 通過runtime.GOMAXPROCS(n)來修改, 通過runtime.NumCPU()查看 * M和G數量:由調度器動態管理, 通過runtime.NumGoroutine()查看G數量, M的最大數量:sched.maxmcount = 10000 * 流程 * 創建G:先加入P的本地隊列, 如果本地隊列已滿, 則加入到全局隊列 * 獲取G:優先從P的本地運列獲取G, 如果本地隊列為空,從全局隊列獲取G, 如果全局隊列也為空,從其他P的本地隊列竊取任務 * 執行G: * \_Gsyscall時: 不在運行隊列, 當前的P釋放M,綁定新的M繼續調度, 舊M上調度的舊G執行完后, 舊G加入某個P的本地隊列或全局隊列,舊M限制的時候會加入全局空閑隊列(舊M在沒有執行完系統調用前不會被其它P綁定) * \_Gwaiting時:不在運行隊列, 釋放G的綁定, 單獨的G等待事件完成后喚醒, 然后加個某個P的本地隊列或全局隊列 * \_Gpreempted時:在運行隊列,被搶占的 G因為時間片用盡或調度器的其他策略，暫時讓出 CPU * \_Gdead時: 當前的G不需要執行,銷毀  （圖片來源：[https://github.com/aceld/golang/blob/main/images/18-go-func調度周期.jpeg](https://github.com/aceld/golang/blob/main/images/18-go-func%E8%B0%83%E5%BA%A6%E5%91%A8%E6%9C%9F.jpeg)) > # GMP數據結構 * G:任務同單位, 通過m字段綁定所屬的M * P:持有G隊列, 通過m字段綁定所屬的M * M: curg當前的G, p字段 當前的P * schedt:管控全局資源分配，調度行為，以及 GC 協調 ~~~ type g struct { // 棧相關字段 stack stack // Goroutine 棧的范圍：[stack.lo, stack.hi) stackguard0 uintptr // 檢測棧溢出的基準指針，通常為 stack.lo + StackGuard stackguard1 uintptr // 特殊棧的基準指針（如 g0 和 gsignal 的系統棧） // 異常處理 _panic *_panic // 指向當前 Goroutine 最近的 panic 信息，用于異常恢復 _defer *_defer // 指向當前 Goroutine 最近的 defer 結構，用于延遲調用管理 // 調度相關 m *m // 當前 Goroutine 所屬的線程 m sched gobuf // Goroutine 的寄存器上下文信息，保存執行狀態 atomicstatus atomic.Uint32 // Goroutine 的狀態，使用原子操作管理（如 _Grunning、_Gwaiting 等） goid uint64 // Goroutine 唯一標識符 schedlink guintptr // 調度鏈表指針，用于 Goroutine 的隊列操作 waitsince int64 // 阻塞時間戳，用于監控 Goroutine 的等待時間 waitreason waitReason // 阻塞原因（如通道操作、鎖等待等） // 搶占與棧調整 preempt bool // 是否觸發搶占，通常由調度器根據時間片或棧檢查觸發 preemptStop bool // 是否在搶占后進入 _Gpreempted 狀態 preemptShrink bool // 是否在安全點縮減棧空間 asyncSafePoint bool // 是否停在異步安全點，棧中可能存在不精確指針 paniconfault bool // 遇到異常地址時是否 panic（而不是崩潰） gcscandone bool // 是否已掃描過當前 Goroutine 的棧，避免重復掃描 throwsplit bool // 是否禁止棧分裂，某些 Goroutine 可能要求完整棧 activeStackChans bool // 是否有指向該 Goroutine 棧的未鎖定通道，影響棧復制 // 通道與同步狀態 parkingOnChan atomic.Bool // 當前 Goroutine 是否停在通道操作上 inMarkAssist bool // 是否參與 GC 標記輔助工作 coroexit bool // 是否在協程切換時退出 // 性能統計與狀態 raceignore int8 // 是否忽略競態檢測，用于屏蔽特定 Goroutine 的事件 nocgocallback bool // 是否禁用從 C 回調到 Go tracking bool // 是否追蹤此 Goroutine 的調度延遲 trackingSeq uint8 // 調度追蹤序列號 trackingStamp int64 // 上次開始追蹤的時間戳 runnableTime int64 // 累計的可運行時間，僅用于性能追蹤 lockedm muintptr // 鎖定的線程 m，表示 Goroutine 被綁定到特定線程 // 信號處理 sig uint32 // 接收到的信號值 writebuf []byte // 用于信號處理的緩沖區 sigcode0 uintptr // 信號的附加信息 sigcode1 uintptr // 信號的附加信息 sigpc uintptr // 觸發信號的程序計數器 // Goroutine 關系 parentGoid uint64 // 父 Goroutine 的 ID gopc uintptr // 創建此 Goroutine 的 `go` 指令的地址 ancestors *[]ancestorInfo // 祖先信息（用于調試） startpc uintptr // Goroutine 執行的入口函數地址 // 性能調試與分析 racectx uintptr // 競態檢測上下文，用于調試競態條件 waiting *sudog // 當前阻塞的 sudog 指針 cgoCtxt []uintptr // Cgo 調用棧上下文信息 labels unsafe.Pointer // 分析器標簽，用于性能統計 timer *timer // 與當前 Goroutine 關聯的定時器 sleepWhen int64 // 睡眠的截止時間（Unix 時間戳） selectDone atomic.Uint32 // 是否參與了 select 操作，以及是否完成 // 性能分析與協程狀態 goroutineProfiled goroutineProfileStateHolder // 當前 Goroutine 的性能分析狀態 coroarg *coro // 協程切換時的參數 trace gTraceState // 當前 Goroutine 的跟蹤信息 // GC 輔助 gcAssistBytes int64 // 當前 Goroutine 的 GC 輔助字節計數，為負值時需參與垃圾回收 } type m struct { // 基礎信息 g0 *g // 默認 Goroutine（系統級 g0，用于調度和運行時管理） morebuf gobuf // 當前 Goroutine 調度的上下文緩沖區 divmod uint32 // ARM 架構的除法/取模分母（liblink 使用） _ uint32 // 對齊填充字段 // 線程及信號相關 procid uint64 // 操作系統線程 ID（供調試器使用） gsignal *g // 用于信號處理的 Goroutine goSigStack gsignalStack // Go 分配的信號處理棧 sigmask sigset // 保存的信號屏蔽集 tls [tlsSlots]uintptr // 線程局部存儲數據 // 啟動及當前 Goroutine 狀態 mstartfn func() // m 啟動時的函數 curg *g // 當前正在運行的 Goroutine catchsig guintptr // 捕獲信號的 Goroutine（在致命信號期間運行） // 調度器相關 p puintptr // 當前線程綁定的 P nextp puintptr // 即將綁定的 P oldp puintptr // 上次綁定的 P（如從系統調用恢復時） id int64 // m 的唯一標識符 mallocing int32 // 是否正在進行內存分配 throwing throwType // 是否正在執行異常（panic）操作 preemptoff string // 禁止搶占的原因（為空則允許搶占） locks int32 // 當前 m 持有的鎖數量 dying int32 // 是否標記為銷毀狀態 profilehz int32 // 性能分析的采樣頻率 spinning bool // 是否處于自旋狀態（等待任務） blocked bool // 是否阻塞在某個通知（note） newSigstack bool // 是否初始化了新的信號棧 printlock int8 // 打印調試日志時的鎖 incgo bool // 是否正在執行 cgo 調用 isextra bool // 是否為額外線程（輔助線程） isExtraInC bool // 是否為 C 中的輔助線程 isExtraInSig bool // 是否為信號處理中的輔助線程 freeWait atomic.Uint32 // 是否可以安全地釋放 g0 并刪除 m needextram bool // 是否需要額外的 m traceback uint8 // 是否啟用追蹤調試 ncgocall uint64 // 累計 cgo 調用次數 ncgo int32 // 當前進行中的 cgo 調用數量 cgoCallersUse atomic.Uint32 // 臨時標記 cgoCallers 是否在使用 cgoCallers *cgoCallers // cgo 堆棧跟蹤 park note // 用于同步的等待通知 alllink *m // 全局 m 鏈表中的下一個 m schedlink muintptr // 調度器鏈表中的下一個 m lockedg guintptr // 當前鎖定的 Goroutine createstack [32]uintptr // 創建此線程的調用棧（用于調試） lockedExt uint32 // 外部鎖定計數（用于 LockOSThread） lockedInt uint32 // 內部鎖定計數 nextwaitm muintptr // 等待鎖的下一個 m // 鎖與等待 mLockProfile mLockProfile // 鎖性能分析信息 profStack []uintptr // 用于內存、阻塞或互斥鎖的棧跟蹤 waitunlockf func(*g, unsafe.Pointer) bool // 解鎖時的回調函數 waitlock unsafe.Pointer // 等待鎖的指針 waitTraceSkip int // 跟蹤時跳過的棧幀數 waitTraceBlockReason traceBlockReason // 阻塞的具體原因 // 系統調用 syscalltick uint32 // 系統調用的計時器 freelink *m // 已釋放 m 鏈表的下一項 trace mTraceState // m 的跟蹤狀態 // 庫調用 libcall libcall // 庫調用信息 libcallpc uintptr // 庫調用的程序計數器 libcallsp uintptr // 庫調用的棧指針 libcallg guintptr // 執行庫調用的 Goroutine winsyscall winlibcall // Windows 系統調用參數 // VDSO（虛擬動態共享對象） vdsoSP uintptr // VDSO 調用的棧指針 vdsoPC uintptr // VDSO 調用的程序計數器 // 搶占與信號 preemptGen atomic.Uint32 // 已完成的搶占信號計數 signalPending atomic.Uint32 // 是否有待處理的搶占信號 // 緩存 pcvalueCache pcvalueCache // PC 值查找緩存 // 隨機數 chacha8 chacha8rand.State // ChaCha8 隨機數生成器狀態 cheaprand uint64 // 簡易隨機數生成器 // 鎖信息 locksHeldLen int // 持有的鎖數量 locksHeld [10]heldLockInfo // 持有的鎖的詳細信息（最多記錄 10 個） } type p struct { id int32 // P 的唯一標識符 status uint32 // P 的狀態（pidle、prunning 等） link puintptr // 鏈接到下一個空閑的 P schedtick uint32 // 每次調度器調用時遞增 syscalltick uint32 // 每次系統調用時遞增 sysmontick sysmontick // sysmon 上次觀察到的計數 m muintptr // 關聯的 M（空閑時為 nil） mcache *mcache // P 的內存緩存 pcache pageCache // 頁面緩存 raceprocctx uintptr // Race Detector 的處理上下文 // 延遲池 deferpool []*_defer // 可用的延遲結構池 deferpoolbuf [32]*_defer // 緩存用于延遲結構池的緩沖區 // Goroutine ID 緩存 goidcache uint64 // 緩存的 Goroutine ID 起始值 goidcacheend uint64 // 緩存的 Goroutine ID 結束值 // 可運行的 Goroutine 隊列 runqhead uint32 // 可運行隊列的頭部 runqtail uint32 // 可運行隊列的尾部 runq [256]guintptr // 可運行隊列，存儲 Goroutine 的指針 // runnext 是當前 Goroutine 已準備好下一個運行的 Goroutine // 它會繼承當前 Goroutine 的剩余時間片，優化調度延遲。 runnext guintptr // 已死亡的 Goroutine 列表 gFree struct { gList n int32 // 列表中 Goroutine 的數量 } sudogcache []*sudog // Sudog 對象的緩存 sudogbuf [128]*sudog // Sudog 緩存的緩沖區 // 堆上的 mspan 對象緩存 mspancache struct { len int // 緩存的 mspan 對象數量 buf [128]*mspan // 緩存的 mspan 對象 } // Pinner 對象緩存，用于減少重復創建的分配 pinnerCache *pinner trace pTraceState // 跟蹤狀態 palloc persistentAlloc // 持久性分配，用于避免互斥鎖 // 垃圾回收 (GC) 狀態 gcAssistTime int64 // assistAlloc 的時間（納秒） gcFractionalMarkTime int64 // fractional mark worker 的時間（納秒） limiterEvent limiterEvent // GC CPU 限制器的事件 gcMarkWorkerMode gcMarkWorkerMode // 下一個標記工作線程的模式 gcMarkWorkerStartTime int64 // 最近標記工作線程開始的時間 gcw gcWork // P 的 GC 工作緩存 wbBuf wbBuf // P 的 GC 寫屏障緩沖區 runSafePointFn uint32 // 如果為 1，則在下一個安全點運行 sched.safePointFn statsSeq atomic.Uint32 // P 的統計信息寫入狀態（偶數：未寫入，奇數：寫入中） // 定時器堆 timers timers // 累積的 goroutine 棧掃描大小 maxStackScanDelta int64 // 棧掃描增量（觸發 gcController.maxStackScan 時刷新） scannedStackSize uint64 // GC 時間掃描的棧大小 scannedStacks uint64 // GC 時間掃描的 Goroutine 數量 preempt bool // 標志是否需要盡快進入調度器 gcStopTime int64 // 上次進入 _Pgcstop 的時間戳 } type schedt struct { goidgen atomic.Uint64 // 全局 Goroutine ID 生成器 lastpoll atomic.Int64 // 上次網絡輪詢的時間，0 表示當前正在輪詢 pollUntil atomic.Int64 // 當前輪詢的睡眠時間 lock mutex // 調度器全局鎖 // M 相關狀態 midle muintptr // 空閑 M 列表，等待工作的 M nmidle int32 // 空閑 M 的數量 nmidlelocked int32 // 被鎖定且空閑的 M 數量 mnext int64 // 已創建的 M 數量及下一個 M 的 ID maxmcount int32 // 允許的最大 M 數量 nmsys int32 // 系統級別的 M 數量，不計入死鎖檢測 nmfreed int64 // 已釋放的 M 總數 ngsys atomic.Int32 // 系統 Goroutine 的數量 // P 相關狀態 pidle puintptr // 空閑 P 列表 npidle atomic.Int32 // 空閑 P 的數量 nmspinning atomic.Int32 // 當前自旋中的 M 數量 needspinning atomic.Uint32 // 是否需要更多自旋 M 的標志，需持有 sched.lock 才可設置 // 全局可運行的 Goroutine 隊列 runq gQueue // 全局隊列 runqsize int32 // 隊列大小 // 調度器禁用控制 disable struct { user bool // 是否禁用用戶 Goroutine 調度 runnable gQueue // 等待調度的 Goroutine 隊列 n int32 // 隊列長度 } // 全局 G 緩存 gFree struct { lock mutex // 緩存鎖 stack gList // 帶棧的 G 列表 noStack gList // 無棧的 G 列表 n int32 // 緩存的 G 數量 } // sudog 對象的全局緩存 sudoglock mutex sudogcache *sudog // 延遲對象的全局緩存 deferlock mutex deferpool *_defer // 等待釋放的 M 列表，通過 m.freelink 鏈接 freem *m // GC 相關狀態 gcwaiting atomic.Bool // 是否正在等待 GC stopwait int32 // 停止等待的計數器 stopnote note // 停止的通知 sysmonwait atomic.Bool // 系統監控是否等待中 sysmonnote note // 系統監控通知 // 下一個 GC 安全點需執行的函數 safePointFn func(*p) // 每個 P 在安全點需要運行的函數 safePointWait int32 // 等待安全點的計數器 safePointNote note // 安全點的通知 profilehz int32 // CPU 性能分析的頻率 // GOMAXPROCS 調整的時間統計 procresizetime int64 // 上次調整 GOMAXPROCS 的時間（納秒） totaltime int64 // 累計 GOMAXPROCS 的積分時間 sysmonlock mutex // 系統監控鎖，阻止系統監控與 runtime 的交互 // 調度延遲分布 timeToRun timeHistogram // G 在 _Grunnable 到 _Grunning 狀態的延遲分布 // P 的空閑時間 idleTime atomic.Int64 // 當前周期的空閑時間，GC 周期重置 // Goroutine 互斥鎖等待時間 totalMutexWaitTime atomic.Int64 // Goroutine 在 _Gwaiting 狀態下等待鎖的總時間 // 停止世界（STW）延遲分布 stwStoppingTimeGC timeHistogram // GC 相關的 STW 停止延遲 stwStoppingTimeOther timeHistogram // 非 GC 的 STW 停止延遲 // STW 總延遲分布 stwTotalTimeGC timeHistogram // GC 相關的 STW 總延遲 stwTotalTimeOther timeHistogram // 非 GC 的 STW 總延遲 // 運行時鎖等待時間 totalRuntimeLockWaitTime atomic.Int64 // G 在 _Grunnable 狀態下等待 runtime 鎖的總時間 } ~~~ > # 協程的調度不是隨機的 * Go 協程調度**不是隨機的**，但它也**不是按順序**的。它是一種基于信號的搶占式、工作竊取的調度模型 * 調度器的目的是在多個協程之間公平分配 CPU 資源，并在必要時暫停某些協程，讓其他協程有機會運行 ~~~ package main import ( "fmt" "time" ) func main() { ch := make(chan struct{}) for i := 0; i < 10; i++ { go func(num int) { for { <-ch fmt.Println(num) } }(i) time.Sleep(time.Millisecond) } time.Sleep(time.Second) for j := 0; j < 10; j++ { ch <- struct{}{} //**不加 `time.Sleep(time.Millisecond)`** 時，多個 Goroutine 幾乎同時啟動，調度器會隨機選擇哪個 Goroutine 先得到 CPU，因此打印順序不確定。 //**加了 `time.Sleep(time.Millisecond)`** 時，每個 Goroutine 啟動后，主協程會休眠 1 毫秒。這個時間足夠讓調度器有機會依次啟動每個 Goroutine，導致它們的打印順序更加有序 //time.Sleep(time.Millisecond) 改變輸出結果 } time.Sleep(time.Minute) } ~~~ > # CSP 模型 * **CSP**（Communicating Sequential Processes，通信順序進程） 是一種并發編程模型，其核心理念是**不要通過共享內存來通信，而要通過通信來實現內存共享**。 * 在 Go 語言中，CSP 模型主要通過**Goroutine**和**Channel**來實現。多個 Goroutine 之間的通信通常使用**Channel**，從而避免了共享內存導致的并發問題。 > # 協程 (捕獲異常 和 協程池) * 直接用go關鍵字開協程，不捕獲異常的話， 如果出現異常，會導致整個程序結束 * Go 語言中的**Goroutine**相較于系統線程來說非常輕量級，其初始棧大小僅為**2KB**。然而，在高并發場景下，大量的 Goroutine 被頻繁創建和銷毀，可能會對性能產生負面影響，并增加**GC（垃圾回收）**的壓力。 * 為了減少 Goroutine 的創建和銷毀所帶來的性能損耗，建議充分**復用 Goroutine**。通過使用**Goroutine 池**或者其他方式來復用已經存在的 Goroutine，可以有效地降低系統的開銷 * goroutine.go ~~~ package main import ( "fmt" ) // WorkerPool 定義一個工作池結構體 type WorkerPool struct { maxWorkers int taskQueue chan func() } // NewWorkerPool 創建一個新的工作池 func NewWorkerPool(maxWorkers int) *WorkerPool { return &WorkerPool{ maxWorkers: maxWorkers, taskQueue: make(chan func()), } } // Start 啟動工作池 func (wp *WorkerPool) Start() { for i := 0; i < wp.maxWorkers; i++ { go wp.worker() } } // worker 執行任務的工作者 goroutine func (wp *WorkerPool) worker() { for task := range wp.taskQueue { safeExecute(task) } } // safeExecute 安全執行任務，捕獲異常 func safeExecute(task func()) { defer func() { if r := recover(); r != nil { fmt.Println("Recovered from panic:", r) } }() task() } // Submit 提交任務到工作池 func (wp *WorkerPool) Submit(task func()) { wp.taskQueue <- task } ~~~ * main.go ~~~ package main import ( "fmt" "time" ) var pool *WorkerPool func init() { pool = NewWorkerPool(20) pool.Start() } func SafeGo(f func()) { pool.Submit(f) } func main() { for i := 0; i < 10; i++ { SafeGo(func(num int) func() { return func() { fmt.Println("A", num) } }(i)) } for i := 0; i < 10; i++ { SafeGo(func(num int) func() { return func() { fmt.Println("B", num) } }(i)) } time.Sleep(time.Second * 3) } ~~~ > # channel 數據結構 * chan T // 可以接收和發送類型為 T 的數據 * chan<- T // 只可以用來發送 T 類型的數據 * <-chan T // 只可以用來接收 T 類型的數據 ~~~ type hchan struct { qcount uint // 通道中當前元素個數 dataqsiz uint // 緩沖區的大小 (無緩沖時為0) buf unsafe.Pointer // (環形緩沖區)指向緩沖區的指針,緩沖區通過這個指針來存儲數據，已經發送但還未被接收的數據 elemsize uint16 // 單個元素的大小（如果通道是 chan int 類型，那么每個元素就是一個 int，elemsize 表示 int 類型的字節大小。） closed uint32 // 標識 channel 是否已關閉 (0 表示未關閉，1 表示已關閉) timer *timer // 用于處理與該通道相關的超時操作 elemtype *_type // 指向類型描述符的指針，它包含了通道中傳輸數據類型的所有信息 sendx uint // 下一個要發送元素的位置 recvx uint // 下一個要接收元素的位置 (每次讀取數據后，recvx 都會遞增，當到達緩沖區末尾時，recvx 會重置為 0，形成環形讀取操作) recvq waitq // 等待接收數據的 goroutine 隊列 sendq waitq // 等待發送數據的 goroutine 隊列 lock mutex // 互斥鎖，用于保護 channel 的操作 } ~~~ > # 操作 nil channel, close channel, 正常 channel | 操作 | nil channel | close channel | 正常 channel | | --- | --- | --- | --- | | close | panic: close of nil channel | panic: close of closed channel | 正常關閉 | | 讀操作 | fatal error: all goroutines are asleep - deadlock! | 有未接收的值可以正常讀， 沒有的話讀到對應類型的零值 | 阻塞/正常讀數據 | | 寫操作 | fatal error: all goroutines are asleep - deadlock! | panic: send on closed channel | 阻塞/正常寫數據 | > # channel 發送和接收數據的本質 ~~~ //向 channel 發送值類型會拷貝， 發送引用類型拷貝的是引用 package main import "fmt" func main() { ch := make(chan []int, 1) s := make([]int, 1) s[0] = 1 ch <- s s[0] = 2 fmt.Println(<-ch) //輸出2 } ~~~ > # range channel ~~~ package main import ( "fmt" "time" ) func main() { ch := make(chan int) go func() { time.Sleep(time.Second * 10) close(ch) }() //如果通道沒有關閉，但生產者沒有再發送數據，range 會持續阻塞，直到有新的數據發送到通道或通道被關閉 for v := range ch { fmt.Println(v) } fmt.Println("Done") } ~~~ > # 使用 select 來多路復用 channel ~~~ package main import ( "fmt" "time" ) // 隨機選擇：當多個 channel 同時滿足條件時，select 會隨機選擇一個執行。(一個channel先準備,另一個channel后準備好,select也是隨機選擇) // 默認分支：可以在 select 中添加 default 分支，當所有的 channel 都沒有數據時，select 可以立即執行 default 分支而不阻塞。 func main() { ch1 := make(chan struct{}) ch2 := make(chan struct{}) go func() { ch1 <- struct{}{} }() go func() { ch2 <- struct{}{} }() go func() { for { select { case <-ch1: fmt.Println("ch1") time.Sleep(1 * time.Second) case <-ch2: fmt.Println("ch2") time.Sleep(1 * time.Second) case <-time.After(2 * time.Second): fmt.Println("超時") time.Sleep(1 * time.Second) default: fmt.Println("default") time.Sleep(1 * time.Second) } } }() time.Sleep(time.Second * 10) close(ch1) //通道關閉后，select 還會執行 time.Sleep(time.Second * 10) } ~~~ > # 如何優雅的關閉 channel ~~~ //channel關閉原則: 不要在消費端關閉channel,不要再生產端有多個并行的時候執行關閉操作 //判斷channel是否關閉: v, ok := <-ch 取值的時候加判斷, 關閉的channel, v 返回對應類型的零值, ok 返回 false。用這種方式去判斷channel 是否關閉有副作用, 會讀出channel里的元素 //如何優雅的關閉channel //--- 1.直接關閉channel用recover捕獲異常 //----2.sync.Once來確保channel只關閉一次 //--- 3.增加狀態字段存channel是否關閉(需要用到鎖) //--- 4.增加一個channel作為關閉信號(如下代碼一) //--- 5.使用context(如下代碼二) //用完的channel, 沒有發送者, 沒有接收者, 也沒有關閉會一直占用內存嗎? 會的 //已關閉的channel, 還有未接收的數據, 但是沒有發送者和接受者會一直占用內存嗎? 不會, 如果沒有任何引用指向已關閉的通道，它會被垃圾回收器回收，并釋放內存 //為什么go不提供一個函數來判斷 channel 是否關閉? 避免競爭條件(獲取的那一時刻沒關閉,后續關閉問題) ~~~ * 代碼1 ~~~ package main import ( "fmt" "sync" "time" ) func main() { ch := make(chan int, 100) chClose := make(chan struct{}) var ( wgSend sync.WaitGroup ) wgSend.Add(10) // 啟動生產者 for i := 0; i < 10; i++ { go func(num int) { defer wgSend.Done() for { select { case <-chClose: fmt.Println("發送關閉", num) return case ch <- num: } } }(i) } // 啟動消費者 for i := 0; i < 10; i++ { go func(num int) { for { select { case <-chClose: fmt.Println("接收關閉", num) return case v := <-ch: fmt.Println("接收到的數據", v) } } }(i) } time.Sleep(time.Second * 10) close(chClose) wgSend.Wait() close(ch) } ~~~ * 代碼2 ~~~ package main import ( "context" "fmt" "sync" "time" ) func main() { ch := make(chan int, 100) ctx, cancel := context.WithCancel(context.Background()) var ( wgSend sync.WaitGroup ) wgSend.Add(10) // 啟動生產者 for i := 0; i < 10; i++ { go func(ctx context.Context, num int) { defer wgSend.Done() for { select { case <-ctx.Done(): fmt.Println("發送關閉", num) return case ch <- num: } } }(ctx, i) } // 啟動消費者 for i := 0; i < 10; i++ { go func(ctx context.Context, num int) { for { select { case <-ctx.Done(): fmt.Println("接收關閉", num) return case v := <-ch: fmt.Println("接收到的數據", v) } } }(ctx, i) } time.Sleep(time.Second * 10) cancel() // 通知所有協程關閉 wgSend.Wait() close(ch) } ~~~ > # 交替打印 ~~~ package main import ( "fmt" "time" ) func main() { //無緩沖通道的特點是發送和接收必須同時發生，否則操作會阻塞(不能是ch := make(chan struct{}, 1),有緩沖了會出現搶占) ch := make(chan struct{}) go func() { for { <-ch fmt.Println(1) ch <- struct{}{} } }() go func() { for { <-ch fmt.Println(2) ch <- struct{}{} } }() ch <- struct{}{} time.Sleep(time.Hour) } ~~~