在學習《The Go Programming Language》第八章並發單元的時候還是遭遇了不少問題，和值得總結思考和記錄的地方。
做一個類似於unix du命令的工具。但是閹割了一些功能，這里應該只實現-c(統計total大小) 和-h(以human比較容易辨識的顯示出來)的功能。
首先我們需要構造一個 能夠返回FileInfo信息數組的函數，我們把它取名為dirEntries:
func dirEntries(dir string) []os.FileInfo { entries, err := ioutil.ReadDir(dir) if err != nil { fmt.Fprintf(os.Stderr, "du: %v\n", err) return nil } return entries }
傳入一個路徑字符串，然后使用ioutil.ReadDir解析這個路徑下面的所有文件以及文件夾生成一個FileInfo的profile。
Fileinfo interface下面包含了：
type FileInfo interface { Name() string // base name of the file Size() int64 // length in bytes for regular files; system-dependent for others Mode() FileMode // file mode bits ModTime() time.Time // modification time IsDir() bool // abbreviation for Mode().IsDir() Sys() interface{} // underlying data source (can return nil) }
多種方法，可以直接調用，其作用就是后面注釋寫的一樣。
有了能夠獲取文件夾下面文件和文件夾的函數之后，我們需要一個調用方用來walk指定的目錄：
// 入參是一個文件目錄，一個INT64的只接收的單向channel func walkDir(dir string, fileSizes chan<- int64) { for _, entry := range dirEntries(dir) { if entry.IsDir() { subdir := filepath.Join(dir, entry.Name()) walkDir(subdir, fileSizes) } else { fileSizes <- entry.Size() } } }
這里我們定義一個目錄，然后需求傳入一個單向接收channel用於在多goroutine中計算總共的文件大小。
使用range方法來遍歷我們上面寫的dirEntries的返回文件或文件夾，如果是文件夾則繼續迭代。
如果不是則將文件大小存入放入fileSizes channel中。
搞定上面兩個函數，我們來寫主函數部分：
func main() { root := "" flag.StringVar(&root, "-p", ".", "input dir.") flag.Parse() fileSizes := make(chan int64) // 起一個goroutine去walk目錄 go func() { walkDir(root, fileSizes) // Walk完畢之后要關閉該channel下面使用range讀取數據的時候才會有盡頭 close(fileSizes) }() var nfiles, nbytes int64 for size := range fileSizes { nfiles++ nbytes += size } fmt.Printf("%d files %.1f GB\n", nfiles, float64(nbytes)/1e9) }
這里注意一點，因為起goroutine的walk函數，和下面同時在range遍歷是在同步進行，如果下面range速度太快讀到管道里面沒有值了會阻塞住等待有數據繼續進來之后讀取，而不是會跳出。只有當close(fileSizes)這句執行到，顯示關閉掉channel之后，才會跳出range循環並且這時已經讀取完了所有的數據。這里有點像，close channel的時候給range發送了一個停止信號一樣，感覺這個利用起來會比較有用？ 后續可能會再研究一下。
讓我們繼續來優化我們的程序，添加一個-v參數，打印出掃描文件的進度，當我們要掃描整個盤的時候，可能會花費大量的時間，我們需要知道進度如何了。
其實這個需求只需要很小的改動，讓我們來重新改寫一下main函數，用select多路復用來完成這個事情。
func main() { root := "" verbose := false tick := make(<-chan time.Time) var nfiles, nbytes int64 flag.StringVar(&root, "p", ".", "input dir.") flag.BoolVar(&verbose, "v", false, "add verbose if you want") flag.Parse() if verbose { tick = time.Tick(500 * time.Millisecond) } fileSizes := make(chan int64) // 起一個goroutine去walk目錄 go func() { walkDir(root, fileSizes) // Walk完畢之后要關閉該channel下面使用range讀取數據的時候才會有盡頭 close(fileSizes) }() loop: for { select { case size, ok := <-fileSizes: if !ok { break loop } nfiles++ nbytes += size case <-tick: fmt.Printf("%d files %.1f GB\n", nfiles, float64(nbytes)/1e9) } } fmt.Printf("%d files %.1f GB\n", nfiles, float64(nbytes)/1e9) }
上面其實都差不多，這里我直接從loop那里開始說吧，遇到這個loop的時候我其實還蠻疑惑的，因為我在go語言保留關鍵字里面並沒有看到他的身影，但是這里他的確是個關鍵字，和里面的break連用 里面break后面跟上的loop 可以直接跳出到最外層loop包裹的循環，而不是break默認的只跳出一層循環。明白了這個道理之后，這個就不難理解了，當我們還在遍歷文件的時候，select 會持續讀取文件大小賦值給size，並且返回true給ok。如果我們開啟了verbose，每隔500毫秒tick會收到來自time.Tick的消息。我們都知道select會在都准備好的情況下隨機pick一個執行，所以這里也或快或慢的被打印進度(前提是同時收到信號，但是實際上這個發生速度可能在nm級別，憑感受很難感覺到誰先)。當最后都執行完畢后filesSizes channel會被上面的攜程函數close()，當close之后，在讀取完剩余數據后，fileSizes會返回給ok nil。就可以跳出循環。
看到這里可能會覺得有點繞，所以要盡可能的多理解一下，當然我們可以讓這個du程序更快。可以注意到我們並沒有在walkdir里面開啟goroutines進行並發處理。下面我將嘗試開啟goroutine處理它們，並且用channel給他們加個鎖控制一下goroutine的數量，在此之前我們先來看看現在完成了的代碼：
package main import ( "fmt" "io/ioutil" "os" "path/filepath" "flag" "time" ) // 入參是一個文件目錄，一個INT64的只接收的單向channel func walkDir(dir string, fileSizes chan<- int64) { for _, entry := range dirEntries(dir) { if entry.IsDir() { subdir := filepath.Join(dir, entry.Name()) walkDir(subdir, fileSizes) } else { fileSizes <- entry.Size() } } } func dirEntries(dir string) []os.FileInfo { entries, err := ioutil.ReadDir(dir) if err != nil { fmt.Fprintf(os.Stderr, "du: %v\n", err) return nil } return entries } func main() { t1 := time.Now() root := "" verbose := false tick := make(<-chan time.Time) var nfiles, nbytes int64 flag.StringVar(&root, "p", ".", "input dir.") flag.BoolVar(&verbose, "v", false, "add verbose if you want") flag.Parse() if verbose { tick = time.Tick(500 * time.Millisecond) } fileSizes := make(chan int64) // 起一個goroutine去walk目錄 go func() { walkDir(root, fileSizes) // Walk完畢之后要關閉該channel下面使用range讀取數據的時候才會有盡頭 close(fileSizes) }() loop: for { select { case size, ok := <-fileSizes: if !ok { break loop } nfiles++ nbytes += size case <-tick: fmt.Printf("%d files %.1f GB\n", nfiles, float64(nbytes)/1e9) } } fmt.Printf("%d files %.1f GB\n", nfiles, float64(nbytes)/1e9) fmt.Println(time.Since(t1)) }
觀察上面代碼可以看出我們並不能直接在這個代碼的基礎上直接給walkDir加上goroutine，這樣會導致channel直接被關閉，然后啥也沒跑就結束了。
我們需要讓主goroutine等待其他goroutine都完成之后再結束，所以主goroutine需要在這里阻塞住，等到得到可以結束的信號之后再結束。
我們可以使用sync.WaitGroup 來對仍舊活躍的walkDir調用進行計數。等到數量為0的時候就算我們可以結束了。
sync.WaitGroup提供了三個方法：
Add：添加或減少goroutine的數量。
Done：相當於Add(-1)。
Wait：阻塞住等待WaitGroup數量變成0.
明白這個道理之后我們改寫了一下代碼，讓它使用sync.WaitGroup來支持同步，最后當所有goroutine都結束之后，關閉channel完成任務。
package main import ( "fmt" "io/ioutil" "os" "path/filepath" "flag" "time" "sync" ) // 入參是一個文件目錄，一個INT64的只接收的單向channel func walkDir(dir string, fileSizes chan<- int64, n *sync.WaitGroup) { defer n.Done() for _, entry := range dirEntries(dir) { if entry.IsDir() { n.Add(1) subdir := filepath.Join(dir, entry.Name()) go walkDir(subdir, fileSizes, n) } else { fileSizes <- entry.Size() } } } func dirEntries(dir string) []os.FileInfo { entries, err := ioutil.ReadDir(dir) if err != nil { fmt.Fprintf(os.Stderr, "du: %v\n", err) return nil } return entries } func main() { t1 := time.Now() root := "" verbose := false tick := make(<-chan time.Time) fileSizes := make(chan int64) var n sync.WaitGroup var nfiles, nbytes int64 flag.StringVar(&root, "p", ".", "input dir.") flag.BoolVar(&verbose, "v", false, "add verbose if you want") flag.Parse() if verbose { tick = time.Tick(500 * time.Millisecond) } n.Add(1) go walkDir(root, fileSizes, &n) go func() { n.Wait() close(fileSizes) }() loop: for { select { case size, ok := <-fileSizes: if !ok { break loop } nfiles++ nbytes += size case <-tick: fmt.Printf("%d files %.1f GB\n", nfiles, float64(nbytes)/1e9) } } fmt.Printf("%d files %.1f GB\n", nfiles, float64(nbytes)/1e9) fmt.Println(time.Since(t1)) }
隨便跑跑。。感覺快得飛起，然而跑不了幾秒就會報錯，這個程序最大的問題就是我們完全沒有辦法之后它會自己打開多少個goroutine，感覺會爆炸。所以我們要限制這種誇張的寫法，使用channel來做一個並發協程池，把同時開啟的goroutine的數量控制一下。
最后上一下完整代碼，注意defer關鍵字，只接收函數，所以我會在釋放鎖的時候使用匿名函數：
package main import ( "fmt" "io/ioutil" "os" "path/filepath" "flag" "time" "sync" ) var token = make(chan int, 100) // 入參是一個文件目錄，一個INT64的只接收的單向channel func walkDir(dir string, fileSizes chan<- int64, n *sync.WaitGroup) { defer n.Done() for _, entry := range dirEntries(dir) { if entry.IsDir() { n.Add(1) subdir := filepath.Join(dir, entry.Name()) go walkDir(subdir, fileSizes, n) } else { fileSizes <- entry.Size() } } } func dirEntries(dir string) []os.FileInfo { token <- 1 defer func() {<-token}() entries, err := ioutil.ReadDir(dir) if err != nil { fmt.Fprintf(os.Stderr, "du: %v\n", err) return nil } return entries } func main() { var nfiles, nbytes int64 var n sync.WaitGroup root := "" verbose := false t1 := time.Now() fileSizes := make(chan int64) tick := make(<-chan time.Time) flag.StringVar(&root, "p", ".", "input dir.") flag.BoolVar(&verbose, "v", false, "add verbose if you want") flag.Parse() if verbose { tick = time.Tick(500 * time.Millisecond) } n.Add(1) go walkDir(root, fileSizes, &n) go func() { n.Wait() close(fileSizes) }() loop: for { select { case size, ok := <-fileSizes: if !ok { break loop } nfiles++ nbytes += size case <-tick: fmt.Printf("%d files %.1f GB\n", nfiles, float64(nbytes)/1e9) } } fmt.Printf("%d files %.1f GB\n", nfiles, float64(nbytes)/1e9) fmt.Println(time.Since(t1)) }
Reference:
https://github.com/gopl-zh/gopl-zh.github.com The Go Programming Language