标签 TIOBE 下的文章

论golang Timer Reset方法使用的正确姿势

2016年,Go语言Tiobe编程语言排行榜上位次的大幅蹿升(2016年12月份Tiobe榜单:go位列第16位,Rating值:1.939%)。与此同时,我们也能切身感受到Go语言在世界范围蓬勃发展,其在中国地界儿上的发展更是尤为猛烈^0^:For gopher们的job变多了、网上关于Go的资料也大有“汗牛充栋”之势。作为职业Gopher^0^,要为这个生态添砖加瓦,就要多思考、多总结,关键还要做到“遇到了问题,就要说出来,给出你的见解”。每篇文章都有自己的切入角度和关注重点,因此Gopher们也无需过于担忧资料的“重复”。

这次,我来说说在使用Go标准库中Timer的Reset方法时遇到的问题。

一、关于Timer原理的一些说明

网络编程方面,从用户视角看,golang表象上是一种“阻塞式”网络编程范式,而支撑这种“阻塞式”范式的则是内置于go编译后的executable file中的runtime。runtime利用网络IO多路复用机制实现多个进行网络通信的goroutine的合理调度。goroutine中的执行函数则相当于你在传统C编程中传给epoll机制的回调函数。golang一定层度上消除了在这方面“回调”这种“逆向思维”给你带来的心智负担,简化了网络编程的复杂性。

但长时间“阻塞”显然不能满足大多数业务情景,因此还需要一定的超时机制。比如:在socket层面,我们通过显式设置net.Dialer的Timeout或使用SetReadDeadline、SetWriteDeadline以及SetDeadline;在应用层协议,比如http,client通过设置timeout参数,server通过TimeoutHandler来限制操作的time limit。这些timeout机制,有些是通过runtime的网络多路复用的timeout机制实现,有些则是通过Timer实现的。

标准库中的Timer让用户可以定义自己的超时逻辑,尤其是在应对select处理多个channel的超时、单channel读写的超时等情形时尤为方便。

1、Timer的创建

Timer是一次性的时间触发事件,这点与Ticker不同,后者则是按一定时间间隔持续触发时间事件。Timer常见的使用场景如下:

场景1:

t := time.AfterFunc(d, f)

场景2:

select {
    case m := <-c:
       handle(m)
    case <-time.After(5 * time.Minute):
       fmt.Println("timed out")
}

或:
t := time.NewTimer(5 * time.Minute)
select {
    case m := <-c:
       handle(m)
    case <-t.C:
       fmt.Println("timed out")
}

从这两个场景中,我们可以看到Timer三种创建姿势:

t:= time.NewTimer(d)
t:= time.AfterFunc(d, f)
c:= time.After(d)

虽然姿势不同,但背后的原理则是相通的。

Timer有三个要素:

* 定时时间:也就是那个d
* 触发动作:也就是那个f
* 时间channel: 也就是t.C

对于AfterFunc这种创建方式而言,Timer就是在超时(timer expire)后,执行函数f,此种情况下:时间channel无用。

//$GOROOT/src/time/sleep.go

func AfterFunc(d Duration, f func()) *Timer {
    t := &Timer{
        r: runtimeTimer{
            when: when(d),
            f:    goFunc,
            arg:  f,
        },
    }
    startTimer(&t.r)
    return t
}

func goFunc(arg interface{}, seq uintptr) {
    go arg.(func())()
}

注意:从AfterFunc源码可以看到,外面传入的f参数并非直接赋值给了内部的f,而是作为wrapper function:goFunc的arg传入的。而goFunc则是启动了一个新的goroutine来执行那个外部传入的f。这是因为timer expire对应的事件处理函数的执行是在go runtime内唯一的timer events maintenance goroutine: timerproc中。为了不block timerproc的执行,必须启动一个新的goroutine。

//$GOROOT/src/runtime/time.go
func timerproc() {
    timers.gp = getg()
    for {
        lock(&timers.lock)
        ... ...
            f := t.f
            arg := t.arg
            seq := t.seq
            unlock(&timers.lock)
            if raceenabled {
                raceacquire(unsafe.Pointer(t))
            }
            f(arg, seq)
            lock(&timers.lock)
        }
        ... ...
        unlock(&timers.lock)
   }
}

而对于NewTimer和After这两种创建方法,则是Timer在超时(timer expire)后,执行一个标准库中内置的函数:sendTime。sendTime将当前当前事件send到timer的时间Channel中,那么说这个动作不会阻塞到timerproc的执行么?答案肯定是不会的,其原因就在下面代码中:

