妙用GO信道(channel)

云迹 · · 16466 次点击 · · 开始浏览    
这是一个创建于 的文章,其中的信息可能已经有所发展或是发生改变。

channel 是 golang 里相当有趣的一个功能,在我使用 golang 编码的经验里,大部分事件都会是在享受 channel 和 goroutine 配合的乐趣。所以本文主要介绍 channel 的一些有趣的用法。

这里有 Oling Cat 翻译的Go编程语言规范里关于 channel(信道)的描述:

信道提供了一种机制,它在两个并发执行的函数之间进行同步,并通过传递(与该信道元素类型相符的)值来进行通信。

这个个描述又乏味、又枯燥。在我第一次阅读的时候,完全不明白这到底是个什么玩意。事实上,可以认为 channel 是一个管道或者先进先出队列,非常简单且轻量。channel 并不是 Golang 首创的。它同样作为内置功能出现在其他语言中。在大多数情况下,它是一个又大、又笨、又复杂的消息队列系统的一个功能。

下面就来一起找点乐子吧!

最常见的方式:生产者/消费者

生产者产生一些数据将其放入 channel;然后消费者按照顺序,一个一个的从 channel 中取出这些数据进行处理。这是最常见的 channel 的使用方式。当 channel 的缓冲用尽时,生产者必须等待(阻塞)。换句话说,若是 channel 中没有数据,消费者就必须等待了。

这个例子的源代码在这里。最好下载到本地运行。

生产者

func producer(c chan int64, max int) {
    defer
    close(c)

    for i:= 0; i < max; i ++ {
        c <- time.Now().Unix()
    }
}

生产者生成“max”个 int64 的数字,并且将其放入 channel “c” 中。需要注意的是,这里用 defer 在函数推出的时候关闭了 channel。

消费者

func consumer(c chan int64) {
    var v int64
    ok := true

    for ok {
        if v, ok = <-c; ok {
            fmt.Println(v)
        }
    }
}

从 channel 中一个一个的读取 int64 的数字,然后将其打印在屏幕上。当 channel 被关闭后,变量“ok”将被设置为“false”。

自增长 ID 生成器

当生让产者可以顺序的生成整数。它就是一个自增长 ID 生成器。我将这个功能封装成了一个包。并将其代码托管在这里。使用示例可以参考这里的代码。

type AutoInc struct {
    start, step int
    queue       chan int
    running     bool
}

func New(start, step int) (ai *AutoInc) {
    ai = &AutoInc{
        start:   start,
        step:    step,
        running: true,
        queue:   make(chan int, 4),
    }

    go ai.process()
    return

}

func (ai *AutoInc) process() {
    defer func() { recover() }()
    for i := ai.start; ai.running; i = i + ai.step {
        ai.queue <- i
    }
}

func (ai *AutoInc) Id() int {
    return <-ai.queue
}

func (ai *AutoInc) Close() {
    ai.running = false
    close(ai.queue)
}

信号量

信号量也是 channel 的一个有趣的应用。这里有一个来自“effective go”的例子。你应当读过了吧?如果还没有,现在就开始读吧……

我在 Gearman 服务的 API 包 gearman-go 中使用了信号量。在 worker/worker.go 的 232 行,在并行的 Worker.exec 的数量达到 Worker.limit 时,将被阻塞。

var sem = make(chan int, MaxOutstanding)

func handle(r *Request) {
    sem <- 1 // 等待放行;
    process(r)
    // 可能需要一个很长的处理过程;
    <-sem // 完成,放行另一个过程。
}

func Serve(queue chan *Request) {
    for {
        req := <-queue
        go handle(req) // 无需等待 handle 完成。
    }
}

随机序列生成器

当然可以修改自增长 ID 生成器。让生产者生成随机数放入 channel。不过这挺无聊的,不是吗?

这里是随机序列生成器的另一个实现。灵感来自语言规范。它会随机的生成 0/1 序列:

func producer(c chan int64, max int) {
    defer close(c)
    for i := 0; i < max; i++ {
        select { // randomized select
        case c <- 0:
        case c <- 1:
        }
    }
}

超时定时器

当一个 channel 被 read/write 阻塞时,它会被永远阻塞下去,直到 channel 被关闭,这时会产生一个 panic。channel 没有内建用于超时的定时器。并且似乎也没有计划向 channel 添加一个这样的功能。但在大多数情况下,我们需要一个超时机制。例如,由于生产者执行的时候发生了错误,所以没有向 channel 放入数据。消费者会被阻塞到 channel 被关闭。每次出错都关闭 channel?这绝对不是一个好主意。

这里有一个解决方案:

c := make(chan int64, 5)
defer close(c)
timeout := make(chan bool)
defer close(timeout)
go func() {
    time.Sleep(time.Second) // 等一秒
    timeout <- true // 向超时队列中放入标志
}()

select {
case <-timeout: // 超时
    fmt.Println("timeout...")
case <-c: // 收到数据
    fmt.Println("Read a date.")
}

你注意到 select 语句了吗?哪个 channel 先有数据,哪个分支先执行。因此……还需要更多的解释吗?

这同样被使用在gearman-go 的客户端 API 实现中,第 238 行。

在本文的英文版本发布后,@mjq 提醒我说可以用 time.After。在项目中,这确实是更好的写法。我得向他道谢!同时我也阅读了 src/time/sleep.go 第 74 行,time.After 的实现。其内部实现与上面的代码完全一致。


有疑问加站长微信联系(非本文作者)

本文来自:开源中国博客

感谢作者:云迹

查看原文:妙用GO信道(channel)

入群交流(和以上内容无关):加入Go大咖交流群,或添加微信:liuxiaoyan-s 备注:入群;或加QQ群:692541889

16466 次点击  
加入收藏 微博
被以下专栏收入,发现更多相似内容
暂无回复
添加一条新回复 (您需要 登录 后才能回复 没有账号 ?)
  • 请尽量让自己的回复能够对别人有帮助
  • 支持 Markdown 格式, **粗体**、~~删除线~~、`单行代码`
  • 支持 @ 本站用户;支持表情(输入 : 提示),见 Emoji cheat sheet
  • 图片支持拖拽、截图粘贴等方式上传