码迷,mamicode.com
首页 > 其他好文 > 详细

golang+sse+angular的心跳机制、angullar的轮询机制、time.Duration和time.NewTicker的学习

时间:2019-10-27 00:55:33      阅读:125      评论:0      收藏:0      [点我收藏+]

标签:stream   and   def   ice   ali   比较   key   rda   com   

长连接断开的原因

  • 连接超时,浏览器自动断开连接
  • 进程被杀死
  • 不可抗拒因素

根据不同情况,高效保活的方式

  • 连接超时:心跳机制
  • 进程保活
  • 断线重连

重点心跳机制

  • 产物
    • 心跳包
    • 心跳应答

轮询与心跳区别

  • 轮询一次相当于:建立一次TCP连接+断开连接
  • 心跳:在已有的连接上进行保活

心跳设计要点

  • 心跳包的规格(内容&大小)
  • 心跳发送间隔时间(按照项目的特性进行判断)
  • 断线重连机制(核心= 如何判断长连接的有效性)

心跳具体实现(基于sse的长连接)

  • 客户端做心跳机制:客户端长时间没有反应,使用心跳机制,证明客户端的存在
  • 服务端做心跳机制:服务端长时间没有反应,使用心跳机制,证明服务端还存在

  • 服务端做心跳机制

思考点:

  • 如何判断连接中断信号(单独的思考,在本次的代码中,没有用于跟心跳机制有关,以后有想法,会补上)
notify := w.(http.CloseNotifier).CloseNotify()
    // log.Println("notify:",<- notify) 会直接堵住的,因为notify它接收连接中断信号
go func(){
    // 太迷了,正确想法就是:只能接收异常的信号,就是网络中断的信号
    fmt.Println("接收连接中断信号")
    <-notify
    userData[r.RemoteAddr] = r.RemoteAddr
    offUser <- r.RemoteAddr
    log.Println(r.RemoteAddr,"just close")
}()
  • 如何将一一对应的客户端和服务端保存
// 接收发送给客户端数据
type RW struct{
    Rw http.ResponseWriter
    T time.Time
}
var rw = make(map[int64]*RW)

// 考虑使用map。记得当正确的数据发送给客户端之后要将对应的map键值删除
delete(rw,a)   // 当发送完之后,就要将这个客户端删除了。a时键值
  • 利用golang中的time.Ticker机制,监听是否有服务端等待,然后进行轮询保活。心跳机制重点(利用协程进行监听)
// 保活,心跳
    go func(){
        defer func(){
            if err := recover();err!=nil{
                fmt.Println(err)
            }
        }()
        fmt.Println("开启保活")
        keepAliveInterval := time.Duration(6000)
        fmt.Println(keepAliveInterval)
        ticker := time.NewTicker(3*time.Second)
        for {
            select{
            case <-ticker.C:
                fmt.Println("保活,心跳机制")
                t1 := time.Now()
                for _,value:= range rw{
                    fmt.Println(value)
                    if t1.Sub(value.T)>keepAliveInterval{
                        fmt.Println("进入保活")
                        f,ok:=value.Rw.(http.Flusher)
                        if !ok{
                            fmt.Fprintf(value.Rw,"不能用来做sse")
                            return
                        }
                        fmt.Fprintf(value.Rw,"data:请耐心等待,我正在努力的加载数据\n\n")
                        f.Flush()
                    }
                }
            }
        }
    }()

样例代码

server.go

package main

import(
    "fmt"
    "log"
    "time"
    "sync"
    "net/http"
)

// 接收发送给客户端数据
type RW struct{
    Rw http.ResponseWriter
    T time.Time
}

var offUser = make(chan string,0)
var userData = make(map[string]string)
var rw = make(map[int64]*RW)
var i int64 = 0
var lock sync.Mutex

func init(){
    log.SetFlags(log.Ltime|log.Lshortfile)
}

func sseService(w http.ResponseWriter,r *http.Request){
    var a int64  // 用来接收key值
    defer func(){
        if err := recover();err!=nil{
            fmt.Println(err)
        }
    }()
    
    lock.Lock()
    i++
    a=i
    lock.Unlock()
    // 提取get请求参数
    fmt.Println("a =",a)
    f,ok := w.(http.Flusher)
    if !ok{
        http.Error(w,"cannot support sse",http.StatusInternalServerError)
        return
    }

    // 用于监听客户端时候已经断开了连接
    notify := w.(http.CloseNotifier).CloseNotify()
    // log.Println("notify:",<- notify) 会直接堵住的,因为notify它接收网络中断信号
    go func(){
        fmt.Println("接收关闭信号")
        <-notify
        offUser <- r.RemoteAddr
        log.Println(r.RemoteAddr,"just close")
    }()

    w.Header().Set("Content-Type", "text/event-stream")
    w.Header().Set("Cache-Control", "no-cache")
    w.Header().Set("Connection", "keep-alive")
    w.Header().Set("Transfer-Encoding", "chunked")
    w.Header().Set("Access-Control-Allow-Origin","*")

    fmt.Fprintf(w,"data:welcome\n\n")
    f.Flush()

    // 将当前的w保存
    fmt.Println("心跳")
    t := time.Now()
    rr := &RW{Rw:w,T:t}
    fmt.Println("rr =",rr)
    rw[a] = rr 

    // 模拟服务端接收发送数据阻塞
    fmt.Println("模拟服务端发送数据阻塞")
    time.Sleep(time.Second*30)
    fmt.Fprintf(w,"data:12345加油\n\n")
    f.Flush()
    delete(rw,a)   // 当发送完之后,就要将这个客户端删除了
}