//$GOROOT/src/time/sleep.go
func NewTimer(d Duration) *Timer {
    c := make(chan Time, 1)
    t := &Timer{
        C: c,
        ... ...
    }
    ... ...
    return t
}

func sendTime(c interface{}, seq uintptr) {
    // Non-blocking send of time on c.
    // Used in NewTimer, it cannot block anyway (buffer).
    // Used in NewTicker, dropping sends on the floor is
    // the desired behavior when the reader gets behind,
    // because the sends are periodic.
    select {
    case c.(chan Time) <- Now():
    default:
    }
}

我们看到NewTimer中创建了一个buffered channel,size = 1。正常情况下,当timer expire,t.C无论是否有goroutine在read,sendTime都可以non-block的将当前时间发送到C中;同时,我们看到sendTime还加了双保险:通过一个select判断c buffer是否已满,一旦满了,直接退出,依然不会block,这种情况在reuse active timer时可能会遇到。

2、Timer的资源释放

很多Go初学者在使用Timer时都会担忧Timer的创建会占用系统资源,比如:

有人会认为:创建一个Timer后,runtime会创建一个单独的Goroutine去计时并在expire后发送当前时间到channel里。
还有人认为:创建一个timer后,runtime会申请一个os级别的定时器资源去完成计时工作。

实际情况并不是这样。恰好近期gopheracademy blog发布了一篇 《How Do They Do It: Timers in Go》,通过对timer源码的分析,讲述了timer的原理,大家可以看看。

go runtime实际上仅仅是启动了一个单独的goroutine,运行timerproc函数,维护了一个”最小堆”,定期wake up后,读取堆顶的timer,执行timer对应的f函数,并移除该timer element。创建一个Timer实则就是在这个最小堆中添加一个element,Stop一个timer,则是从堆中删除对应的element。

同时,从上面的两个Timer常见的使用场景中代码来看,我们并没有显式的去释放什么。从上一节我们可以看到,Timer在创建后可能占用的资源还包括:

  • 0或一个Channel
  • 0或一个Goroutine

这些资源都会在timer使用后被GC回收。

综上,作为Timer的使用者,我们要做的就是尽量减少在使用Timer时对最小堆管理goroutine和GC的压力即可,即:及时调用timer的Stop方法从最小堆删除timer element(如果timer 没有expire)以及reuse active timer。

BTW,这里还有一篇讨论go Timer精度的文章,大家可以拜读一下。

二、Reset到底存在什么问题?

铺垫了这么多,主要还是为了说明Reset的使用问题。什么问题呢?我们来看下面的例子。这些例子主要是为了说明Reset问题,现实中很可能大家都不这么写代码逻辑。当前环境:go version go1.7 darwin/amd64。

1、example1

我们的第一个example如下:

//example1.go

func main() {
    c := make(chan bool)

    go func() {
        for i := 0; i < 5; i++ {
            time.Sleep(time.Second * 1)
            c <- false
        }

        time.Sleep(time.Second * 1)
        c <- true
    }()

    go func() {
        for {
            // try to read from channel, block at most 5s.
            // if timeout, print time event and go on loop.
            // if read a message which is not the type we want(we want true, not false),
            // retry to read.
            timer := time.NewTimer(time.Second * 5)
            defer timer.Stop()
            select {
            case b := <-c:
                if b == false {
                    fmt.Println(time.Now(), ":recv false. continue")
                    continue
                }
                //we want true, not false
                fmt.Println(time.Now(), ":recv true. return")
                return
            case <-timer.C:
                fmt.Println(time.Now(), ":timer expired")
                continue
            }
        }
    }()

    //to avoid that all goroutine blocks.
    var s string
    fmt.Scanln(&s)
}

example1.go的逻辑大致就是 一个consumer goroutine试图从一个channel里读出true,如果读出false或timer expire,那么继续try to read from the channel。这里我们每次循环都创建一个timer,并在go routine结束后Stop该timer。另外一个producer goroutine则负责生产消息,并发送到channel中。consumer中实际发生的行为取决于producer goroutine的发送行为。

example1.go执行的结果如下:

$go run example1.go
2016-12-21 14:52:18.657711862 +0800 CST :recv false. continue
2016-12-21 14:52:19.659328152 +0800 CST :recv false. continue
2016-12-21 14:52:20.661031612 +0800 CST :recv false. continue
2016-12-21 14:52:21.662696502 +0800 CST :recv false. continue
2016-12-21 14:52:22.663531677 +0800 CST :recv false. continue
2016-12-21 14:52:23.665210387 +0800 CST :recv true. return

