通道,超时和计时器(Ticker)
time 包中有一些有趣的功能可以和通道组合使用。
其中就包含了 time.Ticker 结构体,这个对象以指定的时间间隔重复的向通道 C 发送时间值:
1
type Ticker struct {
2
C <-chan Time // the channel on which the ticks are delivered.
3
// contains filtered or unexported fields
4
...
5
}
Copied!
时间间隔的单位是 ns(纳秒,int64),在工厂函数 time.NewTicker 中以 Duration 类型的参数传入:func Newticker(dur) *Ticker
在协程周期性的执行一些事情(打印状态日志,输出,计算等等)的时候非常有用。
调用 Stop() 使计时器停止,在 defer 语句中使用。这些都很好的适应 select 语句:
1
ticker := time.NewTicker(updateInterval)
2
defer ticker.Stop()
3
...
4
select {
5
case u:= <-ch1:
6
...
7
case v:= <-ch2:
8
...
9
case <-ticker.C:
10
logState(status) // call some logging function logState
11
default: // no value ready to be received
12
...
13
}
Copied!
time.Tick() 函数声明为 Tick(d Duration) <-chan Time,当你想返回一个通道而不必关闭它的时候这个函数非常有用:它以 d 为周期给返回的通道发送时间,d是纳秒数。如果需要像下边的代码一样,限制处理频率(函数 client.Call() 是一个 RPC 调用,这里暂不赘述(参见第 15.9 节):
1
import "time"
2
3
rate_per_sec := 10
4
var dur Duration = 1e9 / rate_per_sec
5
chRate := time.Tick(dur) // a tick every 1/10th of a second
6
for req := range requests {
7
<- chRate // rate limit our Service.Method RPC calls
8
go client.Call("Service.Method", req, ...)
9
}
Copied!
这样只会按照指定频率处理请求:chRate 阻塞了更高的频率。每秒处理的频率可以根据机器负载(和/或)资源的情况而增加或减少。
问题 14.1:扩展上边的代码,思考如何承载周期请求数的暴增(提示:使用带缓冲通道和计时器对象)。
定时器(Timer)结构体看上去和计时器(Ticker)结构体的确很像(构造为 NewTimer(d Duration)),但是它只发送一次时间,在 Dration d 之后。
还有 time.After(d) 函数,声明如下:
1
func After(d Duration) <-chan Time
Copied!
Duration d 之后,当前时间被发到返回的通道;所以它和 NewTimer(d).C 是等价的;它类似 Tick(),但是 After() 只发送一次时间。下边有个很具体的示例,很好的阐明了 selectdefault 的作用:
示例 14.11:timer_goroutine.go
1
package main
2
3
import (
4
"fmt"
5
"time"
6
)
7
8
func main() {
9
tick := time.Tick(1e8)
10
boom := time.After(5e8)
11
for {
12
select {
13
case <-tick:
14
fmt.Println("tick.")
15
case <-boom:
16
fmt.Println("BOOM!")
17
return
18
default:
19
fmt.Println(" .")
20
time.Sleep(5e7)
21
}
22
}
23
}
Copied!
输出:
1
.
2
.
3
tick.
4
.
5
.
6
tick.
7
.
8
.
9
tick.
10
.
11
.
12
tick.
13
.
14
.
15
tick.
16
BOOM!
Copied!
习惯用法:简单超时模式
要从通道 ch 中接收数据,但是最多等待1秒。先创建一个信号通道,然后启动一个 lambda 协程,协程在给通道发送数据之前是休眠的:
1
timeout := make(chan bool, 1)
2
go func() {
3
time.Sleep(1e9) // one second
4
timeout <- true
5
}()
Copied!
然后使用 select 语句接收 ch 或者 timeout 的数据:如果 ch 在 1 秒内没有收到数据,就选择到了 time 分支并放弃了 ch 的读取。
1
select {
2
case <-ch:
3
// a read from ch has occured
4
case <-timeout:
5
// the read from ch has timed out
6
break
7
}
Copied!
第二种形式:取消耗时很长的同步调用
也可以使用 time.After() 函数替换 timeout-channel。可以在 select 中通过 time.After() 发送的超时信号来停止协程的执行。以下代码,在 timeoutNs 纳秒后执行 selecttimeout 分支后,执行client.Call 的协程也随之结束,不会给通道 ch 返回值:
1
ch := make(chan error, 1)
2
go func() { ch <- client.Call("Service.Method", args, &reply) } ()
3
select {
4
case resp := <-ch
5
// use resp and reply
6
case <-time.After(timeoutNs):
7
// call timed out
8
break
9
}
Copied!
注意缓冲大小设置为 1 是必要的,可以避免协程死锁以及确保超时的通道可以被垃圾回收。此外,需要注意在有多个 case 符合条件时, selectcase 的选择是伪随机的,如果上面的代码稍作修改如下,则 select 语句可能不会在定时器超时信号到来时立刻选中 time.After(timeoutNs) 对应的 case,因此协程可能不会严格按照定时器设置的时间结束。
1
ch := make(chan int, 1)
2
go func() { for { ch <- 1 } } ()
3
L:
4
for {
5
select {
6
case <-ch:
7
// do something
8
case <-time.After(timeoutNs):
9
// call timed out
10
break L
11
}
12
}
Copied!
第三种形式:假设程序从多个复制的数据库同时读取。只需要一个答案,需要接收首先到达的答案,Query 函数获取数据库的连接切片并请求。并行请求每一个数据库并返回收到的第一个响应:
1
func Query(conns []conn, query string) Result {
2
ch := make(chan Result, 1)
3
for _, conn := range conns {
4
go func(c Conn) {
5
select {
6
case ch <- c.DoQuery(query):
7
default:
8
}
9
}(conn)
10
}
11
return <- ch
12
}
Copied!
再次声明,结果通道 ch 必须是带缓冲的:以保证第一个发送进来的数据有地方可以存放,确保放入的首个数据总会成功,所以第一个到达的值会被获取而与执行的顺序无关。正在执行的协程可以总是可以使用 runtime.Goexit() 来停止。
在应用中缓存数据:
应用程序中用到了来自数据库(或者常见的数据存储)的数据时,经常会把数据缓存到内存中,因为从数据库中获取数据的操作代价很高;如果数据库中的值不发生变化就没有问题。但是如果值有变化,我们需要一个机制来周期性的从数据库重新读取这些值:缓存的值就不可用(过期)了,而且我们也不希望用户看到陈旧的数据。

链接

Last modified 3yr ago
Copy link
Contents
链接