创建一个websocket的服务端
|
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
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
|
package smileimport ( "errors" "log" "net/http" "sync" "time" "github.com/gorilla/websocket")const ( // 允许等待的写入时间 writeWait = 10 * time.Second // Time allowed to read the next pong message from the peer. pongWait = 60 * time.Second // Send pings to peer with this period. Must be less than pongWait. pingPeriod = (pongWait * 9) / 10 // Maximum message size allowed from peer. maxMessageSize = 512)// 最大的连接ID，每次连接都加1 处理var maxConnId int64// 客户端读写消息type wsMessage struct { // websocket.TextMessage 消息类型 messageType int data []byte}// ws 的所有连接// 用于广播var wsConnAll map[int64]*wsConnectionvar upgrader = websocket.Upgrader{ ReadBufferSize: 1024, WriteBufferSize: 1024, // 允许所有的CORS 跨域请求，正式环境可以关闭 CheckOrigin: func(r *http.Request) bool { return true },}// 客户端连接type wsConnection struct { wsSocket *websocket.Conn // 底层websocket inChan chan *wsMessage // 读队列 outChan chan *wsMessage // 写队列 mutex sync.Mutex // 避免重复关闭管道,加锁处理 isClosed bool closeChan chan byte // 关闭通知 id int64}func wsHandler(resp http.ResponseWriter, req *http.Request) { // 应答客户端告知升级连接为websocket wsSocket, err := upgrader.Upgrade(resp, req, nil) if err != nil { log.Println("升级为websocket失败", err.Error()) return } maxConnId++ // TODO 如果要控制连接数可以计算，wsConnAll长度 // 连接数保持一定数量，超过的部分不提供服务 wsConn := &wsConnection{ wsSocket: wsSocket, inChan: make(chan *wsMessage, 1000), outChan: make(chan *wsMessage, 1000), closeChan: make(chan byte), isClosed: false, id: maxConnId, } wsConnAll[maxConnId] = wsConn log.Println("当前在线人数", len(wsConnAll)) // 处理器,发送定时信息，避免意外关闭 go wsConn.processLoop() // 读协程 go wsConn.wsReadLoop() // 写协程 go wsConn.wsWriteLoop()}// 处理队列中的消息func (wsConn *wsConnection) processLoop() { // 处理消息队列中的消息 // 获取到消息队列中的消息，处理完成后，发送消息给客户端 for { msg, err := wsConn.wsRead() if err != nil { log.Println("获取消息出现错误", err.Error()) break } log.Println("接收到消息", string(msg.data)) // 修改以下内容把客户端传递的消息传递给处理程序 err = wsConn.wsWrite(msg.messageType, msg.data) if err != nil { log.Println("发送消息给客户端出现错误", err.Error()) break } }}// 处理消息队列中的消息func (wsConn *wsConnection) wsReadLoop() { // 设置消息的最大长度 wsConn.wsSocket.SetReadLimit(maxMessageSize) wsConn.wsSocket.SetReadDeadline(time.Now().Add(pongWait)) for { // 读一个message msgType, data, err := wsConn.wsSocket.ReadMessage() if err != nil { websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway, websocket.CloseAbnormalClosure) log.Println("消息读取出现错误", err.Error()) wsConn.close() return } req := &wsMessage{ msgType, data, } // 放入请求队列,消息入栈 select { case wsConn.inChan <- req: case <-wsConn.closeChan: return } }}// 发送消息给客户端func (wsConn *wsConnection) wsWriteLoop() { ticker := time.NewTicker(pingPeriod) defer func() { ticker.Stop() }() for { select { // 取一个应答 case msg := <-wsConn.outChan: // 写给websocket if err := wsConn.wsSocket.WriteMessage(msg.messageType, msg.data); err != nil { log.Println("发送消息给客户端发生错误", err.Error()) // 切断服务 wsConn.close() return } case <-wsConn.closeChan: // 获取到关闭通知 return case <-ticker.C: // 出现超时情况 wsConn.wsSocket.SetWriteDeadline(time.Now().Add(writeWait)) if err := wsConn.wsSocket.WriteMessage(websocket.PingMessage, nil); err != nil { return } } }}// 写入消息到队列中func (wsConn *wsConnection) wsWrite(messageType int, data []byte) error { select { case wsConn.outChan <- &wsMessage{messageType, data}: case <-wsConn.closeChan: return errors.New("连接已经关闭") } return nil}// 读取消息队列中的消息func (wsConn *wsConnection) wsRead() (*wsMessage, error) { select { case msg := <-wsConn.inChan: // 获取到消息队列中的消息 return msg, nil case <-wsConn.closeChan: } return nil, errors.New("连接已经关闭")}// 关闭连接func (wsConn *wsConnection) close() { log.Println("关闭连接被调用了") wsConn.wsSocket.Close() wsConn.mutex.Lock() defer wsConn.mutex.Unlock() if wsConn.isClosed == false { wsConn.isClosed = true // 删除这个连接的变量 delete(wsConnAll, wsConn.id) close(wsConn.closeChan) }}// 启动程序func StartWebsocket(addrPort string) { wsConnAll = make(map[int64]*wsConnection) http.HandleFunc("/ws", wsHandler) http.ListenAndServe(addrPort, nil)} |
以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持服务器之家。
原文链接：https://studygolang.com/articles/23126
