golang网络编程之基于TCP协议实现长连接 golang心跳检测

0 1142
索鸟 2021-01-28
需要:0索币
package main

// golang实现带有心跳检测的tcp长连接
// server
import (
	"fmt"
	"net"
	"time"
)

// message struct:
// c#d

var (
	Req_REGISTER byte = 1 // 1 --- c register cid
	Res_REGISTER byte = 2 // 2 --- s response

	Req_HEARTBEAT byte = 3 // 3 --- s send heartbeat req
	Res_HEARTBEAT byte = 4 // 4 --- c send heartbeat res

	Req byte = 5 // 5 --- cs send data
	Res byte = 6 // 6 --- cs send ack
)

type CS struct {
	Rch chan []byte
	Wch chan []byte
	Dch chan bool
	u   string
}

func NewCs(uid string) *CS {
	return &CS{Rch: make(chan []byte), Wch: make(chan []byte), u: uid}
}

var CMap map[string]*CS

func main() {
	CMap = make(map[string]*CS)
	listen, err := net.ListenTCP("tcp", &net.TCPAddr{net.ParseIP("127.0.0.1"), 6666, ""})
	if err != nil {
		fmt.Println("监听端口失败:", err.Error())
		return
	}
	fmt.Println("已初始化连接,等待客户端连接...")
	go PushGRT()
	Server(listen)
	select {}
}

func PushGRT() {
	for {
		time.Sleep(15 * time.Second)
		for k, v := range CMap {
			fmt.Println("push msg to user:" + k)
			v.Wch <- []byte{Req, '#', 'p', 'u', 's', 'h', '!'}
		}
	}
}

func Server(listen *net.TCPListener) {
	for {
		conn, err := listen.AcceptTCP()
		if err != nil {
			fmt.Println("接受客户端连接异常:", err.Error())
			continue
		}
		fmt.Println("客户端连接来自:", conn.RemoteAddr().String())
		// handler goroutine
		go Handler(conn)
	}
}

func Handler(conn net.Conn) {
	defer conn.Close()
	data := make([]byte, 128)
	var uid string
	var C *CS
	for {
		conn.Read(data)
		fmt.Println("客户端发来数据:", string(data))
		if data[0] == Req_REGISTER { // register
			conn.Write([]byte{Res_REGISTER, '#', 'o', 'k'})
			uid = string(data[2:])
			C = NewCs(uid)
			CMap[uid] = C
			//			fmt.Println("register client")
			//			fmt.Println(uid)
			break
		} else {
			conn.Write([]byte{Res_REGISTER, '#', 'e', 'r'})
		}
	}
	//	WHandler
	go WHandler(conn, C)

	//	RHandler
	go RHandler(conn, C)

	//	Worker
	go Work(C)
	select {
	case <-C.Dch:
		fmt.Println("close handler goroutine")
	}
}

// 正常写数据
// 定时检测 conn die => goroutine die
func WHandler(conn net.Conn, C *CS) {
	// 读取业务Work 写入Wch的数据
	ticker := time.NewTicker(20 * time.Second)
	for {
		select {
		case d := <-C.Wch:
			conn.Write(d)
		case <-ticker.C:
			if _, ok := CMap[C.u]; !ok {
				fmt.Println("conn die, close WHandler")
				return
			}
		}
	}
}

// 读客户端数据 + 心跳检测
func RHandler(conn net.Conn, C *CS) {
	// 心跳ack
	// 业务数据 写入Wch

	for {
		data := make([]byte, 128)
		// setReadTimeout
		err := conn.SetReadDeadline(time.Now().Add(10 * time.Second))
		if err != nil {
			fmt.Println(err)
		}
		if _, derr := conn.Read(data); derr == nil {
			// 可能是来自客户端的消息确认
			//           	     数据消息
			fmt.Println(data)
			if data[0] == Res {
				fmt.Println("recv client data ack")
			} else if data[0] == Req {
				fmt.Println("recv client data")
				fmt.Println(data)
				conn.Write([]byte{Res, '#'})
				// C.Rch <- data
			}

			continue
		}

		conn.Write([]byte{Req_HEARTBEAT, '#'})
		fmt.Println("send ht packet")
		conn.SetReadDeadline(time.Now().Add(2 * time.Second))
		if _, herr := conn.Read(data); herr == nil {
			// fmt.Println(string(data))
			fmt.Println("resv ht packet ack")
		} else {
			delete(CMap, C.u)
			fmt.Println("delete user!")
			return
		}
	}
}

func Work(C *CS) {
	time.Sleep(5 * time.Second)
	C.Wch <- []byte{Req, '#', 'h', 'e', 'l', 'l', 'o'}

	time.Sleep(15 * time.Second)
	C.Wch <- []byte{Req, '#', 'h', 'e', 'l', 'l', 'o'}
	// 从读ch读信息
	/*	ticker := time.NewTicker(20 * time.Second)
		for {
			select {
			case d := <-C.Rch:
				C.Wch <- d
			case <-ticker.C:
				if _, ok := CMap[C.u]; !ok {
					return
				}
			}

		}
	*/ // 往写ch写信息
}
package main

