go语言心跳超时的实现示例

159 阅读 0 评论 105 点赞

我是靠谱客的博主友好小虾米，这篇文章主要介绍go语言心跳超时的实现示例，现在分享给大家，希望可以做个参考。

一、背景

本文描述的是客户端接收心跳信息的超时实现。心跳超时，或者接受信息超过限定时间在分布式系统中出现的次数比较多。常见的就有hadoop中节点超时，或者日志中出现timeout的字样。

在学习go语言中，我也根据go语言的机制实现了心跳超时的这个问题。踩过坑，趟过水。

二、心跳超时的实现

2.1 通过select case （设计概念比较多）

这种方法实现心跳，需要对go语言中的channel和select case 机制有所了解。select代码段中没有包含default条件时，会一直阻塞到有通道操作。

需要注意的是！！！！ select语言只会阻塞一次，且执行一次。如果需要多次判断，或者可能有多个case条件需要满足，那就需要增加for语句。

首先需要知道的是select是专为channel设计的，所以说每个case表达式都必须是包含操作通道的表达式。下面这段代码是描述了随机抽取一个channel发消息，正常情况下，不会触发超时。为了触发超时，注释掉通道发送数据操作。超时五秒，则触发超时。

复制代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
package main
 
import (
    "fmt"
    "math/rand"
    "time"
)
 
func main() {
    // 准备好三个通道。
    intChannels := [3]chan int{
        make(chan int, 1),
        make(chan int, 1),
        make(chan int, 1),
    }
    // 随机选择一个通道，并向它发送元素值。
    index := rand.Intn(3)
    fmt.Printf("The index: %dn", index)
 
    //‼️ 取消这行代码的注视，超时条件的选择就会触发。
    //intChannels[index] <- index
    // 哪一个通道中有可取的元素值，哪个对应的分支就会被执行。
    select {
    case <-intChannels[0]:
        fmt.Println("The first candidate case is selected.")
    case <-intChannels[1]:
        fmt.Println("The second candidate case is selected.")
    case elem := <-intChannels[2]:
        fmt.Printf("The third candidate case is selected, the element is %d.n", elem)
    case <-time.After(5 * time.Second):
        fmt.Println("timed out")
    }
}

2.2 通过time.sleep（简单有效）

通过time.sleep()实现超时操作，是比较巧妙的。一般来说心跳超时是一个双方交互的行为。

下面画一个图来描述一下。

为了方便理解，定义双方都使用共同时间。

下面是代码。

基本的逻辑是：

1、先给客户端设置一个下次超时的时间

2、客户端每次收到心跳的时候，更新这个时间

3、开启一个独立的线程，一致判断当前客户端是否超时。

ps：结合时效和性能，可以间隔一定的时间来进行判断。

复制代码

package main
 
import (
    "fmt"
    "sync"
    "time"
)
 
type Client struct {
    lock sync.Mutex //加锁
    nextTimeOutTime time.Time //下次超时时间
}
 
const tenSec = 10
/**
刷新每次的心跳超时机制
 */
func (client *Client) freshTimeOutTime()  {
    client.lock.Lock()
    defer client.lock.Unlock()
    client.nextTimeOutTime =time.Now().Add(tenSec*time.Second)
}
 
//开启一个gp，每隔500ms判断有没有超时
func (client *Client) judgeTimeOut()  {
    for  {
        time.Sleep(500*time.Millisecond)
        fmt.Printf("%v 在判断是否超时n", client.nextTimeOutTime)
        if time.Now().After(client.nextTimeOutTime) {
            fmt.Printf("%v 超时了n", client.nextTimeOutTime)
        }
    }
}
 
//客户端收到以后，修改下次心跳超时时间
func (client *Client) receiveHeart()  {
    client.freshTimeOutTime()
}
 
//开启一个模拟ping 客户端的线程
func pingClient(client *Client)  {
    for true {
        time.Sleep(11*time.Second)
        fmt.Printf("%v 请求发送时间n", time.Now())
        client.receiveHeart()
    }
 
}
 
func main() {
    client := Client{
        lock:            sync.Mutex{},
        nextTimeOutTime: time.Time{},
    }
    //在当前时刻，更新下次的超时时刻是10s中后
    client.freshTimeOutTime()
 
 
    go pingClient(&client)
 
 
    go client.judgeTimeOut()
 
    for true {
 
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
package main
 
import (
    "fmt"
    "sync"
    "time"
)
 
type Client struct {
    lock sync.Mutex //加锁
    nextTimeOutTime time.Time //下次超时时间
}
 
const tenSec = 10
/**
刷新每次的心跳超时机制
 */
func (client *Client) freshTimeOutTime()  {
    client.lock.Lock()
    defer client.lock.Unlock()
    client.nextTimeOutTime =time.Now().Add(tenSec*time.Second)
}
 
//开启一个gp，每隔500ms判断有没有超时
func (client *Client) judgeTimeOut()  {
    for  {
        time.Sleep(500*time.Millisecond)
        fmt.Printf("%v 在判断是否超时n", client.nextTimeOutTime)
        if time.Now().After(client.nextTimeOutTime) {
            fmt.Printf("%v 超时了n", client.nextTimeOutTime)
        }
    }
}
 
//客户端收到以后，修改下次心跳超时时间
func (client *Client) receiveHeart()  {
    client.freshTimeOutTime()
}
 
//开启一个模拟ping 客户端的线程
func pingClient(client *Client)  {
    for true {
        time.Sleep(11*time.Second)
        fmt.Printf("%v 请求发送时间n", time.Now())
        client.receiveHeart()
    }
 
}
 
func main() {
    client := Client{
        lock:            sync.Mutex{},
        nextTimeOutTime: time.Time{},
    }
    //在当前时刻，更新下次的超时时刻是10s中后
    client.freshTimeOutTime()
 
 
    go pingClient(&client)
 
 
    go client.judgeTimeOut()
 
    for true {
 
    }
}