输出如预期。但在这个过程中,我们新创建了6个Timer。

2、example2

如果我们不想重复创建这么多Timer实例,而是reuse现有的Timer实例,那么我们就要用到Timer的Reset方法,见下面example2.go,考虑篇幅,这里仅列出consumer routine代码,其他保持不变:

//example2.go
.... ...
// consumer routine
    go func() {
        // try to read from channel, block at most 5s.
        // if timeout, print time event and go on loop.
        // if read a message which is not the type we want(we want true, not false),
        // retry to read.
        timer := time.NewTimer(time.Second * 5)
        for {
            // timer is active , not fired, stop always returns true, no problems occurs.
            if !timer.Stop() {
                <-timer.C
            }
            timer.Reset(time.Second * 5)
            select {
            case b := <-c:
                if b == false {
                    fmt.Println(time.Now(), ":recv false. continue")
                    continue
                }
                //we want true, not false
                fmt.Println(time.Now(), ":recv true. return")
                return
            case <-timer.C:
                fmt.Println(time.Now(), ":timer expired")
                continue
            }
        }
    }()
... ...

按照go 1.7 doc中关于Reset使用的建议:

To reuse an active timer, always call its Stop method first and—if it had expired—drain the value from its channel. For example:

if !t.Stop() {
        <-t.C
}
t.Reset(d)

我们改造了example1,形成example2的代码。由于producer行为并未变更,实际example2执行时,每次循环Timer在被Reset之前都没有expire,也没有fire a time to channel,因此timer.Stop的调用均返回true,即成功将timer从“最小堆”中移除。example2的执行结果如下:

$go run example2.go
2016-12-21 15:10:54.257733597 +0800 CST :recv false. continue
2016-12-21 15:10:55.259349877 +0800 CST :recv false. continue
2016-12-21 15:10:56.261039127 +0800 CST :recv false. continue
2016-12-21 15:10:57.262770422 +0800 CST :recv false. continue
2016-12-21 15:10:58.264534647 +0800 CST :recv false. continue
2016-12-21 15:10:59.265680422 +0800 CST :recv true. return

和example1并无二致。

3、example3

现在producer routine的发送行为发生了变更:从以前每隔1s发送一次数据变成了每隔7s发送一次数据,而consumer routine不变:

//example3.go

//producer routine
    go func() {
        for i := 0; i < 10; i++ {
            time.Sleep(time.Second * 7)
            c <- false
        }

        time.Sleep(time.Second * 7)
        c <- true
    }()

我们来看看example3.go的执行结果:

$go run example3.go
2016-12-21 15:14:32.764410922 +0800 CST :timer expired

程序hang住了。你能猜到在哪里hang住的吗?对,就是在drain t.C的时候hang住了:

           // timer may be not active and may not fired
            if !timer.Stop() {
                <-timer.C //drain from the channel
            }
            timer.Reset(time.Second * 5)

producer的发送行为发生了变化,Comsumer routine在收到第一个数据前有了一次time expire的事件,for loop回到loop的开始端。这时timer.Stop函数返回的不再是true,而是false,因为timer已经expire,最小堆中已经不包含该timer了,Stop在最小堆中找不到该timer,返回false。于是example3代码尝试抽干(drain)timer.C中的数据。但timer.C中此时并没有数据,于是routine block在channel recv上了。

在Go 1.8以前版本中,很多人遇到了类似的问题,并提出issue,比如:

time: Timer.Reset is not possible to use correctly #14038

不过go team认为这还是文档中对Reset的使用描述不够充分导致的,于是在Go 1.8中对Reset方法的文档做了补充Go 1.8 beta2中Reset方法的文档改为了:

Resetting a timer must take care not to race with the send into t.C that happens when the current timer expires. If a program has already received a value from t.C, the timer is known to have expired, and t.Reset can be used directly. If a program has not yet received a value from t.C, however, the timer must be stopped and—if Stop reports that the timer expired before being stopped—the channel explicitly drained:

if !t.Stop() {
        <-t.C
}
t.Reset(d)

大致意思是:如果明确time已经expired,并且t.C已经被取空,那么可以直接使用Reset;如果程序之前没有从t.C中读取过值,这时需要首先调用Stop(),如果返回true,说明timer还没有expire,stop成功删除timer,可直接reset;如果返回false,说明stop前已经expire,需要显式drain channel。

4、example4

