websocket简介：
WebSocket 是一种网络传输协议，可在单个 TCP 连接上进行全双工通信，位于 OSI 模型的应用层。WebSocket 协议在 2011 年由 IETF 标准化为 RFC 6455，后由 RFC 7936 补充规范。
WebSocket 使得客户端和服务器之间的数据交换变得更加简单，允许服务端主动向客户端推送数据。在 WebSocket API 中，浏览器和服务器只需要完成一次握手，两者之间就可以创建持久性的连接，并进行双向数据传输。
理解各种协议和通信层、套接字的含义
IP：网络层协议；（高速公路）
TCP和UDP：传输层协议；（卡车）
HTTP：应用层协议；（货物）。HTTP(超文本传输协议)是建立在TCP协议之上的一种应用。HTTP连接最显著的特点是客户端发送的每次请求都需要服务器回送响应，在请求结束后，会主动释放连接。从建立连接到关闭连接的过程称为“一次连接”。
SOCKET：套接字，TCP/IP网络的API。(港口码头/车站)Socket是应用层与TCP/IP协议族通信的中间软件抽象层，它是一组接口。socket是在应用层和传输层之间的一个抽象层，它把TCP/IP层复杂的操作抽象为几个简单的接口供应用层调用已实现进程在网络中通信。
Websocket：同HTTP一样也是应用层的协议，但是它是一种双向通信协议，是建立在TCP之上的，解决了服务器与客户端全双工通信的问题，包含两部分:一部分是“握手”，一部分是“数据传输”。握手成功后，数据就直接从 TCP 通道传输，与 HTTP 无关了。
*注:什么是单工、半双工、全工通信？
数据只能单向传送为单工；
数据能双向传送但不能同时双向传送称为半双工；
数据能够同时双向传送则称为全双工。
上面是简单的接受了websocket情况以及和其他协议的区别及联系，在做之前，还是要了解下这块，对后期实战有帮助。
websocket开源选择：
在go语言中，websocket组件比较多的，可以到go仓库搜索下：
今天以仓库使用最多一个开源框架（gorilla）进行实战落地，以及讲解整个过程细节
websocket (github.com/gorilla/websocket)
上面介绍完整体开源情况。
业务背景：
1、数据实时推送到前端进行图形化显示
2、报警数据需要实时推送各端，例如web、cliet、安卓、ios等等其他客户端
我前几年一直从事的是java，可以看看我的播客，基本上与java有关，go也是最近两三周学习的，我其他播客有具体说明，因为公司业务需要，所有就简单的学了一下go语言，作为刚接触go不久的技术人员，如何面对新技术进行探索和落地的，大家可以跟着我的思路进行学习模仿，这样后期学习稍微比较快一些。
实战：
1、在实战之前，我们先看看官网使用说明：
这是最简单的，方式，没有其他业务，我们如何进行最佳实现呢？还是要看官网：
传送门：websocket/examples/chat at main · gorilla/websocket · GitHub
具体源码不说了，但是里面有几个重要点已经出现了，也是我们要学习的思想，这也是为什么要看开源代码的原因，要学习他们的思想和编码技巧。
稍微解释下Hub结构体里的字段：
// Registered clients.
// 这是保存客户端连接的信息，map，从这里可以看出来，所有的客户端都要保存到这里，这时可以想到，后期保存到redis里，从这里进行扩展即可。
clients map[*Client]bool
// Inbound messages from the clients.
// 这个就是广播数据了，但是demo了用了字节链，目的是了并行执行，提高效率，字节目的是接收所有情况的数据
broadcast chan []byte
// Register requests from the clients.
// 这个比较好理解了，是新客户端进行连接时触发的链条，为什么走链，也是为了并行执行，也就是异步执行
register chan *Client
// Unregister requests from clients.
//这个就是取消注册，也就是关闭客户端连接
unregister chan *Client
其实这几个字段已经把我们的框架整体搭建起来了，clients负责存储客户端，broadcast负责服务端发送给客户端数据的，register负责用来监听新建连接，unregister负责关闭客户端连接的
其实websocket也就是干这个事情的，例如websocket服务端收集所有客户端，然后根据需要进行发送消息给客户，然后就是关闭，大致流程就是这样的。
在看客户端怎么进行封装的：
这个就是针对上面的Hub进行组装结构体实例的，这里就不介绍了，本次实战也是根据这个来的，大家看懂这个基本上后续其他开源的websocket都没啥大问题。
接下来真正进行项目实战：
1、第一步创建websocket的客户端管理结构体：
// clientManager
// @Description: 客户端管理者
type clientManager struct {
//客户端存储的地方，我这边是用map进行存储，这块可以放到redis上，也是可以的，根据情况扩展即可
//这里可以使用map[string]*client,也是可以的，如果这样设计，方便后期进行匹配比较简单，直接匹配key即可，这种方式也是可以的，匹配客户端对象里的key
clients map[*client]bool
//广播数据链，进行业务限制，如果没有业务限制，直接使用[]byte 比较通用；用chan进行异步处理
broadcast chan model.BusinessDataWrapper
//客户端注册链；用chan进行异步处理
register chan *client
//客户端关闭链；用chan进行异步处理
unregister chan *client
}
结构体首字母小写，目的不用暴露出去，用于内部使用即可；每个字段不解释了，上面注释已经写好了，其实和官网是一样的。
2、客户端结构体：
// client
// @Description: 客户端信息
type client struct {
//每个客户端连接后都要进行生成唯一key，因为业务场景需求，不同的用户或设备接受的数据要一一对应，后期这块要会做权限控制
key string
//客户端连接对象
socket *websocket.Conn
//数据发送链
send chan []byte
}
3、编写启动方法，其实官网写的很像，稍微进行重构，更加符合当前项目
func (m *clientManager) start(callBackFunc func(c *client, manager *clientManager, businessDataWrapper model.BusinessDataWrapper) bool) {
for {
select {
case client := <-m.register: //进行连接
m.clients[client] = true
msg := "有一个新连接出现，已连接成功,客户端key：" + client.key
log.Info(ctx, msg)
//发送数据，这块可以不用发送，等后续进行注释即可
//m.send([]byte(msg), client)
case client := <-m.unregister: //关闭连接，进行释放资源
if _, ok := m.clients[client]; ok {
close(client.send)
delete(m.clients, client)
msg := "客户端key：" + client.key + ",已关闭"
log.Info(ctx, msg)
//m.send([]byte(msg), client)
}
case businessDataWrapper := <-m.broadcast: //广播数据
for client := range m.clients {
//这里其实就是发送数据了,
//进行转换成json字符串
//businessDataJsonStr, _ := json.Marshal(businessDataWrapper)
//broadCastSend(m, client, businessDataJsonStr)
//进行回调,我这里面目的为了后期这块不在调整代码了，在启动时进行业务处理，方便后期扩展用的，如果毕竟简单，直接使用broadCastSend(m, client, businessDataJsonStr)进行推送信息
callBackFunc(client, m, businessDataWrapper)
}
}
}
}
// 广播进行发送数据
func broadCastSend(manager *clientManager, client *client, businessDataJsonByte []byte) {
select {
case client.send <- businessDataJsonByte:
default:
fmt.Println("关闭连接了，，，，，")
close(client.send)
delete(manager.clients, client)
}
}
4、某一个客户端群发给其他客户端，排除自己客户端
//发送数据，这快类似群发消息
func (m *clientManager) send(message []byte, ignore *client) {
for client := range m.clients {
//ignore，这是忽略本身客户端，因为这是群发消息，自己可以不用接收了
if client != ignore {
//将数据写入到通道链
client.send <- message
}
}
}
5、推送数据，第四步是写入到发送通道里，还没真正发送，这块就是真正推送到客户端机制：
func (c *client) write(manager clientManager) {
defer func() {
manager.unregister <- c
c.socket.Close()
log.Info(ctx, c.key, "客户端进行关闭")
}()
for {
select {
//如果客户端有数据要进行写出去
case message, ok := <-c.send:
if !ok {
c.socket.WriteMessage(websocket.CloseMessage, []byte{})
log.Info(ctx, c.key, "发送关闭提示")
return
}
//这里才是真正的把数据推送到客户端
err := c.socket.WriteMessage(websocket.TextMessage, message)
if err != nil {
manager.unregister <- c
c.socket.Close()
log.Info(ctx, c.key, "数据写入失败，进行关闭！")
break
}
}
}
}
6、接收客户端发送的数据
func (c *client) read(manager clientManager) {
defer func() {
manager.unregister <- c
c.socket.Close()
log.Info(ctx, c.key, "客户端进行关闭")
}()
for {
_, message, err := c.socket.ReadMessage()
if err != nil {
manager.unregister <- c
c.socket.Close()
log.Info(ctx, c.key, "读数据出现异常，直接关闭。")
break
}
//后期可以注释掉
log.Info(ctx, c.key, "接收到客户端发送的数据", string(message))
//读到数据，进行业务操作，目前我这边项目只需要推送到客户端即可，所以暂时不做业务了，其他需要做业务，这里做个监听即可
}
}
7、提供创建客户端管理函数
// 创建客户端管理器
func newClientManager() *clientManager {
return &clientManager{
//广播数据，model.BusinessDataWrapper是我具体业务数据，可以换成[]byte接收
broadcast: make(chan model.BusinessDataWrapper),
register: make(chan *client),
unregister: make(chan *client),
clients: make(map[*client]bool),
}
}
以上就是整体的封装处理，可以看到和官方的demo很像，只是结合了一些业务场景而已，其他的都一样的。
接下来进行和业务进行集成：
1、创建管理器，上面也说了有两个业务场景, 一个是原始数据推送 ，另一个是报警数据推送,所以创建两个管理器出来：
// 原始ws客户端管理器
var rowDataManagerNew = newClientManager()
// 报警数据ws客户端管理器
var alarmDataManagerNew = newClientManager()
2、我们注册路由上，通过上面也能推断，需要指定两个路由路径
// 初始化websocket协议配置
var upgrader = websocket.Upgrader{
ReadBufferSize: 1024,
WriteBufferSize: 1024,
CheckOrigin: func(r *http.Request) bool { return true }, //允许跨域 、 允许同源
}
// registerRawDataClientConn
//
// @Author zhaosy
// @Description: 注册原始数据客户端连接
// @date 2024-07-16 18:12:19
func registerRawDataClientConn(w http.ResponseWriter, r *http.Request, businessType string, businessId string, userName string) {
if lang.IsEmpty(businessType) {
io.WriteString(w, "businessType 不能为空")
}
//生成客户端
conn, err := upgrader.Upgrade(w, r, nil)
if err != nil {
log.Error(ctx, err.Error(), err)
io.WriteString(w, "这是一个websocket连接,不是API.")
return
}
clientId := guid.S()
//这个key是我随机生成的一个key，包含了一些业务，大家根据需要进行设置，也可以随机生成就行，就是一行字符串，如果对key没有要求，其实key不用处理也行哈
key := websocketRowDataCachePrefix(businessType, businessId, userName, clientId)
//初始化客户端对象
client := &client{
key: key,
socket: conn,
send: make(chan []byte),
}
rowDataManagerNew.register <- client
//开启读
go client.read(*rowDataManagerNew)
//开起写
go client.write(*rowDataManagerNew)
}
// registerAlarmClient
//
// @Author zhaosy
// @Description: 注册报警客户端连接
// @date 2024-07-16 19:40:52
func registerAlarmClientConn(w http.ResponseWriter, r *http.Request) {
//生成客户端
conn, err := upgrader.Upgrade(w, r, nil)
if err != nil {
log.Error(ctx, err.Error(), err)
io.WriteString(w, "这是一个websocket,不是API.")
return
}
clientId := guid.S()
//这个key是我随机生成的一个key，包含了一些业务，大家根据需要进行设置，也可以随机生成就行，就是一行字符串，如果对key没有要求，其实key不用处理也行哈
key := websocketAlarmCachePrefix(clientId)
//初始化客户端对象
client := &client{
key: key,
socket: conn,
send: make(chan []byte),
}
alarmDataManagerNew.register <- client
//开启读
go client.read(*alarmDataManagerNew)
//开启写
go client.write(*alarmDataManagerNew)
}
上面看到
w http.ResponseWriter, r *http.Request 这两个参数应该就知道怎么做了吧，直接绑定到路由路由，也是官网那种方式
这是注册到go的http路由上了，后续通过path路径进行访问即可。
3、如何与我们的业务数据进行绑定？
还需要提供包函数出去
// SendRowDataBusinessData
//
// @Author zhaosy
// @Description: 接收业务数据进行推送到websocket
// @date 2024-07-16 18:19:46
func SendRowDataBusinessData(data model.BusinessData) {
rowDataManagerNew.broadcast <- model.SetRowDataWsWrapper(data)
}
// SendAlarmBusinessData
//
// @Author zhaosy
// @Description: 接收报警数据，推送到websocket客户端，这是我项目的业务，大家换成string即可
// @date 2024-07-16 19:42:13
func SendAlarmBusinessData(data model.AlarmBusinessData) {
alarmDataManagerNew.broadcast <- model.SetAlarmWsWrapper(data)
}
仅供参考。
这里是业务数据推送websocket的入口：
4、最后一步是管理器要启动了，启动前，大家知道我写了 回调函数，要进行实现下，具体业务了，所以大家参考即可：
func init() {
//启动原始数据websocket
go rowDataManagerNew.start(func(c *client, manager *clientManager, businessDataWrapper model.BusinessDataWrapper) bool {
if consts.ZERO == businessDataWrapper.DataType { //原始数据,推送
businessData := businessDataWrapper.BusinessData
if businessData.BusinessId == "" {
//进行转换成json字符串
businessDataJsonStr, _ := json.Marshal(businessData)
//广播所有客户端
broadCastSend(manager, c, businessDataJsonStr)
return true
}
//找到对应客户端 --可以通过拼接缓存key进行匹配也是可以的
if strings.Contains(c.key, websocketRowDataCachePrefix(businessData.BusinessType, businessData.BusinessId, "", "")) {
//进行转换成json字符串
businessDataJsonStr, _ := json.Marshal(businessData)
//广播指定客户端
broadCastSend(manager, c, businessDataJsonStr)
return true
}
return false
}
return false
})
//启动报警数据推送
go alarmDataManagerNew.start(func(c *client, manager *clientManager, businessDataWrapper model.BusinessDataWrapper) bool {
// 报警数据,推送
if consts.ONE == businessDataWrapper.DataType {
alarmBusinessData := businessDataWrapper.AlarmBusinessData
if alarmBusinessData.BusinessId != "" {
//找到对应客户端 --可以通过拼接缓存key进行匹配也是可以的
if strings.Contains(c.key, websocketAlarmCachePrefix("")) {
//进行转换成json字符串
alarmBusinessDataJsonStr, _ := json.Marshal(alarmBusinessData)
//广播指定客户端
broadCastSend(manager, c, alarmBusinessDataJsonStr)
return true
}
}
}
return false
})
}
这样就可以了，启动整体就没问题了，我这边用的是goframe框架，所以我单独提供了这两个：
func NewWs() *webSocket {
return &webSocket{}
}
type webSocket struct {
}
// RawDataWSHandle
//
// @Author zhaosy
// @Description: 原始数据websocket
// @date 2024-07-16 15:00:04
func (w *webSocket) RawDataWSHandle(r *ghttp.Request) {
//获取参数
businessType := r.Get("businessType")
businessId := r.Get("businessId")
userName := r.Get("userName")
registerRawDataClientConn(r.Response.BufferWriter, r.Request, businessType.String(), businessId.String(), userName.String())
}
// AlarmWSHandle
//
// @Author zhaosy
// @Description: 报警websocket处理器
// @date 2024-07-16 19:43:57
func (w *webSocket) AlarmWSHandle(r *ghttp.Request) {
registerAlarmClientConn(r.Response.BufferWriter, r.Request)
}
在goframe里cmd里进行绑定：
//websocket--原始数据websocket推送
s.BindHandler("/ws/rowdata/{businessType}/{businessId}/{userName}", websocket.NewWs().RawDataWSHandle)
//websocket--报警数据websocket推送
s.BindHandler("/ws/alarm", websocket.NewWs().AlarmWSHandle)
5、进行测试：
启动正常，日志也输出来了，进行测试
上面是连接正常，
通过业务数据进行测试：
以上就是本次研究的结果，整体上go的websocket比较简单，后面有机会，会重新进行重构，重构单独封装可以随意使用。
发一个整的代码：
// Package websocket
// @Author zhaosy
// @Date 2024/7/16 下午2:52:00
// @Desc websocket相关
package websocket
import (
"context"
"encoding/json"
"fmt"
"github.com/gogf/gf/v2/frame/g"
"github.com/gogf/gf/v2/net/ghttp"
"github.com/gogf/gf/v2/util/guid"
"github.com/gorilla/websocket"
"io"
"net/http"
"skynet/internal/consts"
"skynet/internal/model"
"skynet/utility/lang"
"strings"
)
var (
ctx = context.TODO()
log = g.Log()
)
func init() {
//启动原始数据websocket
go rowDataManagerNew.start(func(c *client, manager *clientManager, businessDataWrapper model.BusinessDataWrapper) bool {
if consts.ZERO == businessDataWrapper.DataType { //原始数据,推送
businessData := businessDataWrapper.BusinessData
if businessData.BusinessId == "" {
//进行转换成json字符串
businessDataJsonStr, _ := json.Marshal(businessData)
//广播所有客户端
broadCastSend(manager, c, businessDataJsonStr)
return true
}
//找到对应客户端 --可以通过拼接缓存key进行匹配也是可以的
if strings.Contains(c.key, websocketRowDataCachePrefix(businessData.BusinessType, businessData.BusinessId, "", "")) {
//进行转换成json字符串
businessDataJsonStr, _ := json.Marshal(businessData)
//广播指定客户端
broadCastSend(manager, c, businessDataJsonStr)
return true
}
return false
}
return false
})
//启动报警数据推送
go alarmDataManagerNew.start(func(c *client, manager *clientManager, businessDataWrapper model.BusinessDataWrapper) bool {
// 报警数据,推送
if consts.ONE == businessDataWrapper.DataType {
alarmBusinessData := businessDataWrapper.AlarmBusinessData
if alarmBusinessData.BusinessId != "" {
//找到对应客户端 --可以通过拼接缓存key进行匹配也是可以的
if strings.Contains(c.key, websocketAlarmCachePrefix("")) {
//进行转换成json字符串
alarmBusinessDataJsonStr, _ := json.Marshal(alarmBusinessData)
//广播指定客户端
broadCastSend(manager, c, alarmBusinessDataJsonStr)
return true
}
}
}
return false
})
}
func NewWs() *webSocket {
return &webSocket{}
}
// websocketRowDataCachePrefix
//
// @Author zhaosy
// @Description: 原始数据缓存前缀
// @date 2024-07-16 18:15:55
func websocketRowDataCachePrefix(businessType, businessId, userName, clientId string) string {
key := "websocket:rowdata"
if lang.IsNotEmpty(businessType) {
key = key + ":" + businessType
if lang.IsNotEmpty(businessId) {
key = key + ":" + businessId
if lang.IsNotEmpty(userName) {
key = key + ":" + userName
}
}
}
if lang.IsNotEmpty(clientId) {
key = key + ":" + clientId
}
return key
}
// websocketAlarmCachePrefix
//
// @Author zhaosy
// @Description: 报警数据前缀
// @date 2024-07-16 19:36:24
func websocketAlarmCachePrefix(clientId string) string {
//后期要加组织机构，有权限控制这块，需要进行处理，暂时先不去处理
key := "websocket:alarm"
if lang.IsNotEmpty(clientId) {
key = key + ":" + clientId
}
return key
}
type webSocket struct {
}
// RawDataWSHandle
//
// @Author zhaosy
// @Description: 原始数据websocket
// @date 2024-07-16 15:00:04
func (w *webSocket) RawDataWSHandle(r *ghttp.Request) {
//获取参数
businessType := r.Get("businessType")
businessId := r.Get("businessId")
userName := r.Get("userName")
registerRawDataClientConn(r.Response.BufferWriter, r.Request, businessType.String(), businessId.String(), userName.String())
}
// AlarmWSHandle
//
// @Author zhaosy
// @Description: 报警websocket处理器
// @date 2024-07-16 19:43:57
func (w *webSocket) AlarmWSHandle(r *ghttp.Request) {
registerAlarmClientConn(r.Response.BufferWriter, r.Request)
}
// SendRowDataBusinessData
//
// @Author zhaosy
// @Description: 接收业务数据进行推送到websocket
// @date 2024-07-16 18:19:46
func SendRowDataBusinessData(data model.BusinessData) {
rowDataManagerNew.broadcast <- model.SetRowDataWsWrapper(data)
}
// SendAlarmBusinessData
//
// @Author zhaosy
// @Description: 接收报警数据，推送到websocket客户端
// @date 2024-07-16 19:42:13
func SendAlarmBusinessData(data model.AlarmBusinessData) {
alarmDataManagerNew.broadcast <- model.SetAlarmWsWrapper(data)
}
// 初始化websocket协议配置
var upgrader = websocket.Upgrader{
ReadBufferSize: 1024,
WriteBufferSize: 1024,
CheckOrigin: func(r *http.Request) bool { return true }, //允许跨域 、 允许同源
}
// registerRawDataClientConn
//
// @Author zhaosy
// @Description: 注册原始数据客户端连接
// @date 2024-07-16 18:12:19
func registerRawDataClientConn(w http.ResponseWriter, r *http.Request, businessType string, businessId string, userName string) {
if lang.IsEmpty(businessType) {
io.WriteString(w, "businessType 不能为空")
}
//生成客户端
conn, err := upgrader.Upgrade(w, r, nil)
if err != nil {
log.Error(ctx, err.Error(), err)
io.WriteString(w, "这是一个websocket连接,不是API.")
return
}
clientId := guid.S()
key := websocketRowDataCachePrefix(businessType, businessId, userName, clientId)
//初始化客户端对象
client := &client{
key: key,
socket: conn,
send: make(chan []byte),
}
rowDataManagerNew.register <- client
//开启读
go client.read(*rowDataManagerNew)
//开起写
go client.write(*rowDataManagerNew)
}
// registerAlarmClient
//
// @Author zhaosy
// @Description: 注册报警客户端连接
// @date 2024-07-16 19:40:52
func registerAlarmClientConn(w http.ResponseWriter, r *http.Request) {
//生成客户端
conn, err := upgrader.Upgrade(w, r, nil)
if err != nil {
log.Error(ctx, err.Error(), err)
io.WriteString(w, "这是一个websocket,不是网站.")
return
}
clientId := guid.S()
key := websocketAlarmCachePrefix(clientId)
//初始化客户端对象
client := &client{
key: key,
socket: conn,
send: make(chan []byte),
}
alarmDataManagerNew.register <- client
//开启读
go client.read(*alarmDataManagerNew)
//开启写
go client.write(*alarmDataManagerNew)
}
// 原始ws客户端管理器
var rowDataManagerNew = newClientManager()
// 报警数据ws客户端管理器
var alarmDataManagerNew = newClientManager()
// **********************************以下是websocket进行封装，可以直接使用******************************************
// 创建客户端管理器
func newClientManager() *clientManager {
return &clientManager{
broadcast: make(chan model.BusinessDataWrapper),
register: make(chan *client),
unregister: make(chan *client),
clients: make(map[*client]bool),
}
}
// clientManager
// @Description: 客户端管理者
type clientManager struct {
//客户端存储的地方，我这边是用map进行存储，这块可以放到redis上，也是可以的，根据情况扩展即可
//这里可以使用map[string]*client,也是可以的，如果这样设计，方便后期进行匹配比较简单，直接匹配key即可，这种方式也是可以的，匹配客户端对象里的key
clients map[*client]bool
//广播数据链，进行业务限制，如果没有业务限制，直接使用[]byte 比较通用；用chan进行异步处理
broadcast chan model.BusinessDataWrapper
//客户端注册链；用chan进行异步处理
register chan *client
//客户端关闭链；用chan进行异步处理
unregister chan *client
}
// client
// @Description: 客户端信息
type client struct {
//每个客户端连接后都要进行生成唯一key，因为业务场景需求，不同的用户或设备接受的数据要一一对应，后期这块要会做权限控制
key string
//客户端连接对象
socket *websocket.Conn
//数据发送链
send chan []byte
}
// start
//
// @Author zhaosy
// @Description: websocket启动
// @date 2024-07-17 10:55:28
func (m *clientManager) start(callBackFunc func(c *client, manager *clientManager, businessDataWrapper model.BusinessDataWrapper) bool) {
for {
select {
case client := <-m.register: //进行连接
m.clients[client] = true
msg := "有一个新连接出现，已连接成功,客户端key：" + client.key
log.Info(ctx, msg)
//发送数据，这块可以不用发送，等后续进行注释即可
//m.send([]byte(msg), client)
case client := <-m.unregister: //关闭连接，进行释放资源
if _, ok := m.clients[client]; ok {
close(client.send)
delete(m.clients, client)
msg := "客户端key：" + client.key + ",已关闭"
log.Info(ctx, msg)
//m.send([]byte(msg), client)
}
case businessDataWrapper := <-m.broadcast: //广播数据
for client := range m.clients {
//这里其实就是发送数据了,
//进行转换成json字符串
//businessDataJsonStr, _ := json.Marshal(businessDataWrapper)
//broadCastSend(m, client, businessDataJsonStr)
//进行回调,我这里面目的为了后期这块不在调整代码了，在启动时进行业务处理，方便后期扩展用的，如果毕竟简单，直接使用broadCastSend(m, client, businessDataJsonStr)进行推送信息
callBackFunc(client, m, businessDataWrapper)
}
}
}
}
// 广播进行发送数据
func broadCastSend(manager *clientManager, client *client, businessDataJsonByte []byte) {
select {
case client.send <- businessDataJsonByte:
default:
fmt.Println("关闭连接了，，，，，")
close(client.send)
delete(manager.clients, client)
}
}
// send
//
// @Author zhaosy
// @Description: 这快类似群发消息
// @date 2024-07-16 16:40:37
func (m *clientManager) send(message []byte, ignore *client) {
for client := range m.clients {
//ignore，这是忽略本身客户端，因为这是群发消息，自己可以不用接收了
if client != ignore {
//将数据写入到通道链
client.send <- message
}
}
}
func (c *client) read(manager clientManager) {
defer func() {
manager.unregister <- c
c.socket.Close()
log.Info(ctx, c.key, "客户端进行关闭")
}()
for {
_, message, err := c.socket.ReadMessage()
if err != nil {
manager.unregister <- c
c.socket.Close()
log.Info(ctx, c.key, "读数据出现异常，直接关闭。")
break
}
//后期可以注释掉
log.Info(ctx, c.key, "接收到客户端发送的数据", string(message))
//读到数据，进行业务操作，目前我这边项目只需要推送到客户端即可，所以暂时不做业务了，其他需要做业务，这里做个监听即可
}
}
// write
//
// @Author zhaosy
// @Description: 写入数据
// @date 2024-07-16 16:52:47
func (c *client) write(manager clientManager) {
defer func() {
manager.unregister <- c
c.socket.Close()
log.Info(ctx, c.key, "客户端进行关闭")
}()
for {
select {
//如果客户端有数据要进行写出去
case message, ok := <-c.send:
if !ok {
c.socket.WriteMessage(websocket.CloseMessage, []byte{})
log.Info(ctx, c.key, "发送关闭提示")
return
}
//这里才是真正的把数据推送到客户端
err := c.socket.WriteMessage(websocket.TextMessage, message)
if err != nil {
manager.unregister <- c
c.socket.Close()
log.Info(ctx, c.key, "数据写入失败，进行关闭！")
break
}
}
}
}
ok。结束
本文转载自: https://blog.csdn.net/u011410254/article/details/140489357
版权归原作者 云游遍天下 所有， 如有侵权，请联系我们删除。
版权归原作者 云游遍天下 所有， 如有侵权，请联系我们删除。