码迷,mamicode.com
首页 > 其他好文 > 详细

go channel 案例分析

时间:2014-09-19 03:24:25      阅读:304      评论:0      收藏:0      [点我收藏+]

标签:style   blog   http   color   io   os   使用   ar   for   

断断续续理了一下关于channel的一些概念,现在可以把下面的程序理清楚了。

1. source code

这个程序来自于《Go语言程序设计》 7.2.2 并发的Grep, 程序如下。

package main

import (
    "bufio"
    "bytes"
    "fmt"
    "io"
    "log"
    "os"
    "path/filepath"
    "regexp"
)

type Job struct {
    filename string
    results  chan<- Result
}

type Result struct {
    filename string
    fino     int
    line     string
}

var numberOfWorkers = 4

func main() {
    //retrieve input arguments
    if len(os.Args) < 3 || os.Args[1] == "-1" || os.Args[1] == "--help" {
        fmt.Printf("usage:  %s <regrex> <files>\n", filepath.Base(os.Args[0]))
        os.Exit(1)
    }

    if lineRx, err := regexp.Compile(os.Args[1]); err != nil {
        log.Fatalf("invalid regexp: %s\n", err)
    } else {
        //do the real work
        grep(lineRx, os.Args[2:])
    }
}

func grep(lineRx *regexp.Regexp, filenames []string) {
    jobs := make(chan Job, numberOfWorkers)
    done := make(chan struct{}, numberOfWorkers)
    results := make(chan Result, minimun(1000, len(filenames)))

    go addJobs(jobs, filenames, results) //produce Jobs
    for i := 0; i < numberOfWorkers; i++ {
        go doJobs(done, lineRx, jobs) //consume Jobs concurrent
    }
    go awaitCompletion(done, results) //wait until all the Jobs done
    processResults(results)
}

func addJobs(jobs chan<- Job, filenames []string, result chan<- Result) {
    for _, filename := range filenames {
        jobs <- Job{filename, result}
    }
    close(jobs)
}

func doJobs(done chan<- struct{}, lineRx *regexp.Regexp, jobs <-chan Job) {
    for job := range jobs {
        job.Do(lineRx)
    }
    done <- struct{}{}
}

func awaitCompletion(done <-chan struct{}, results chan Result) {
    for i := 0; i < numberOfWorkers; i++ {
        <-done
    }
    //???what will happen to data in results???
    close(results)
}

func processResults(results <-chan Result) {
    for result := range results {
        fmt.Printf("%s:%d:%s\n", result.filename, result.fino, result.line)
    }
}

func minimun(a int, b int) int {
    if a <= b {
        return a
    } else {
        return b
    }
}

func (job Job) Do(lineRx *regexp.Regexp) {
    file, err := os.Open(job.filename)
    if err != nil {
        log.Printf("error: %s\n", err)
        return
    }
    defer file.Close()

    reader := bufio.NewReader(file)
    for lino := 1; ; lino++ {
        line, err := reader.ReadBytes(\n)
        if err != nil {
            if err != io.EOF {
                log.Printf("error:%d: %s\n", lino, err)
            }
            break
        }

        line = bytes.TrimRight(line, "\n\r")
        if lineRx.Match(line) {
            job.results <- Result{job.filename, lino, string(line)}
        }
    }
}

 

编译程序:

go build program

program  a12345b   a.txt  b.txt c.txt...

2.编写并发go routinue 的一些模式说明

同步通信时需要避开两个陷阱:

陷阱一:主线程提前退出

当其他线程工作没有完成,而主线程提前退出。主线程退出会导致其他线程强制退出,而得不到想要的结果。

常见的解决方式是 让主 gorountine在done通道上等待,根据接收到的消息判断工作是否完成。另一种是使用sync.WaitGroup。

陷阱二:死锁

注意读写线程之间的关系,例如:不关闭 写chanel 会导致 使用range 读数据的 rountine堵塞。

 

3.程序同步关系图

这里我画出程序执行流程:

bubuko.com,布布扣

0-1 : prepare channels

2:     addJobs start

3:     in addJobs,   close(jobs)  to notice the reader

4:     in doJobs,     finish read jobs(not blocked here),  the consumer won‘t be controlled by the producer

5:     in do Jobs,   close done to inform that the reader is free

6, 7, 8, 9:   ....

10:  whole program finished

 

从上述流程上可以看出,都是生产者 通过一定的形式 通知消费者。告知消费者,产品已经生产完成,因此消费者不需要等待了,消费者只需要把剩下的任务完成就可以,消费者不需要受控于生产者了。

而这种通知 通过两种形式完成。

第一种: close channel.  例如:  close(jobs),  close(results)

第二种:读写done 通道。例如:  done <- struct{}{},   <-done。  由于done占用资源比较小,程序中并没有把它关闭。

 

continue to the github project  gocrawl.....

go channel 案例分析

标签:style   blog   http   color   io   os   使用   ar   for   

原文地址:http://www.cnblogs.com/harrysun/p/3980501.html

(0)
(0)
   
举报
评论 一句话评论(0
登录后才能评论!
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码!