我们的example3就是“time已经expired,并且t.C已经被取空,那么可以直接使用Reset ”这第一种情况,我们应该直接reset,而不用显式drain channel。如何将这两种情形合二为一,很直接的想法就是增加一个开关变量isChannelDrained,标识timer.C是否已经被取空,如果取空,则直接调用Reset。如果没有,则drain Channel。

增加一个变量总是麻烦的,RussCox也给出一个未经详尽验证的方法,我们来看看用这种方法改造的example4.go:

//example4.go

//consumer
    go func() {
        // try to read from channel, block at most 5s.
        // if timeout, print time event and go on loop.
        // if read a message which is not the type we want(we want true, not false),
        // retry to read.
        timer := time.NewTimer(time.Second * 5)
        for {
            // timer may be not active, and fired
            if !timer.Stop() {
                select {
                case <-timer.C: //try to drain from the channel
                default:
                }
            }
            timer.Reset(time.Second * 5)
            select {
            case b := <-c:
                if b == false {
                    fmt.Println(time.Now(), ":recv false. continue")
                    continue
                }
                //we want true, not false
                fmt.Println(time.Now(), ":recv true. return")
                return
            case <-timer.C:
                fmt.Println(time.Now(), ":timer expired")
                continue
            }
        }
    }()

执行结果:

$go run example4.go
2016-12-21 15:38:16.704647957 +0800 CST :timer expired
2016-12-21 15:38:18.703107177 +0800 CST :recv false. continue
2016-12-21 15:38:23.706665507 +0800 CST :timer expired
2016-12-21 15:38:25.705314522 +0800 CST :recv false. continue
2016-12-21 15:38:30.70900638 +0800 CST :timer expired
2016-12-21 15:38:32.707482917 +0800 CST :recv false. continue
2016-12-21 15:38:37.711260142 +0800 CST :timer expired
2016-12-21 15:38:39.709668705 +0800 CST :recv false. continue
2016-12-21 15:38:44.71337522 +0800 CST :timer expired
2016-12-21 15:38:46.710880007 +0800 CST :recv false. continue
2016-12-21 15:38:51.713813305 +0800 CST :timer expired
2016-12-21 15:38:53.713063822 +0800 CST :recv true. return

我们利用一个select来包裹channel drain,这样无论channel中是否有数据,drain都不会阻塞住。看似问题解决了。

5、竞争条件

如果你看过timerproc的代码,你会发现其中的这样一段代码:

// go1.7
// $GOROOT/src/runtime/time.go
            f := t.f
            arg := t.arg
            seq := t.seq
            unlock(&timers.lock)
            if raceenabled {
                raceacquire(unsafe.Pointer(t))
            }
            f(arg, seq)
            lock(&timers.lock)

我们看到在timerproc执行f(arg, seq)这个函数前,timerproc unlock了timers.lock,也就是说f的执行并没有在锁内。

前面说过,f的执行是什么?

对于AfterFunc来说,就是启动一个goroutine,并在这个新goroutine中执行用户传入的函数;
对于After和NewTimer这种创建姿势创建的timer而言,f的执行就是sendTime的执行,也就是向t.C中send 当前时间。

注意:这时候timer expire过程中sendTime的执行与“drain channel”是分别在两个goroutine中执行的,谁先谁后,完全依靠runtime调度。于是example4.go中的看似没有问题的代码,也可能存在问题(当然需要时间粒度足够小,比如ms级的Timer)。

如果sendTime的执行发生在drain channel执行前,那么就是example4.go中的执行结果:Stop返回false(因为timer已经expire了),显式drain channel会将数据读出,后续Reset后,timer正常执行;
如果sendTime的执行发生在drain channel执行后,那么问题就来了,虽然Stop返回false(因为timer已经expire),但drain channel并没有读出任何数据。之后,sendTime将数据发到channel中。timer Reset后的Timer中的Channel实际上已经有了数据,于是当进入下面的select执行体时,”case <-timer.C:”瞬间返回,触发了timer事件,没有启动超时等待的作用。

这也是issue:*time: Timer.C can still trigger even after Timer.Reset is called #11513中问到的问题。

go官方文档中对此也有描述:

Note that it is not possible to use Reset's return value correctly, as there is a race condition between draining the channel and the new timer expiring. Reset should always be invoked on stopped or expired channels, as described above. The return value exists to preserve compatibility with existing programs.

三、真的有Reset方法的正确使用姿势吗?

综合上述例子和分析,Reset的使用似乎没有理想的方案,但一般来说,在特定业务逻辑下,Reset还是可以正常工作的,就如example4那样。即便出现问题,如果了解了Reset背后的原理,问题解决起来也是会很快很准的。