// golang实现带有心跳检测的tcp长连接
// server

import (
	"fmt"
	"net"
)

var (
	Req_REGISTER byte = 1 // 1 --- c register cid
	Res_REGISTER byte   = 2 // 2 --- s response

	Req_HEARTBEAT byte = 3 // 3 --- s send heartbeat req
	Res_HEARTBEAT byte = 4 // 4 --- c send heartbeat res

	Req  byte = 5 // 5 --- cs send data
	Res  byte = 6 // 6 --- cs send ack
)

var Dch chan bool
var Rch chan []byte
var Wch chan []byte

func main() {
	Dch = make(chan bool)
	Rch = make(chan []byte)
	Wch = make(chan []byte)
	addr, err := net.ResolveTCPAddr("tcp", "127.0.0.1:6666")
	conn, err := net.DialTCP("tcp", nil, addr)
//	conn, err := net.Dial("tcp", "127.0.0.1:6666")
	if err != nil {
		fmt.Println("连接服务端失败:", err.Error())
		return
	}
	fmt.Println("已连接服务器")
	defer conn.Close()
	go Handler(conn)
	select {
	    case <- Dch:
		    fmt.Println("关闭连接")
	}
}

func Handler(conn *net.TCPConn) {
	// 直到register ok
	data := make([]byte, 128)
	for {
		conn.Write([]byte{Req_REGISTER, '#', '2'})
		conn.Read(data)
//		fmt.Println(string(data))
		if data[0] == Res_REGISTER {
			break
		}
	}
//	fmt.Println("i'm register")
	go RHandler(conn)
	go WHandler(conn)
	go Work()
}

func RHandler(conn *net.TCPConn) {

	for {
		// 心跳包,回复ack
	data := make([]byte, 128)
		i,_ := conn.Read(data)
		if i == 0 {
			Dch <- true
			return
		}
		if data[0] == Req_HEARTBEAT {
			fmt.Println("recv ht pack")
			conn.Write([]byte{Res_REGISTER,'#','h'})
			fmt.Println("send ht pack ack")
		} else if data[0] == Req {
			fmt.Println("recv data pack")
			fmt.Printf("%v\n",string(data[2:]))
			Rch <- data[2:]
			conn.Write([]byte{Res,'#'})
		}
	}
}

func WHandler(conn net.Conn) {
	for {
		select {
			case msg := <- Wch:
				fmt.Println((msg[0]))
				fmt.Println("send data after: " + string(msg[1:]))
				conn.Write(msg)
		}
	}

}

func Work() {
	for {
		select {
		case msg := <- Rch:
				fmt.Println("work recv " + string(msg))
				Wch <- []byte{Req,'#','x','x','x','x','x'}
		}
	}

}

来源:https://my.oschina.net/sharelinux/blog/699725

回帖
  • 消灭零回复
相关主题
2020年最新最新Kubernetes视频教程(K8s)教程 2
程序员转型之制作网课变现,月入过万告别996 1
索鸟快传2.0发布啦 1
两个不同网络的电脑怎么实现文件的互相访问呢? 1
网盘多账号登录软件 1
Java实战闲云旅游项目基于vue+element-ui 1
单点登录技术解决方案基于OAuth2.0的网关鉴权RSA算法生成令牌 1
QT5获取剪贴板上文本信息QT设置剪贴板内容 1
springboot2实战在线购物系统电商系统 1
python web实战之爱家租房项目 1
windows COM实用入门教程 1
C++游戏开发之C++实现的水果忍者游戏 1
计算机视觉库opencv教程 1
node.js实战图书管理系统express框架实现 1
C++实战教程之远程桌面远程控制实战 1
相关主题
PHP7报A non well formed numeric value encountered 0
Linux系统下关闭mongodb的几种命令分享 0
mongodb删除数据、删除集合、删除数据库的命令 0
Git&Github极速入门与攻坚实战课程 0
python爬虫教程使用Django和scrapy实现 0
libnetsnmpmibs.so.31: cannot open shared object file 0
数据结构和算法视频教程 0
redis的hash结构怎么删除数据呢? 0
C++和LUA解析器的数据交互实战视频 0
mongodb errmsg" : "too many users are authenticated 0
C++基础入门视频教程 0
用30个小时精通C++视频教程可能吗? 0
C++分布式多线程游戏服务器开发视频教程socket tcp boost库 0
C++培训教程就业班教程 0
layui的util工具格式时间戳为字符串 0
C++实战教程之远程桌面远程控制实战 1
网络安全培训视频教程 0
LINUX_C++软件工程师视频教程高级项目实战 0
C++高级数据结构与算法视频教程 0
跨域问题很头疼?通过配置nginx轻松解决ajax跨域问题 0