func testClose(w http.ResponseWriter,r *http.Request){
    fmt.Println("remoteAddr:",r.RemoteAddr) 
    fmt.Println("userData:",userData)   
    // 用于监听客户端时候已经断开了连接
    notify := w.(http.CloseNotifier).CloseNotify()
    go func(){
        fmt.Println("接收连接中断信号")
        <-notify
        userData[r.RemoteAddr] = r.RemoteAddr
        offUser <- r.RemoteAddr
        log.Println(r.RemoteAddr,"just close")
    }()
    time.Sleep(time.Second*1)
    fmt.Fprintln(w,"这里任意数字")
}


func main(){
    fmt.Println("sse1")

    // 获取中断的客户端
    go func(){
        fmt.Println("监听关闭的客户端")
        for{
            select{
            case user:=<-offUser:
                log.Println("userOff:",user)
            }
        }
    }()

    // 保活,心跳
    go func(){
        defer func(){
            if err := recover();err!=nil{
                fmt.Println(err)
            }
        }()
        fmt.Println("开启保活")
        keepAliveInterval := time.Duration(6000)
        fmt.Println(keepAliveInterval)
        ticker := time.NewTicker(3*time.Second)
        for {
            select{
            case <-ticker.C:
                fmt.Println("保活,心跳机制")
                t1 := time.Now()
                for _,value:= range rw{
                    fmt.Println(value)
                    if t1.Sub(value.T)>keepAliveInterval{
                        fmt.Println("进入保活")
                        f,ok:=value.Rw.(http.Flusher)
                        if !ok{
                            fmt.Fprintf(value.Rw,"不能用来做sse")
                            return
                        }
                        fmt.Fprintf(value.Rw,"data:请耐心等待,我正在努力的加载数据\n\n")
                        f.Flush()
                    }
                }
            }
        }
    }()

    http.HandleFunc("/sse",sseService)
    http.HandleFunc("/testClose",testClose)
    http.ListenAndServe(":8080",nil)
}

client(angular)

  sse(){
    let that = this
    if ("EventSource" in window){
      console.log("可以使用EventSource")
    }else{
      return
    }
    var url = "http://localhost:8080/sse?pid="+12345
    var es = new EventSource(url)
    // 监听事件
    // 连接事件
    es.onopen = function(e:any){
      console.log("我进来啦")
      console.log(e)
    }

    // message事件
    es.onmessage = function(e){
      that.Data = e.data
      if (e.data=="12345加油"){  // 后端通知前端结束发送信息
        console.log("12345加油,这是服务端正确想发送的数据")
        es.close()
      }else{
        console.log(e.data)
      }
    }
    es.addEventListener("error",(e:any)=>{
      // 这里的e要声明变量,否则回报没有readyState属性
      console.log("e.target",e.target)
      console.log("SSEERROR:",e.target.readyState)
      if(e.target.readyState == 0){
        // 重连
        console.log("Reconnecting...")
        es.close()  // 不开启服务端,直接关闭
      }
      if(e.target.readyState==2){
        // 放弃
        console.log("give up.")
      }
    },false);
  }

学习心跳机制附带的知识点

angular设置轮询

  • setInterval()方法重复调用一个函数或执行一个代码段,在每次调用之间具有固定的时间延迟
  • clearInterval()删除重复调用
 myTest = setInterval(()=>{
      var i:number = 1
      console.log("轮询还是心跳")
      if(i===4){
        return
      }
      i++
    },1500)  // 一旦实例化,就会直接运行

  test(){
   clearInterval(this.myTest)   // 清除重复运行函数
  }

time.Duration

  • Duration的基本单位是纳秒
  • 作用:打印时间时,根据最合适的时间单位打印;用于时间比较
keepAliveInterval := time.Duration(3) 

// 打印数据值
3ns

time.NewTicker

  • 创建一个轮询机制,规定隔一段时间处理一次函数
ticker := time.NewTicker(500 * time.Millisecond)
done := make(chan bool)

go func(){
    for{
        select{
        case <-done:
            return
        case t := <-ticker.C: // 500微秒轮询一次
            fmt.Println("Tick at",t)
        }
    }
}()

time.Sleep(10*time.Second)
ticker.Stop()
done<-true
fmt.Println("ticker stopper")

总结

  • 学到一招:对于有是接口的方法:直接去看相对应实现的源代码

golang+sse+angular的心跳机制、angullar的轮询机制、time.Duration和time.NewTicker的学习

标签:stream   and   def   ice   ali   比较   key   rda   com   

原文地址:https://www.cnblogs.com/MyUniverse/p/11746159.html

(0)
(0)
   
举报
评论 一句话评论(0
登录后才能评论!
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码!