文中的相关代码可以在这里下载

四、参考资料

Golang官方有关Timer的issue list:

runtime: special case timer channels #8898
time:timer stop ,how to use? #14947
time: document proper usage of Timer.Stop #14383
*time: Timer.Reset is not possible to use correctly #14038
Time.After doesn’t release memory #15781
runtime: timerproc does not get to run under load #15706
time: time.After uses memory until duration times out #15698
time:timer stop panic #14946
*time: Timer.C can still trigger even after Timer.Reset is called #11513
time: Timer.Stop documentation incorrect for Timer returned by AfterFunc #17600

相关资料:

  1. go中的定时器timer
  2. Go内部实现之timer
  3. Go定时器
  4. How Do They Do It: Timers in Go
  5. timer在go可以有多精确

Golang Channel用法简编

在进入正式内容前,我这里先顺便转发一则消息,那就是Golang 1.3.2已经正式发布了。国内的golangtc已经镜像了golang.org的安装包下载页面,国内go程序员与爱好者们可以到"Golang中 国",即golangtc.com去下载go 1.3.2版本。

Go这门语言也许你还不甚了解,甚至是完全不知道,这也有情可原,毕竟Go在TIOBE编程语言排行榜上位列30开外。但近期使用Golang 实现的一杀手级应用 Docker你却不该不知道。docker目前火得是一塌糊涂啊。你去国内外各大技术站点用眼轻瞥一下,如 果没有涉及到“docker”字样新闻的站点建 议你以后就不要再去访问了^_^。Docker是啥、怎么用以及基础实践可以参加国内一位仁兄的经验之作:《 Docker – 从入门到实践》。

据我了解,目前国内试水Go语言开发后台系统的大公司与初创公司日益增多,比如七牛、京东、小米,盛大,金山,东软,搜狗等,在这里我们可以看到一些公司的Go语言应用列表,并且目前这个列表似乎依旧在丰富中。国内Go语言的推广与布道也再稳步推进中,不过目前来看多以Go入 门与基础为主题,Go idioms、tips或Best Practice的Share并不多见,想必国内的先行者、布道师们还在韬光养晦,积攒经验,等到时机来临再厚积薄发。另外国内似乎还没有一个针对Go的 布道平台,比如Golang技术大会之类的的平台。

在国外,虽然Go也刚刚起步,但在Golang share的广度和深度方面显然更进一步。Go的国际会议目前还不多,除了Golang老东家Google在自己的各种大会上留给Golang展示自己的 机会外,由 Gopher Academy 发起的GopherCon 会议也于今年第一次举行,并放出诸多高质量资料,在这里可以下载。欧洲的Go语言大会.dotgo也即将开幕,估计后续这两个大会将撑起Golang技术分享 的旗帜。

言归正传,这里要写的东西并非原创,自己的Go仅仅算是入门级别,工程经验、Best Practice等还谈不上有多少,因此这里主要是针对GopherCon2014上的“舶来品”的学习心得。来自CloudFlare的工程师John Graham-Cumming谈了关于 Channel的实践经验,这里针对其分享的内容,记录一些学习体会和理解,并结合一些外延知识,也可以算是一种学习笔记吧,仅供参考。

一、Golang并发基础理论

Golang在并发设计方面参考了C.A.R Hoare的CSP,即Communicating Sequential Processes并发模型理论。但就像John Graham-Cumming所说的那样,多数Golang程序员或爱好者仅仅停留在“知道”这一层次,理解CSP理论的并不多,毕竟多数程序员是搞工程 的。不过要想系统学习CSP的人可以从这里下载到CSP论文的最新版本。

维基百科中概要罗列了CSP模型与另外一种并发模型Actor模型的区别:

Actor模型广义上讲与CSP模型很相似。但两种模型就提供的原语而言,又有一些根本上的不同之处:
    – CSP模型处理过程是匿名的,而Actor模型中的Actor则具有身份标识。
    – CSP模型的消息传递在收发消息进程间包含了一个交会点,即发送方只能在接收方准备好接收消息时才能发送消息。相反,actor模型中的消息传递是异步 的,即消息的发送和接收无需在同一时间进行,发送方可以在接收方准备好接收消息前将消息发送出去。这两种方案可以认为是彼此对偶的。在某种意义下,基于交 会点的系统可以通过构造带缓冲的通信的方式来模拟异步消息系统。而异步系统可以通过构造带消息/应答协议的方式来同步发送方和接收方来模拟交会点似的通信 方式。
    – CSP使用显式的Channel用于消息传递,而Actor模型则将消息发送给命名的目的Actor。这两种方法可以被认为是对偶的。某种意义下,进程可 以从一个实际上拥有身份标识的channel接收消息,而通过将actors构造成类Channel的行为模式也可以打破actors之间的名字耦合。

