长连接断开的原因
- 连接超时，浏览器自动断开连接
- 进程被杀死
- 不可抗拒因素
根据不同情况，高效保活的方式
- 连接超时：心跳机制
- 进程保活
- 断线重连
重点心跳机制
- 产物
- 心跳包
- 心跳应答
轮询与心跳区别
- 轮询一次相当于：建立一次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)
// 打印数据值
3nstime.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")总结
- 学到一招：对于有是接口的方法：直接去看相对应实现的源代码
原文地址：https://www.cnblogs.com/MyUniverse/p/11746159.html
时间： 2024-11-08 18:54:14