1. 超时和取消
两种方式：
- done channel
- context.Context
相关内容参考：https://www.kancloud.cn/mutouzhang/go/596853
2. 心跳
两种不同类型的心跳：
- 以固定时间间隔产生的心跳
以固定时间间隔产生的心跳对于并发来说很有用，它可能在等待处理某个工作单元执行某个任务时发生，由于你不知道这项工作在什么时刻会执行，所以goroutine可能会持续等待。心跳是一种向监听者发出信号的方式，即一切都很好，当前静默是正常的。
// golang中通过time.Tick实现固定时间间隔的心跳机制
doWork := func(done <-chan interface{}, pulseInterval time.Duration) (<-chan interface{}, <-chan time.Time) {
heartbeat := make(chan interface{}) //1
results := make(chan time.Time)
go func() {
defer close(heartbeat)
defer close(results)
pulse := time.Tick(pulseInterval) //2：一个心跳周期
workGen := time.Tick(2 * pulseInterval) //3：两个心跳周期
sendPulse := func() {
select {
case heartbeat <- struct{}{}:
default: //4
}
}
sendResult := func(r time.Time) {
for {
select {
case <-done:
return
case <-pulse: //5
sendPulse()
case results <- r:
return
}
}
}
for {
select {
case <-done:
return
case <-pulse: //5
sendPulse() // 1个心跳周期
case r := <-workGen:
sendResult(r) // 2个心跳周期发送结果数据
}
}
}()
return heartbeat, results
}
// 消费程序
done := make(chan interface{})
time.AfterFunc(10*time.Second, func() { close(done) }) //1: 10s后，关闭done channel
const timeout = 2 * time.Second //2
heartbeat, results := doWork(done, timeout/2) //3
for { // for-select从通道中循环获取心跳包和结果数据
select {
case _, ok := <-heartbeat: //4
if ok == false {
return
}
fmt.Println("pulse")
case r, ok := <-results: //5
if ok == false {
return
}
fmt.Printf("results %v\n", r.Second())
case <-time.After(timeout): //6 : select中设置相应的超时机制，一方面控制忙读取速率，另一方面防止长时间阻塞等待
return
}
}
- 在工作单元开始时产生的心跳
doWork := func(done <-chan interface{}) (<-chan interface{}, <-chan int) {
heartbeatStream := make(chan interface{}, 1) //1
workStream := make(chan int)
go func() {
defer close(heartbeatStream)
defer close(workStream)
for i := 0; i < 10; i++ {
select { //2
case heartbeatStream <- struct{}{}:
default: //3
}
select {
case <-done:
return
case workStream <- rand.Intn(10):
}
}
}()
return heartbeatStream, workStream
}
done := make(chan interface{})
defer close(done)
heartbeat, results := doWork(done)
for {
select {
case _, ok := <-heartbeat:
if ok {
fmt.Println("pulse")
} else {
return
}
case r, ok := <-results:
if ok {
fmt.Printf("results %v\n", r)
} else {
return
}
}
}
3. 请求并发复制处理
复制请求，转发给多个处理单元进行处理，并获取最早的处理返回结果。
// done <-chan interface{} 和 result chan<- int : 范围访问约束
doWork := func(done <-chan interface{}, id int, wg *sync.WaitGroup, result chan<- int) {
started := time.Now()
defer wg.Done()
// 模拟随机加载
simulatedLoadTime := time.Duration(1+rand.Intn(5)) * time.Second
select {
case <-done:
case <-time.After(simulatedLoadTime): // 通过随机数模拟goroutine的处理时延
}
select {
case <-done: // 及时取消返回
case result <- id:
}
took := time.Since(started)
// 显示处理程序将花费多长时间
if took < simulatedLoadTime { // 当该goroutine通过done被通知取消执行时，实际处理时间会小于为其分配的处理时间
took = simulatedLoadTime
}
fmt.Printf("%v took %v\n", id, took)
}
done := make(chan interface{})
result := make(chan int)
var wg sync.WaitGroup
wg.Add(10)
for i := 0; i < 10; i++ { //1：开启10个goroutine来同时处理该次请求
go doWork(done, i, &wg, result)
}
firstReturned := <-result //2：取最早执行完请求的goroutine的返回结果
close(done) //3: done channel，广播通知其它goroutine取消处理该请求
wg.Wait()
fmt.Printf("Received an answer from #%v\n", firstReturned)
解决方式：每个处理请求的goroutine中监听done channel的广播信号，并在接收到done channel的广播信号时，立刻停止对该请求的处理。
PS：这种复制请求并发处理，并取最早返回时间的方式，在大型web系统中一般不会采用(处理放大)，在此处做下了解即可。另外，对于需要及时响应用户请求，在一些业务场景下可以走异步处理的流程。
4. 速率限制(限流)
- 令牌桶
- 漏桶
5. goroutine异常监控
- 子goroutine在退出时向主goroutine发送信息(正常信息或异常信息)，主goroutine在接收到子goroutine发送的信息后进行相应处理，包括重启子goroutine等。
- 主goroutine和子goroutine之间通过心跳包保活，主goroutine周期性地监控子goroutine的运行状态。
监控goroutine的异常退出(channel通信)：https://www.ctolib.com/topics-19119.html