二、Go Channel基本操作语法

Go Channel的基本操作语法如下:

c := make(chan bool) //创建一个无缓冲的bool型Channel

c <- x        //向一个Channel发送一个值
<- c          //从一个Channel中接收一个值
x = <- c      //从Channel c接收一个值并将其存储到x中
x, ok = <- c  //从Channel接收一个值,如果channel关闭了或没有数据,那么ok将被置为false

不带缓冲的Channel兼具通信和同步两种特性,颇受青睐。

三、Channel用作信号(Signal)的场景

1、等待一个事件(Event)

等待一个事件,有时候通过close一个Channel就足够了。例如:

//testwaitevent1.go
package main

import "fmt"

func main() {
        fmt.Println("Begin doing something!")
        c := make(chan bool)
        go func() {
                fmt.Println("Doing something…")
                close(c)
        }()
        <-c
        fmt.Println("Done!")
}

这里main goroutine通过"<-c"来等待sub goroutine中的“完成事件”,sub goroutine通过close channel促发这一事件。当然也可以通过向Channel写入一个bool值的方式来作为事件通知。main goroutine在channel c上没有任何数据可读的情况下会阻塞等待。

关于输出结果:

根据《Go memory model》中关于close channel与recv from channel的order的定义:The closing of a channel happens before a receive that returns a zero value because the channel is closed.

我们可以很容易判断出上面程序的输出结果:

Begin doing something!
Doing something…
Done!

如果将close(c)换成c<-true,则根据《Go memory model》中的定义:A receive from an unbuffered channel happens before the send on that channel completes.
"<-c"要先于"c<-true"完成,但也不影响日志的输出顺序,输出结果仍为上面三行。

2、协同多个Goroutines

同上,close channel还可以用于协同多个Goroutines,比如下面这个例子,我们创建了100个Worker Goroutine,这些Goroutine在被创建出来后都阻塞在"<-start"上,直到我们在main goroutine中给出开工的信号:"close(start)",这些goroutines才开始真正的并发运行起来。

//testwaitevent2.go
package main

import "fmt"

func worker(start chan bool, index int) {
        <-start
        fmt.Println("This is Worker:", index)
}

func main() {
        start := make(chan bool)
        for i := 1; i <= 100; i++ {
                go worker(start, i)
        }
        close(start)
        select {} //deadlock we expected
}

3、Select

【select的基本操作】
select是Go语言特有的操作,使用select我们可以同时在多个channel上进行发送/接收操作。下面是select的基本操作。

select {
case x := <- somechan:
    // … 使用x进行一些操作

case y, ok := <- someOtherchan:
    // … 使用y进行一些操作,
    //
检查ok值判断someOtherchan是否已经关闭

case outputChan <- z:
    // … z值被成功发送到Channel上时

default:
    // … 上面case均无法通信时,执行此分支
}

【惯用法:for/select】

我们在使用select时很少只是对其进行一次evaluation,我们常常将其与for {}结合在一起使用,并选择适当时机从for{}中退出。

for {
        select {
        case x := <- somechan:
            // … 使用x进行一些操作

        case y, ok := <- someOtherchan:
            // … 使用y进行一些操作,
            // 检查ok值判断someOtherchan是否已经关闭

        case outputChan <- z:
            // … z值被成功发送到Channel上时

        default:
            // … 上面case均无法通信时,执行此分支
        }
}

【终结workers】

下面是一个常见的终结sub worker goroutines的方法,每个worker goroutine通过select监视一个die channel来及时获取main goroutine的退出通知。

//testterminateworker1.go
package main

import (
    "fmt"
    "time"
)

func worker(die chan bool, index int) {
    fmt.Println("Begin: This is Worker:", index)
    for {
        select {
        //case xx:
            //做事的分支
        case <-die:
            fmt.Println("Done: This is Worker:", index)
            return
        }
    }
}

func main() {
    die := make(chan bool)

    for i := 1; i <= 100; i++ {
        go worker(die, i)
    }

    time.Sleep(time.Second * 5)
    close(die)
    select {}
//deadlock we expected
}

【终结验证】

