goroutine機(jī)制可以方便地實(shí)現(xiàn)異步處理
package main
import (
"log"
"time"
"github.com/gin-gonic/gin"
)
func main() {
// 1.創(chuàng)建路由
// 默認(rèn)使用了2個(gè)中間件Logger(), Recovery()
r := gin.Default()
// 1.異步
r.GET("/long_async", func(c *gin.Context) {
// 需要搞一個(gè)副本
copyContext := c.Copy()
// 異步處理
go func() {
time.Sleep(3 * time.Second)
log.Println("異步執(zhí)行：" + copyContext.Request.URL.Path)
}()
})
// 2.同步
r.GET("/long_sync", func(c *gin.Context) {
time.Sleep(3 * time.Second)
log.Println("同步執(zhí)行：" + c.Request.URL.Path)
})
r.Run(":8000")
}
補(bǔ)充：Golang的channel使用以及并發(fā)同步技巧
在學(xué)習(xí)《The Go Programming Language》第八章并發(fā)單元的時(shí)候還是遭遇了不少問題，和值得總結(jié)思考和記錄的地方。
做一個(gè)類似于unix du命令的工具。但是閹割了一些功能，這里應(yīng)該只實(shí)現(xiàn)-c(統(tǒng)計(jì)total大小) 和-h(以human比較容易辨識(shí)的顯示出來)的功能。
首先我們需要構(gòu)造一個(gè) 能夠返回FileInfo信息數(shù)組的函數(shù)，我們把它取名為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
}
傳入一個(gè)路徑字符串，然后使用ioutil.ReadDir解析這個(gè)路徑下面的所有文件以及文件夾生成一個(gè)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)
}
多種方法，可以直接調(diào)用，其作用就是后面注釋寫的一樣。
有了能夠獲取文件夾下面文件和文件夾的函數(shù)之后，我們需要一個(gè)調(diào)用方用來walk指定的目錄：
// 入?yún)⑹且粋€(gè)文件目錄，一個(gè)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()
}
}
}
這里我們定義一個(gè)目錄，然后需求傳入一個(gè)單向接收channel用于在多goroutine中計(jì)算總共的文件大小。
使用range方法來遍歷我們上面寫的dirEntries的返回文件或文件夾，如果是文件夾則繼續(xù)迭代。
如果不是則將文件大小存入放入fileSizes channel中。
搞定上面兩個(gè)函數(shù)，我們來寫主函數(shù)部分：
func main() {
root := ""
flag.StringVar(root, "-p", ".", "input dir.")
flag.Parse()
fileSizes := make(chan int64)
// 起一個(gè)goroutine去walk目錄
go func() {
walkDir(root, fileSizes)
// Walk完畢之后要關(guān)閉該channel下面使用range讀取數(shù)據(jù)的時(shí)候才會(huì)有盡頭
close(fileSizes)
}()
var nfiles, nbytes int64
for size := range fileSizes {
nfiles++
nbytes += size
}
fmt.Printf("%d files %.1f GB\n", nfiles, float64(nbytes)/1e9)
}
這里注意一點(diǎn)，因?yàn)槠餲oroutine的walk函數(shù)，和下面同時(shí)在range遍歷是在同步進(jìn)行，如果下面range速度太快讀到管道里面沒有值了會(huì)阻塞住等待有數(shù)據(jù)繼續(xù)進(jìn)來之后讀取，而不是會(huì)跳出。只有當(dāng)close(fileSizes)這句執(zhí)行到，顯示關(guān)閉掉channel之后，才會(huì)跳出range循環(huán)并且這時(shí)已經(jīng)讀取完了所有的數(shù)據(jù)。這里有點(diǎn)像，close channel的時(shí)候給range發(fā)送了一個(gè)停止信號(hào)一樣，感覺這個(gè)利用起來會(huì)比較有用？ 后續(xù)可能會(huì)再研究一下。
讓我們繼續(xù)來優(yōu)化我們的程序，添加一個(gè)-v參數(shù)，打印出掃描文件的進(jìn)度，當(dāng)我們要掃描整個(gè)盤的時(shí)候，可能會(huì)花費(fèi)大量的時(shí)間，我們需要知道進(jìn)度如何了。
其實(shí)這個(gè)需求只需要很小的改動(dòng)，讓我們來重新改寫一下main函數(shù)，用select多路復(fù)用來完成這個(gè)事情。
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)
// 起一個(gè)goroutine去walk目錄
go func() {
walkDir(root, fileSizes)
// Walk完畢之后要關(guān)閉該channel下面使用range讀取數(shù)據(jù)的時(shí)候才會(huì)有盡頭
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)
}
上面其實(shí)都差不多，這里我直接從loop那里開始說吧，遇到這個(gè)loop的時(shí)候我其實(shí)還蠻疑惑的，因?yàn)槲以趃o語言保留關(guān)鍵字里面并沒有看到他的身影，但是這里他的確是個(gè)關(guān)鍵字，和里面的break連用 里面break后面跟上的loop 可以直接跳出到最外層loop包裹的循環(huán)，而不是break默認(rèn)的只跳出一層循環(huán)。明白了這個(gè)道理之后，這個(gè)就不難理解了，當(dāng)我們還在遍歷文件的時(shí)候，select 會(huì)持續(xù)讀取文件大小賦值給size，并且返回true給ok。如果我們開啟了verbose，每隔500毫秒tick會(huì)收到來自time.Tick的消息。我們都知道select會(huì)在都準(zhǔn)備好的情況下隨機(jī)pick一個(gè)執(zhí)行，所以這里也或快或慢的被打印進(jìn)度(前提是同時(shí)收到信號(hào)，但是實(shí)際上這個(gè)發(fā)生速度可能在nm級(jí)別，憑感受很難感覺到誰先)。當(dāng)最后都執(zhí)行完畢后filesSizes channel會(huì)被上面的攜程函數(shù)close()，當(dāng)close之后，在讀取完剩余數(shù)據(jù)后，fileSizes會(huì)返回給ok nil。就可以跳出循環(huán)。
看到這里可能會(huì)覺得有點(diǎn)繞，所以要盡可能的多理解一下，當(dāng)然我們可以讓這個(gè)du程序更快。可以注意到我們并沒有在walkdir里面開啟goroutines進(jìn)行并發(fā)處理。下面我將嘗試開啟goroutine處理它們，并且用channel給他們加個(gè)鎖控制一下goroutine的數(shù)量，在此之前我們先來看看現(xiàn)在完成了的代碼：
package main
import (
"fmt"
"io/ioutil"
"os"
"path/filepath"
"flag"
"time"
)
// 入?yún)⑹且粋€(gè)文件目錄，一個(gè)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)
// 起一個(gè)goroutine去walk目錄
go func() {
walkDir(root, fileSizes)
// Walk完畢之后要關(guān)閉該channel下面使用range讀取數(shù)據(jù)的時(shí)候才會(huì)有盡頭
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))
}
觀察上面代碼可以看出我們并不能直接在這個(gè)代碼的基礎(chǔ)上直接給walkDir加上goroutine，這樣會(huì)導(dǎo)致channel直接被關(guān)閉，然后啥也沒跑就結(jié)束了。
我們需要讓主goroutine等待其他goroutine都完成之后再結(jié)束，所以主goroutine需要在這里阻塞住，等到得到可以結(jié)束的信號(hào)之后再結(jié)束。
我們可以使用sync.WaitGroup 來對(duì)仍舊活躍的walkDir調(diào)用進(jìn)行計(jì)數(shù)。等到數(shù)量為0的時(shí)候就算我們可以結(jié)束了。
sync.WaitGroup提供了三個(gè)方法：
Add：添加或減少goroutine的數(shù)量。
Done：相當(dāng)于Add(-1)。
Wait：阻塞住等待WaitGroup數(shù)量變成0.
明白這個(gè)道理之后我們改寫了一下代碼，讓它使用sync.WaitGroup來支持同步，最后當(dāng)所有g(shù)oroutine都結(jié)束之后，關(guān)閉channel完成任務(wù)。
package main
import (
"fmt"
"io/ioutil"
"os"
"path/filepath"
"flag"
"time"
"sync"
)
// 入?yún)⑹且粋€(gè)文件目錄，一個(gè)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))
}
隨便跑跑。。感覺快得飛起，然而跑不了幾秒就會(huì)報(bào)錯(cuò)，這個(gè)程序最大的問題就是我們完全沒有辦法之后它會(huì)自己打開多少個(gè)goroutine，感覺會(huì)爆炸。所以我們要限制這種夸張的寫法，使用channel來做一個(gè)并發(fā)協(xié)程池，把同時(shí)開啟的goroutine的數(shù)量控制一下。
最后上一下完整代碼，注意defer關(guān)鍵字，只接收函數(shù)，所以我會(huì)在釋放鎖的時(shí)候使用匿名函數(shù)：
package main
import (
"fmt"
"io/ioutil"
"os"
"path/filepath"
"flag"
"time"
"sync"
)
var token = make(chan int, 100)
// 入?yún)⑹且粋€(gè)文件目錄，一個(gè)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
以上為個(gè)人經(jīng)驗(yàn)，希望能給大家一個(gè)參考，也希望大家多多支持腳本之家。如有錯(cuò)誤或未考慮完全的地方，望不吝賜教。
您可能感興趣的文章:- golang中for循環(huán)遍歷channel時(shí)需要注意的問題詳解
- golang實(shí)現(xiàn)基于channel的通用連接池詳解
- Golang優(yōu)雅關(guān)閉channel的方法示例
- golang中單向channel的語法介紹
- 解決Golang中g(shù)oroutine執(zhí)行速度的問題
- GOLANG使用Context管理關(guān)聯(lián)goroutine的方法
- 關(guān)于golang利用channel和goroutine完成統(tǒng)計(jì)素?cái)?shù)的思路
