并发编程（2）

Posted on | Visitors:

并发编程示例

定时任务

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
func demo(input chan interface{}) {
    t1 := time.NewTimer(time.Second * 5)
    t2 := time.NewTimer(time.Second * 10)

    for {
        select {
        case msg <- input:
            println(msg)

        case <-t1.C:
            println("5s timer")
            t1.Reset(time.Second * 5)

        case <-t2.C:
            println("10s timer")
            t2.Reset(time.Second * 10)
        }
    }
}

断续器

1
2
3
4
5
6
func main(){
    ticker := time.NewTicker(time.Second)
	for t := range ticker.C {
		fmt.Println("ticker", t)
	}
}

超时

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
func main(){
    ch1 := make(chan int, 1)
    ch2 := make(chan int, 1)

    select {
        case e1 := <-ch1:
        //如果ch1通道成功读取数据，则执行该case处理语句
            fmt.Printf("1th case is selected. e1=%v",e1)
        case e2 := <-ch2:
        //如果ch2通道成功读取数据，则执行该case处理语句
            fmt.Printf("2th case is selected. e2=%v",e2)
        case <- time.After(2 * time.Second):
            fmt.Println("Timed out")
    }
}

自定义定时器

1
2
3
4
5
6
7
8
9
10
11
12
13
func main(){
    var t *time.Timer

    f := func(){
        fmt.Printf("Expiration time : %v.\n", time.Now())
        fmt.Printf("C`s len: %d\n", len(t.C))
    }

    t = time.AfterFunc(1*time.Second, f)
    //让当前Goroutine 睡眠2s，确保大于内容的完整
    //这样做原因是，time.AfterFunc的调用不会被阻塞。它会以一部的方式在到期事件来临执行我们自定义函数f。
    time.Sleep(2 * time.Second)
}

使用时间控制停止ticker

1
2
3
4
5
6
7
8
9
10
11
12
13
14
func main(){
    //初始化断续器,间隔2s
    var ticker *time.Ticker = time.NewTicker(1 * time.Second)

    go func() {
        for t := range ticker.C {
            fmt.Println("Tick at", t)
        }
    }()

    time.Sleep(time.Second * 5)   //阻塞，则执行次数为sleep的休眠时间/ticker的时间
    ticker.Stop()       //停止ticker
    fmt.Println("Ticker stopped")
}

并发读取文件夹所有文件的长度（使用通道）

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
func test1() {
    // 遍历文件夹，获取文件名
	dir := "./db"
	paths, err := ioutil.ReadDir(dir)
	check(err)
	var files []string
	for _, path := range paths {
		if !path.IsDir() {
			files = append(files, path.Name())
		}
	}

	type FileInfo struct {
		FileName string
		Len      int
		Err      error
	}
	way := make(chan FileInfo, 100)
	for _, file := range files {
		go func(fileName string) {
			content, err := ioutil.ReadFile(fmt.Sprintf("%s/%s", dir, fileName))
            // 无论对错，都交给通道收取方
			way <- FileInfo{
				FileName: fileName,
				Len:      len(content),
				Err:      err,
			}
		}(file)
	}

    // 我们知道需要从通道中取多少条数据
	for i := 0; i < len(files); i++ {
		fileInfo := <-way
		if fileInfo.Err != nil {
			fmt.Printf("[err] file:%s, err: %s \n", fileInfo.FileName, fileInfo.Err.Error())
			return
		}
		fmt.Printf("file:%s, len: %d \n", fileInfo.FileName, fileInfo.Len)
	}
}

并发读取 文件夹所有文件的长度（使用sync.WaitGroup）

这个使用的其实就是计数器的道理

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
func test1() {
	dir := "./db"
	paths, err := ioutil.ReadDir(dir)
	check(err)
	var files []string
	for _, path := range paths {
		if !path.IsDir() {
			files = append(files, path.Name())
		}
	}

    // sizes用来计算总长度和阻塞直至所有goroutine走完
	sizes := make(chan int64)
    // 计数器
	var wg sync.WaitGroup
	for _, file := range files {
		wg.Add(1)
		go func(fileName string) {
			defer wg.Done()
			fileInfo, _ := os.Stat(fmt.Sprintf("%s/%s", dir, fileName))
			sizes <- fileInfo.Size()
		}(file)
	}
	go func() {
		wg.Wait()
		close(sizes)
	}()
	var total int64
	for size := range sizes {
		fmt.Printf("size:%d\n", size)
		total += size
	}
	fmt.Printf("total size: %d\n", total)
}

令牌-另一种信号计数器

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
func check(err error) {
	if err != nil {
		panic(err)
	}
}

func main() {
	// 链接列表
	workList := make(chan []string)
	//  存储 已经处理的 链接，使用 map是为了去重
	seen := make(map[string]bool)
	// n是为了保证 爬虫一直  爬
	n := 1
	// 初始化 
	go func() { workList <- []string{"http://www.jianshu.com"} }()
	for i := n; i > 0; i++ {
		list := <-workList
		for _, link := range list {
			if !seen[link] {
				seen[link] = true
				n++
				go func(link string) {
					workList <- crawl(link)
				}(link)

			}
		}
	}
}

var tokens = make(chan struct{}, 20)

// 获取网页的链接  通过令牌
func crawl(url string) (list []string) {
	fmt.Println(url)
	tokens <- struct{}{}
	list, err := Extract(url)
	<-tokens
	if err != nil {
		log.Print(err)
		return
	}
	return
}

// 获取网页的链接
func Extract(url string) (list []string, err error) {
	resp, err := http.Get(url)
	if err != nil {
		return
	}
	if resp.StatusCode != http.StatusOK {
		resp.Body.Close()
		return nil, fmt.Errorf("getting %s, err %s", url, err.Error())
	}
	doc, err := html.Parse(resp.Body)
	resp.Body.Close()
	if err != nil {
		return nil, fmt.Errorf("parsing %s as HTML, err %s", url, err.Error())
	}
	visitNode := func(n *html.Node) {
		if n.Type == html.ElementNode && n.Data == "a" {
			for _, a := range n.Attr {
				if a.Key != "href" {
					continue
				}
				link, err := resp.Request.URL.Parse(a.Val)
				if err != nil {
					continue
				}
				list = append(list, link.String())
			}

		}
	}
	// 将函数作为参数传入 （匿名函数）
	forEachNode(doc, visitNode, nil)
	return
}

func forEachNode(n *html.Node, pre, next func(node *html.Node)) {
	if pre != nil {
		pre(n)
	}
	for n := n.FirstChild; n != nil; n = n.NextSibling {
		forEachNode(n, pre, next)
	}
	if next != nil {
		next(n)
	}

}