有时候终结一个worker后,main goroutine想确认worker routine是否真正退出了,可采用下面这种方法:

//testterminateworker2.go
package main

import (
    "fmt"
    //"time"
)

func worker(die chan bool) {
    fmt.Println("Begin: This is Worker")
    for {
        select {
        //case xx:
        //做事的分支
        case <-die:
            fmt.Println("Done: This is Worker")
            die <- true
            return
        }
    }
}

func main() {
    die := make(chan bool)

    go worker(die)

    die <- true
    <-die
    fmt.Println("Worker goroutine has been terminated")
}

【关闭的Channel永远不会阻塞】

下面演示在一个已经关闭了的channel上读写的结果:

//testoperateonclosedchannel.go
package main

import "fmt"

func main() {
        cb := make(chan bool)
        close(cb)
        x := <-cb
        fmt.Printf("%#v\n", x)

        x, ok := <-cb
        fmt.Printf("%#v %#v\n", x, ok)

        ci := make(chan int)
        close(ci)
        y := <-ci
        fmt.Printf("%#v\n", y)

        cb <- true
}

$go run testoperateonclosedchannel.go
false
false false
0
panic: runtime error: send on closed channel

可以看到在一个已经close的unbuffered channel上执行读操作,回返回channel对应类型的零值,比如bool型channel返回false,int型channel返回0。但向close的channel写则会触发panic。不过无论读写都不会导致阻塞。

【关闭带缓存的channel】

将unbuffered channel换成buffered channel会怎样?我们看下面例子:

//testclosedbufferedchannel.go
package main

import "fmt"

func main() {
        c := make(chan int, 3)
        c <- 15
        c <- 34
        c <- 65
        close(c)
        fmt.Printf("%d\n", <-c)
        fmt.Printf("%d\n", <-c)
        fmt.Printf("%d\n", <-c)
        fmt.Printf("%d\n", <-c)

        c <- 1
}

$go run testclosedbufferedchannel.go
15
34
65
0
panic: runtime error: send on closed channel

可以看出带缓冲的channel略有不同。尽管已经close了,但我们依旧可以从中读出关闭前写入的3个值。第四次读取时,则会返回该channel类型的零值。向这类channel写入操作也会触发panic。

【range】

Golang中的range常常和channel并肩作战,它被用来从channel中读取所有值。下面是一个简单的实例:

//testrange.go
package main

import "fmt"

func generator(strings chan string) {
        strings <- "Five hour's New York jet lag"
        strings <- "and Cayce Pollard wakes in Camden Town"
        strings <- "to the dire and ever-decreasing circles"
        strings <- "of disrupted circadian rhythm."
        close(strings)
}

func main() {
        strings := make(chan string)
        go generator(strings)
        for s := range strings {
                fmt.Printf("%s\n", s)
        }
        fmt.Printf("\n")
}

四、隐藏状态

下面通过一个例子来演示一下channel如何用来隐藏状态:

1、例子:唯一的ID服务

//testuniqueid.go
package main

import "fmt"

func newUniqueIDService() <-chan string {
        id := make(chan string)
        go func() {
                var counter int64 = 0
                for {
                        id <- fmt.Sprintf("%x", counter)
                        counter += 1
                }
        }()
        return id
}
func main() {
        id := newUniqueIDService()
        for i := 0; i < 10; i++ {
                fmt.Println(<-id)
        }
}

$ go run testuniqueid.go
0
1
2
3
4
5
6
7
8
9

newUniqueIDService通过一个channel与main goroutine关联,main goroutine无需知道uniqueid实现的细节以及当前状态,只需通过channel获得最新id即可。

五、默认情况

我想这里John Graham-Cumming主要是想告诉我们select的default分支的实践用法。

1、select  for non-blocking receive

idle:= make(chan []byte, 5) //用一个带缓冲的channel构造一个简单的队列

select {
case b = <-idle:
 //尝试从idle队列中读取
    …
default:  //队列空,分配一个新的buffer
        makes += 1
        b = make([]byte, size)
}

2、select for non-blocking send

idle:= make(chan []byte, 5) //用一个带缓冲的channel构造一个简单的队列

select {
case idle <- b: //尝试向队列中插入一个buffer
        //…
default: //队列满?

}

六、Nil Channels

1、nil channels阻塞

对一个没有初始化的channel进行读写操作都将发生阻塞,例子如下:

package main

func main() {
        var c chan int
        <-c
}

$go run testnilchannel.go
fatal error: all goroutines are asleep – deadlock!

package main

