Golang使用gin框架實現(xiàn)一個完整的聊天室功能
用到的技術(shù)
websocket、gin、mysql、redis、協(xié)程、通道
實現(xiàn)思路
說到聊天室可以有多種方法實現(xiàn),例如:使用單純的MySQL也可以實現(xiàn),但是為什么要選擇使用websocket去實現(xiàn)呢?有什么優(yōu)勢呢?
websocket是基于TCP/IP,獨立的HTTP協(xié)議的雙向通信協(xié)議,這就使實時的消息通知成為可能, 同時又符合Go高效處理高并發(fā)的語言特點,結(jié)合聊天室又是高并發(fā)的,所以采取的室websocket進行消息的轉(zhuǎn)接,MySQL持久化聊天消息,redis用于做一些判斷。
首先用戶在進入App時,客戶端和服務(wù)端建立一個websocket連接,并開啟一個通道。
當服務(wù)端收到客戶端的消息后,將消息寫入通道里,服務(wù)端監(jiān)聽通道的消息,并將消息取出,使用接收人的websocket連接將消息廣播到接收人那里。
實現(xiàn)代碼
下面開始實現(xiàn):
創(chuàng)建模型,用于關(guān)系的確立及數(shù)據(jù)的傳輸
//數(shù)據(jù)庫存儲消息結(jié)構(gòu)體,用于持久化歷史記錄
type ChatMessage struct {
gorm.Model
Direction string //這條消息是從誰發(fā)給誰的
SendID int //發(fā)送者id
RecipientID int //接受者id
GroupID string //群id,該消息要發(fā)到哪個群里面去
Content string //內(nèi)容
Read bool //是否讀了這條消息
}
//群聊結(jié)構(gòu)體
type Group struct {
ID string ` gorm:"primaryKey"` //群id
CreatedAt time.Time
UpdatedAt time.Time
DeletedAt gorm.DeletedAt `gorm:"index"`
GroupName string `json:"group_name"` //群名
GroupContent string `json:"group_content"` //群簽名
GroupIcon string `json:"group_icon"` //群頭像
GroupNum int //群人數(shù)
GroupOwnerId int //群主id
Users []User `gorm:"many2many:users_groups;"` //群成員
}
type UsersGroup struct {
GroupId string `json:"group_id"`
UserId int `json:"user_id"`
}
// 用于處理請求后返回一些數(shù)據(jù)
type ReplyMsg struct {
From string `json:"from"`
Code int `json:"code"`
Content string `json:"content"`
}
// 發(fā)送消息的類型
type SendMsg struct {
Type int `json:"type"`
RecipientID int `json:"recipient_id"` //接受者id
Content string `json:"content"`
}
// 用戶類
type Client struct {
ID string //消息的去向
RecipientID int //接受者id
SendID int //發(fā)送人的id
GroupID string //群聊id
Socket *websocket.Conn //websocket連接對象
Send chan []byte //發(fā)送消息用的管道
}
// 廣播類,包括廣播內(nèi)容和源用戶
type Broadcast struct {
Client *Client
Message []byte
Type int
}
// 用戶管理,用于管理用戶的連接及斷開連接
type ClientManager struct {
Clients map[string]*Client
Broadcast chan *Broadcast
Reply chan *Client
Register chan *Client
Unregister chan *Client
}
//創(chuàng)建一個用戶管理對象
var Manager = ClientManager{
Clients: make(map[string]*Client), // 參與連接的用戶,出于性能的考慮,需要設(shè)置最大連接數(shù)
Broadcast: make(chan *Broadcast),
Register: make(chan *Client), //新建立的連接訪放入這里面
Reply: make(chan *Client),
Unregister: make(chan *Client), //新斷開的連接放入這里面
}創(chuàng)建連接
func WsHandle(c *gin.Context) {
myid := c.Query("myid")
userid, err := strconv.Atoi(myid)
if err != nil {
zap.L().Error("轉(zhuǎn)換失敗", zap.Error(err))
ResponseError(c, CodeParamError)
}
//將http協(xié)議升級為ws協(xié)議
conn, err := (&websocket.Upgrader{CheckOrigin: func(r *http.Request) bool {
return true
}}).Upgrade(c.Writer, c.Request, nil)
if err != nil {
http.NotFound(c.Writer, c.Request)
return
}
//創(chuàng)建一個用戶客戶端實例,用于記錄該用戶的連接信息
client := new(model.Client)
client = &model.Client{
ID: myid + "->",
SendID: userid,
Socket: conn,
Send: make(chan []byte),
}
//使用管道將實例注冊到用戶管理上
model.Manager.Register <- client
//開啟兩個協(xié)程用于讀寫消息
go Read(client)
go Write(client)
}
//用于讀管道中的數(shù)據(jù)
func Read(c *model.Client) {
//結(jié)束把通道關(guān)閉
defer func() {
model.Manager.Unregister <- c
//關(guān)閉連接
_ = c.Socket.Close()
}()
for {
//先測試一下連接能不能連上
c.Socket.PongHandler()
sendMsg := new(model.SendMsg)
err := c.Socket.ReadJSON(sendMsg)
c.RecipientID = sendMsg.RecipientID
if err != nil {
zap.L().Error("數(shù)據(jù)格式不正確", zap.Error(err))
model.Manager.Unregister <- c
_ = c.Socket.Close()
return
}
//根據(jù)要發(fā)送的消息類型去判斷怎么處理
//消息類型的后端調(diào)度
switch sendMsg.Type {
case 1: //私信
SingleChat(c, sendMsg)
case 2: //獲取未讀消息
UnreadMessages(c)
case 3: //拉取歷史消息記錄
HistoryMsg(c, sendMsg)
case 4: //群聊消息廣播
GroupChat(c, sendMsg)
}
}
}
//用于將數(shù)據(jù)寫進管道中
func Write(c *model.Client) {
defer func() {
_ = c.Socket.Close()
}()
for {
select {
//讀取管道里面的信息
case message, ok := <-c.Send:
//連接不到就返回消息
if !ok {
_ = c.Socket.WriteMessage(websocket.CloseMessage, []byte{})
return
}
fmt.Println(c.ID+"接收消息:", string(message))
replyMsg := model.ReplyMsg{
Code: int(CodeConnectionSuccess),
Content: fmt.Sprintf("%s", string(message)),
}
msg, _ := json.Marshal(replyMsg)
//將接收的消息發(fā)送到對應(yīng)的websocket連接里
rwLocker.Lock()
_ = c.Socket.WriteMessage(websocket.TextMessage, msg)
rwLocker.Unlock()
}
}
}后端調(diào)度
//聊天的后端調(diào)度邏輯
//單聊
func SingleChat(c *model.Client, sendMsg *model.SendMsg) {
//獲取當前用戶發(fā)出到固定用戶的消息
r1, _ := redis.REDIS.Get(context.Background(), c.ID).Result()
//從redis中取出固定用戶發(fā)給當前用戶的消息
id := CreateId(strconv.Itoa(c.RecipientID), strconv.Itoa(c.SendID))
r2, _ := redis.REDIS.Get(context.Background(), id).Result()
//根據(jù)redis的結(jié)果去做未關(guān)注聊天次數(shù)限制
if r2 >= "3" && r1 == "" {
ResponseWebSocket(c.Socket, CodeLimiteTimes, "未相互關(guān)注,限制聊天次數(shù)")
return
} else {
//將消息寫入redis
redis.REDIS.Incr(context.Background(), c.ID)
//設(shè)置消息的過期時間
_, _ = redis.REDIS.Expire(context.Background(), c.ID, time.Hour*24*30*3).Result()
}
fmt.Println(c.ID+"發(fā)送消息:", sendMsg.Content)
//將消息廣播出去
model.Manager.Broadcast <- &model.Broadcast{
Client: c,
Message: []byte(sendMsg.Content),
}
}
//查看未讀消息
func UnreadMessages(c *model.Client) {
//獲取數(shù)據(jù)庫中的未讀消息
msgs, err := mysql.GetMessageUnread(c.SendID)
if err != nil {
ResponseWebSocket(c.Socket, CodeServerBusy, "服務(wù)繁忙")
}
for i, msg := range msgs {
replyMsg := model.ReplyMsg{
From: msg.Direction,
Content: msg.Content,
}
message, _ := json.Marshal(replyMsg)
_ = c.Socket.WriteMessage(websocket.TextMessage, message)
//發(fā)送完后將消息設(shè)為已讀
msgs[i].Read = true
err := mysql.UpdateMessage(&msgs[i])
if err != nil {
ResponseWebSocket(c.Socket, CodeServerBusy, "服務(wù)繁忙")
}
}
}
//拉取歷史消息記錄
func HistoryMsg(c *model.Client, sendMsg *model.SendMsg) {
//拿到傳過來的時間
timeT := TimeStringToGoTime(sendMsg.Content)
//查找聊天記錄
//做一個分頁處理,一次查詢十條數(shù)據(jù),根據(jù)時間去限制次數(shù)
//別人發(fā)給當前用戶的
direction := CreateId(strconv.Itoa(c.RecipientID), strconv.Itoa(c.SendID))
//當前用戶發(fā)出的
id := CreateId(strconv.Itoa(c.SendID), strconv.Itoa(c.RecipientID))
msgs, err := mysql.GetHistoryMsg(direction, id, timeT, 10)
if err != nil {
ResponseWebSocket(c.Socket, CodeServerBusy, "服務(wù)繁忙")
}
//把消息寫給用戶
for _, msg := range *msgs {
replyMsg := model.ReplyMsg{
From: msg.Direction,
Content: msg.Content,
}
message, _ := json.Marshal(replyMsg)
_ = c.Socket.WriteMessage(websocket.TextMessage, message)
//發(fā)送完后將消息設(shè)為已讀
if err != nil {
ResponseWebSocket(c.Socket, CodeServerBusy, "服務(wù)繁忙")
}
}
}
//群聊消息廣播
func GroupChat(c *model.Client, sendMsg *model.SendMsg) {
//根據(jù)消息類型判斷是否為群聊消息
//先去數(shù)據(jù)庫查詢該群下的所有用戶
users, err := mysql.GetAllGroupUser(strconv.Itoa(sendMsg.RecipientID))
if err != nil {
ResponseWebSocket(c.Socket, CodeServerBusy, "服務(wù)繁忙")
}
//向群里面的用戶廣播消息
for _, user := range users {
//獲取群里每個用戶的連接
if int(user.ID) == c.SendID {
continue
}
c.ID = strconv.Itoa(c.SendID) + "->"
c.GroupID = strconv.Itoa(sendMsg.RecipientID)
c.RecipientID = int(user.ID)
model.Manager.Broadcast <- &model.Broadcast{
Client: c,
Message: []byte(sendMsg.Content),
}
}
}轉(zhuǎn)發(fā)消息
//用于在啟動時進行監(jiān)聽
func Start(manager *model.ClientManager) {
for {
fmt.Println("<-----監(jiān)聽通信管道----->")
select {
//監(jiān)測model.Manager.Register這個的變化,有新的東西加入管道時會被監(jiān)聽到,從而建立連接
case conn := <-model.Manager.Register:
fmt.Println("建立新連接:", conn.ID)
//將新建立的連接加入到用戶管理的map中,用于記錄連接對象,以連接人的id為鍵,以連接對象為值
model.Manager.Clients[conn.ID] = conn
//返回成功信息
controller.ResponseWebSocket(conn.Socket, controller.CodeConnectionSuccess, "已連接至服務(wù)器")
//斷開連接,監(jiān)測到變化,有用戶斷開連接
case conn := <-model.Manager.Unregister:
fmt.Println("連接失敗:", conn.ID)
if _, ok := model.Manager.Clients[conn.ID]; ok {
controller.ResponseWebSocket(conn.Socket, controller.CodeConnectionBreak, "連接已斷開")
}
//關(guān)閉當前用戶使用的管道
//close(conn.Send)
//刪除用戶管理中的已連接的用戶
delete(model.Manager.Clients, conn.ID)
case broadcast := <-model.Manager.Broadcast: //廣播消息
message := broadcast.Message
recipientID := broadcast.Client.RecipientID
//給一個變量用于確定狀態(tài)
flag := false
contentid := createId(strconv.Itoa(broadcast.Client.SendID), strconv.Itoa(recipientID))
rID := strconv.Itoa(recipientID) + "->"
//遍歷客戶端連接map,查找該用戶有沒有在線,判斷的是對方的連接例如:1要向2發(fā)消息,我現(xiàn)在是用戶1,那么我需要判斷2->1是否存在在用戶管理中
for id, conn := range model.Manager.Clients {
//如果找不到就說明用戶不在線,與接收人的id比較
if id != rID {
continue
}
//走到這一步,就說明用戶在線,就把消息放入管道里面
select {
case conn.Send <- message:
flag = true
default: //否則就把該連接從用戶管理中刪除
close(conn.Send)
delete(model.Manager.Clients, conn.ID)
}
}
//判斷完之后就把將消息發(fā)給用戶
if flag {
fmt.Println("用戶在線應(yīng)答")
controller.ResponseWebSocket(model.Manager.Clients[rID].Socket, controller.CodeConnectionSuccess, string(message))
//把消息插到數(shù)據(jù)庫中
msg := model.ChatMessage{
Direction: contentid,
SendID: broadcast.Client.SendID,
RecipientID: recipientID,
GroupID: broadcast.Client.GroupID,
Content: string(message),
Read: true,
}
err := mysql.DB.Create(&msg).Error
if err != nil {
zap.L().Error("在線發(fā)送消息出現(xiàn)了錯誤", zap.Error(err))
}
} else { //如果不在線
controller.ResponseWebSocket(broadcast.Client.Socket, controller.CodeConnectionSuccess, "對方不在線")
//把消息插到數(shù)據(jù)庫中
msg := model.ChatMessage{
Direction: contentid,
SendID: broadcast.Client.SendID,
RecipientID: recipientID,
GroupID: broadcast.Client.GroupID,
Content: string(message),
Read: false,
}
err := mysql.DB.Create(&msg).Error
if err != nil {
zap.L().Error("不在線發(fā)送消息出現(xiàn)了錯誤", zap.Error(err))
}
}
}
}
}
func createId(uid, toUid string) string {
return uid + "->" + toUid
}到此這篇關(guān)于Golang使用gin框架實現(xiàn)一個完整的聊天室功能的文章就介紹到這了,更多相關(guān)Golang實現(xiàn)聊天室內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家!
相關(guān)文章
利用golang實現(xiàn)封裝trycatch異常處理實例代碼
Go語言追求簡潔優(yōu)雅,所以go語言不支持傳統(tǒng)的 try…catch…finally 這種異常,最近發(fā)現(xiàn)了不錯的trycatch包,下面這篇文章主要跟大家分享了關(guān)于利用golang實現(xiàn)封裝trycatch異常處理的實例代碼,需要的朋友可以參考下。2017-07-07
gin通過go build -tags實現(xiàn)json包切換及庫分析
這篇文章主要為大家介紹了gin通過go build -tags實現(xiàn)json包切換及庫分析,有需要的朋友可以借鑒參考下,希望能夠有所幫助,祝大家多多進步,早日升職加薪2023-09-09