func main() {
        var c chan int
        c <- 1
}

$go run testnilchannel.go
fatal error: all goroutines are asleep – deadlock!

2、nil channel在select中很有用

看下面这个例子:

//testnilchannel_bad.go
package main

import "fmt"
import "time"

func main() {
        var c1, c2 chan int = make(chan int), make(chan int)
        go func() {
                time.Sleep(time.Second * 5)
                c1 <- 5
                close(c1)
        }()

        go func() {
                time.Sleep(time.Second * 7)
                c2 <- 7
                close(c2)
        }()

        for {
                select {
                case x := <-c1:
                        fmt.Println(x)
                case x := <-c2:
                        fmt.Println(x)
                }
        }
        fmt.Println("over")
}

我们原本期望程序交替输出5和7两个数字,但实际的输出结果却是:

5
0
0
0
… … 0死循环

再仔细分析代码,原来select每次按case顺序evaluate:
    – 前5s,select一直阻塞;
    – 第5s,c1返回一个5后被close了,“case x := <-c1”这个分支返回,select输出5,并重新select
    – 下一轮select又从“case x := <-c1”这个分支开始evaluate,由于c1被close,按照前面的知识,close的channel不会阻塞,我们会读出这个 channel对应类型的零值,这里就是0;select再次输出0;这时即便c2有值返回,程序也不会走到c2这个分支
    – 依次类推,程序无限循环的输出0

我们利用nil channel来改进这个程序,以实现我们的意图,代码如下:

//testnilchannel.go
package main

import "fmt"
import "time"

func main() {
        var c1, c2 chan int = make(chan int), make(chan int)
        go func() {
                time.Sleep(time.Second * 5)
                c1 <- 5
                close(c1)
        }()

        go func() {
                time.Sleep(time.Second * 7)
                c2 <- 7
                close(c2)
        }()

        for {
                select {
                case x, ok := <-c1:
                        if !ok {
                                c1 = nil
                        } else {
                                fmt.Println(x)
                        }
                case x, ok := <-c2:
                        if !ok {
                                c2 = nil
                        } else {
                                fmt.Println(x)
                        }
                }
                if c1 == nil && c2 == nil {
                        break
                }
        }
        fmt.Println("over")
}

$go run testnilchannel.go
5
7
over

可以看出:通过将已经关闭的channel置为nil,下次select将会阻塞在该channel上,使得select继续下面的分支evaluation。

七、Timers

1、超时机制Timeout

带超时机制的select是常规的tip,下面是示例代码,实现30s的超时select:

func worker(start chan bool) {
        timeout := time.After(30 * time.Second)
        for {
                select {
                     // … do some stuff
                case <- timeout:
                    return
                }
        }
}

2、心跳HeartBeart

与timeout实现类似,下面是一个简单的心跳select实现:

func worker(start chan bool) {
        heartbeat := time.Tick(30 * time.Second)
        for {
                select {
                     // … do some stuff
                case <- heartbeat:
                    //… do heartbeat stuff
                }
        }
}

如发现本站页面被黑,比如:挂载广告、挖矿等恶意代码,请朋友们及时联系我。十分感谢! Go语言第一课 Go语言精进之路1 Go语言精进之路2 Go语言编程指南
商务合作请联系bigwhite.cn AT aliyun.com

欢迎使用邮件订阅我的博客

输入邮箱订阅本站,只要有新文章发布,就会第一时间发送邮件通知你哦!

这里是 Tony Bai的个人Blog,欢迎访问、订阅和留言! 订阅Feed请点击上面图片

如果您觉得这里的文章对您有帮助,请扫描上方二维码进行捐赠 ,加油后的Tony Bai将会为您呈现更多精彩的文章,谢谢!

如果您希望通过微信捐赠,请用微信客户端扫描下方赞赏码:

如果您希望通过比特币或以太币捐赠,可以扫描下方二维码:

比特币:

以太币:

如果您喜欢通过微信浏览本站内容,可以扫描下方二维码,订阅本站官方微信订阅号“iamtonybai”;点击二维码,可直达本人官方微博主页^_^:
本站Powered by Digital Ocean VPS。
选择Digital Ocean VPS主机,即可获得10美元现金充值,可 免费使用两个月哟! 著名主机提供商Linode 10$优惠码:linode10,在 这里注册即可免费获 得。阿里云推荐码: 1WFZ0V立享9折!


View Tony Bai's profile on LinkedIn
DigitalOcean Referral Badge

文章

评论

  • 正在加载...

分类

标签

归档



View My Stats