Golang：实时消息交互系统

一、项目框架

1.1 Server端：

1.1.1Server：服务器
—OnlineMap：纪录都有哪些用户在线。(key:用户名,value:用户对象)，当给一个用户发消息时，在OnlineMap中查询是否在线并找到用户对象找到对应连接。
—Message channel：server用来广播的channel，如果收到一个消息，先将消息写道该channel中，该channel再将消息转发给特定user的channel或者广播给全部user的channel，每个user就会将自己channel中收到的消息发给客户端。
1.1.2User：在线的用户
user这部门有两个goroutine，读写分离模型。
—一个goroutine负责从user channel中读消息，一旦有消息，就会立即将消息发送给客户端。
—handler go：永世壅闭等待客户端发消息，read。
1.2 Client端：

二、底子server构建

2.1 server.go

<1.创建server结构体(Ip、Port)
<2.提供接口创建、初始化server结构体对象
<3.写一个start启动服务器的server的类方法，listen、accept、go handler
<4.Handler方法，用于收到客户端的业务处理

1. package main
2. import "net"
3. import "fmt"
5. type Server struct {
6.         Ip string
7.         Port int
8. }
10. func (this *Server) Handler(conn net.Conn) {//创建成功与客户端建立好连接的套接字
11.         //...当前连接的业务
12.         fmt.Println("链接建立成功！")
13. }
15. //创建一个server的接口-用于创建server对象-仅仅是对外提供的接口
16. func NewServer(ip string,port int) *Server {
17.         server := &Server{ //把当前创建对象的地址传给返回值
18.                 Ip : ip,
19.                 Port : port,
20.         }
21.         return server
22. }
24. //启动服务器的方法-server的成员方法
25. func (this *Server) Start() {
27.         //socket listen;Listen接口返回listener对象和error
28.         listener,err := net.Listen("tcp",fmt.Sprintf("%s:%d",this.Ip,this.Port))//"127.0.0.1:8888"
29.         if err != nil {
30.                 fmt.Println("Listener accept err:",err)
31.                 return
32.         }
34.         //close listen socket
35.         defer listener.Close()
37.         //
38.         for {
39.                 //accept; 返回一个连接对象conn和err，这个conn可以中有读写等操作--跟客户端建立成功的套接字
40.                 conn,err := listener.Accept()
41.                 if err != nil {
42.                         fmt.Println("Listener accept err:",err)
43.                         continue
44.                 }
45.                 //do handler业务回调
46.                 go this.Handler(conn)//让一个go去执行业务，主go成继续循环接收下一个连接
47.         }
48. }

复制代码

2.2 main.go

1. package main //server.go和main.go都属于main包，不需要再Import
3. func main() {
4.         server := NewServer("127.0.0.1",8888)
5.         server.Start()
6. }

复制代码

二、用户上线及广播功能 

client客户端在server中怎样表现呢？封装一个user类。这个user类中包含一个conn与客户端简历链接，在线的user存到OnlineMap中，每个user对象还会绑定一个channel，用来接收Message channel中的消息，被user的goroutine壅闭监听，收到就会通过conn发送给client客户端。Message channel是接收客户端的消息，Message goroutine壅闭监听channel，一旦有消息，就遍历OnlineMap去通过user channel广播给在线用户。

2.1封装一个User类  

<1.User类中成员变量：name、addr、c(server)、conn(与客户端的连接)
<2.提供创建并初始化User对象的接口，并在初始化时启动goroutine对user channel监听。
<3.监听方法。从user channel中读数据，读出后将数据通过conn写入client。

1. package main
2. import "net"
4. type User struct {
5.         Name string
6.         Addr string
7.         C chan string
8.         Conn net.Conn
9. }
11. //创建一个用户的API
12. func NewUser(conn net.Conn) *User {
13.         userAddr := conn.RemoteAddr().String() //从conn获取addr
14.         user := &User{
15.                 Name : userAddr,
16.                 Addr : userAddr,
17.                 C : make(chan string),
18.                 Conn : conn,
19.         }
21.         //启动监听当前user channel消息的goroutine
22.         go user.ListenMessage()
24.         return user
25. }
27. //监听当前User channel的方法，一旦有消息，就直接发送给对端客户端
28. func (this *User) ListenMessage() {
29.         for {
30.                 //一直从管道中读数据
31.                 msg := <-this.C
32.                 //向客户端写消息
33.                 this.Conn.Write([]byte(msg + "\n"))
34.         }
35. }

复制代码

2.2对Server类的更改

<1.新增成员变量OnlineMap、mapLock(掩护onlinemap)、Message(转发的channel)
<2.server在listen时每accept一个conn就表明一个用户上线，每一个客户端连接都分配一个gorountine，将每一个客户端conn封装成user对象，加入到OnlineMap中，再向Message channel中写数据。
<3.在start服务器时新增gorountine去监听Message有没有收到用户创建连接的消息，收到后就将该消息广播给全部user的user channel。

1. package main
2. import "net"
3. import "fmt"
4. import "sync"
6. type Server struct {
7.         Ip string
8.         Port int
10.         //在线用户的列表
11.         OnlineMap map[string]*User
12.         mapLock sync.RWMutex
14.         //消息广播的channel
15.         Message chan string
16. }
18. //监听message广播消息channel的gorountine，一旦有消息就发送给全部在线的user
19. func (this *Server) ListenMessage() {
20.         for {
21.                 msg := <-this.Message//server中start时仅有一个gorountine专门来监听message channel,而每一个用来处理监听user连接业务的gorountine像这一个message channel中写数据
22.                 //将msg发送给全部在线的User
23.                 this.mapLock.Lock()
24.                 for _,cli := range this.OnlineMap {
25.                         cli.C <- msg //广播给每一个user的channel
26.                 }
27.                 this.mapLock.Unlock()
28.         }
29. }
31. func(this *Server) BroadCast(user *User,msg string){
32.         sendMsg := "["+user.Addr+"]"+user.Name+":"+msg
33.         //给server类的channel写数据
34.         this.Message <- sendMsg//将打包好的消息给massage channel,由它进行广播。server端listen监听客户端，连接上后启用goroutine去处理这个conn，将这个conn封装成user，将这个user给到hash
35. }
37. //user的goroutine
38. func (this *Server) Handler(conn net.Conn) {//创建成功与客户端建立好连接的套接字
39.         //处理上线的用户
40.         // fmt.Println("链接建立成功！")
41.         user := NewUser(conn)
42.         //用户上线，将用户加入到onlineMap中
43.         this.mapLock.Lock()
44.         this.OnlineMap[user.Name]=user
45.         this.mapLock.Unlock()
47.         //广播当前用户上线消息
48.         this.BroadCast(user,"已上线")
50.         //当前的handler不能结束-该goroutine就会死亡，子goroutine就会死亡
51. }
53. //创建一个server的接口-用于创建server对象-仅仅是对外提供的接口
54. func NewServer(ip string,port int) *Server {
55.         server := &Server{ //把当前创建对象的地址传给返回值
56.                 Ip : ip,
57.                 Port : port,
58.                 OnlineMap : make(map[string]*User),
59.                 Message : make(chan string),
60.         }
61.         return server
62. }
64. //启动服务器的方法-server的成员方法
65. func (this *Server) Start() {
67.         //socket listen;Listen接口返回listener对象和error
68.         listener,err := net.Listen("tcp",fmt.Sprintf("%s:%d",this.Ip,this.Port))//"127.0.0.1:8888"
69.         if err != nil {
70.                 fmt.Println("Listener accept err:",err)
71.                 return
72.         }
74.         //close listen socket
75.         defer listener.Close()
77.         //启动监听Message的gorountine
78.         go this.ListenMessage()
80.         //
81.         for {
82.                 //accept; 返回一个连接对象conn和err，这个conn可以中有读写等操作--跟客户端建立成功的套接字
83.                 conn,err := listener.Accept()//此时就有用户上线了
84.                 if err != nil {
85.                         fmt.Println("Listener accept err:",err)
86.                         continue
87.                 }
88.                 //do handler业务回调
89.                 go this.Handler(conn)//让一个go去执行业务，主go成继续循环接收下一个连接
90.         }
91. }

复制代码

三、用户消息广播功能 

在接收每个client的conn的go程中又创建子go程去壅闭从conn中读数据，读不到就以为对方下线。

1. //user的goroutine，从client->onlinemap->message channel的线
2. func (this *Server) Handler(conn net.Conn) {//创建成功与客户端建立好连接的套接字
3.         //处理上线的用户
4.         // fmt.Println("链接建立成功！")
5.         user := NewUser(conn)
6.         //用户上线，将用户加入到onlineMap中
7.         this.mapLock.Lock()
8.         this.OnlineMap[user.Name]=user
9.         this.mapLock.Unlock()
11.         //广播当前用户上线消息
12.         this.BroadCast(user,"已上线")
14.         //接收客户端发送的消息,一直监听，从conn中读；
15.         go func() {
16.                 buf := make([]byte,4096)
17.                 for {
18.                         //Read(b []byte) (n int,err error) 读成功返回n为读到的字节数，读失败err
19.                         n,err := conn.Read(buf)
20.                         if n==0 {
21.                                 this.BroadCast(user,"下线") //读不到消息就是下线了吗？依旧可...
22.                                 return
23.                         }
24.                         if err != nil && err != io.EOF {//不为空且不是读到文件末尾
25.                                 fmt.Println("Conn Read err:",err)
26.                                 return
27.                         }
28.                         //提取用户的消息(去除'\n')
29.                         msg := string(buf[:n-1])//从0到n-1
31.                         //将得到的消息进行广播
32.                         this.BroadCast(user,msg)
33.                 }
34.         }()
36.         //当前的handler不能结束-该goroutine就会死亡，子goroutine就会死亡
37.         select {}//让当前handler阻塞？？？？
38. }

复制代码

四、用户业务封装

<2.在User类中新增Server关联(组合)，通过关联可以对server中的属性和方法操纵。 
<1.将用户上线、用户下线、用户消息处理在User类中封装。

1. //用户上线的业务
2. func (this *User) Online() {
3.         //用户上线，将用户加入到onlineMap中
4.         this.server.mapLock.Lock()
5.         this.server.OnlineMap[this.Name]=this
6.         this.server.mapLock.Unlock()
8.         //广播当前用户上线消息
9.         this.server.BroadCast(this,"已上线")
10. }
12. //用户下线的业务
13. func (this *User) Offline() {
14.         //用户下线，将用户从onlineMap中删除
15.         this.server.mapLock.Lock()
16.         delete(this.server.OnlineMap,this.Name)
17.         this.server.mapLock.Unlock()
19.         //广播当前用户上线消息
20.         this.server.BroadCast(this,"下线")
21. }
23. //用户处理消息的业务,将该用户(客户端的)消息广播
24. func (this *User) DoMessage(msg string) {
25.         this.server.BroadCast(this,msg)
26. }

复制代码

五、在线用户查询 

遍历OnlineMap表通过conn发送给客户端全部在线用户

1. //给当前User对应的客户端发送消息
2. func (this *User) SendMsg(msg string) {
3.         this.Conn.Write([]byte(msg))
4. }
6. //用户处理消息的业务,将该用户(客户端的)消息广播
7. func (this *User) DoMessage(msg string) {
8.         if msg == "who" {
9.                 //查询当前在线用户都有哪些
11.                 this.server.mapLock.Lock()
12.                 for _,user := range this.server.OnlineMap {
13.                         OnlineMsg := "["+user.Addr+"]"+user.Name+":"+"在线...\n"
14.                         this.SendMsg(OnlineMsg)
15.                 }
16.                 this.server.mapLock.Unlock()
17.         } else {
18.                 this.server.BroadCast(this,msg)
19.         }
20. }

复制代码

六、修改用户名 

新用户名不能存在，如果存在提示。不存在就开始在OnlineMap删除旧用户名，添加新用户名。

1. else if len(msg) > 7 && msg[:7]=="rename|"{
2.                 //消息格式：renanme|张三
3.                 newName := strings.Split(msg,"|")[1]
4.                 //判断name是否存在
5.                 _, ok := this.server.OnlineMap[newName]
6.                 if ok {
7.                         this.SendMsg("当前用户名被使用\n")
8.                 }else{
9.                         this.server.mapLock.Lock()
10.                         delete(this.server.OnlineMap,this.Name)
11.                         this.server.OnlineMap[newName]=this
12.                         this.server.mapLock.Unlock()
14.                         this.Name = newName
15.                         this.SendMsg("您已经更新用户名:"+this.Name +"\n")
16.                 }
17.         }

复制代码

七、超时强踢功能

 如果用户长时间不保持活跃，就断开它的连接。如果Read到用户数据就重置定时器，如果超过十秒没有收到用户消息，就将连接断开。

1. //当前的handler不能结束-该goroutine就会死亡，子goroutine就会死亡
2.         for {
3.                 select { //select会阻塞监控管道channel的消息，如果超时会满足case，触发，select不会阻塞
4.                         case <-isLive:
5.                                 //当前用户活跃应重置定时器--不做处理，只为了激活select顺序执行time.After重置定时器
6.                         case <-time.After(time.Second*10): //time.After 返回一个单向通道（<-chan time.Time），该通道在指定的时间间隔后发送当前时间值‌
7.                         //如果超时，将当前的user强制关闭
8.                         user.SendMsg("你被超时踢出了")
9.                         close(user.C)//关闭channel
10.                         conn.Close()//关闭连接
11.                         return//退出当前hanler
12.                 }
13.         }

复制代码

八、用户私聊功能 

<1.目前只实现了将用户的消息进行广播的功能。客户端先通过who查询在线用户，再发送给服务器具有格式的消息指明私聊消息发送给哪位在线用户；
<2.从msg中拿到对方userName，在OnlineMap找到它的user对象，用它的user对象调用sendMsg函数通过conn给其发送消息。
<3.私聊消息的本质是在OnlineMap中保存的多个用户的连接中找到目的用户的长连接conn向它write。从client A的conn读到的消息进行处理后在Map找到它要发送的client B的conn去write。

1. else if len(msg) > 4 && msg[:3] == "to" {
2.                 //消息格式：to|张三|消息内容
3.                 //<1.获取对方的用户名
4.                 remoteName := strings.Split(msg,"|")[1]
5.                 if remoteName == "" {
6.                         this.SendMsg("消息格式不正确，请使用"to|张三|你好啊"格式,\n")
7.                         return
8.                 }
9.                 //<2.根据用户名得到对方的user对象
10.                 remoteUser,ok := this.server.OnlineMap[remoteName]
11.                 if !ok {
12.                         this.SendMsg("该用户名不存在\n")
13.                         return
14.                 }
15.                 //<3.获取消息内容，通过对方的User对象将消息发送过去
16.                 content := strings.Split(msg,"|")[2]
17.                 if content == "" {
18.                         this.SendMsg("无消息内容，请重发\n")
19.                         return
20.                 }
21.                 remoteUser.SendMsg(this.Name + "对您说:"+content) //从msg中拿到对方userName，在OnlineMap找到它的user对象，用它的user对象调用sendMsg函数通过conn给其发送消息。
22.         }

复制代码

九、客户端基本构建 

1. package main
3. import "net"
4. import "fmt"
6. type Client struct {
7.         ServerIp string //服务器Ip
8.         ServerPort int
9.         Name string //客户端名称
10.         conn net.Conn //连接句柄
11. }
13. func NewClient(serverIp string,serverPort int) *Client {
14.         //创建客户端对象
15.         client := &Client{
16.                 ServerIp: serverIp,
17.                 ServerPort: serverPort,
18.         }
20.         //连接server,客户端去连接服务器func Dial(network,address string) (Conn,error)
21.         conn, err := net.Dial("tcp",fmt.Sprintf("%s:%d",serverIp,serverPort))
22.         if err != nil {
23.                 fmt.Println("net.Dial error:",err)
24.                 return nil
25.         }
26.         client.conn = conn
28.         //返回对象
29.         return client
30. }
32. func main() {
33.         client := NewClient("127.0.0.1",8888)
34.         if client == nil{
35.                 fmt.Println(">>>>>连接服务器失败>>>>>")
36.                 return
37.         }
38.         fmt.Println(">>>>>连接服务器成功>>>>>")
39.         //启动客户端的业务
40.         select {}
41. }

复制代码

十、解析命令行 

1. var serverIp string
2. var serverPort int
3. func init() { //将两个形参绑定到flag包中
4.         flag.StringVar(&serverIp,"ip","127.0.0.1","设置服务器IP地址(默认是127.0.0.1)")
5.         flag.IntVar(&serverPort,"port",8888,"设置服务器Port地址(默认是8888)")
6. }
8. func main() {
9.         //命令行解析
10.         flag.Parse()
11. }

复制代码

十一、添加客户端菜单

1. func (client *Client) menu() bool {
2.         var flag int
3.         fmt.Println("************************************")
4.         fmt.Println("1.公聊模式")
5.         fmt.Println("2.私聊模式")
6.         fmt.Println("3.更新用户名")
7.         fmt.Println("0.退出")
8.         fmt.Println("************************************")
11.         fmt.Scanln(&flag)
13.         if flag >= 0 && flag <= 3{
14.                 client.flag = flag
15.                 return true
16.         }else {
17.                 fmt.Println(">>>>请输入合法范围内的数字<<<<")
18.                 return false
19.         }
20. }
22. func (client *Client) Run() {
23.         for client.flag != 0{
24.                 for client.menu() != true {//阻塞式等待用户输入正确的flag或者退出
25.                 }
26.                 //根据不同的模式处理不同的业务
27.                 switch client.flag {
28.                 case 1:
29.                         //公聊模式
30.                         fmt.Println("公聊模式")
31.                         break
32.                 case 2:
33.                         //私聊模式
34.                         fmt.Println("私聊模式")
35.                         break
36.                 case 3:
37.                         //更新用户名
38.                         fmt.Println("更新用户名")
39.                         break
40.                 }
41.         }
42. }

复制代码

十二、公聊、私聊、更新用户名接口

12.1更新用户名接口实现

1. func (client *Client) UpdateName() bool {
2.         fmt.Println(">>>>>请输入用户名<<<<<")
3.         fmt.Scanln(&client.Name)
5.         sendMsg := "rename|" + client.Name +"\n"
6.         _,err := client.conn.Write([]byte(sendMsg)) //从命令行中读消息并发送给服务器
7.         if err != nil {
8.                 fmt.Println("conn.Write err:",err)
9.                 return false
10.         }
11.         return true
12. }

复制代码

12.2公聊模式接口实现

1. func (client *Client) PublicChat() {
2.         //提示用户输入消息
3.         var chatMsg string
4.         fmt.Println(">>>>>请输入聊天内容,exit退出")
5.         fmt.Scanln(&chatMsg)
6.        
7.         for chatMsg != "exit"{
8.                 //发给服务器
9.                 if len(chatMsg) != 0{
10.                         sendMsg := chatMsg + "\n"//协议中有\n
11.                         _,err := client.conn.Write([]byte(sendMsg)) //发送给服务器
12.                         if err != nil {
13.                                 fmt.Println("conn Write err:",err)
14.                                 break
15.                         }
16.                 }
17.                 //只要用户不退出，就一直在公聊模式中
18.                 chatMsg = ""
19.                 fmt.Println(">>>>>请输入聊天内容,exit退出")
20.                 fmt.Scanln(&chatMsg)
21.         }
22. }

复制代码

12.3私聊模式接口实现 

1. func (client *Client) PrivateChat() {
2.         var remoteName string
3.         var chatMsg string
4.         client.SelectUsers()
5.         fmt.Println(">>>>>请输入聊天对象[用户名],exit退出：")
6.         fmt.Scanln(&remoteName)
8.         for remoteName != "exit"{
9.                 fmt.Println(">>>>>请输入消息内容，exit退出")
10.                 fmt.Scanln(&chatMsg)
11.                 for chatMsg != "exit" {
12.                         //消息不为空则发送
13.                         if len(chatMsg) != 0{
14.                                 sendMsg := "to|" + remoteName + "|" + chatMsg + "\n\n"
15.                                 _,err := client.conn.Write([]byte(sendMsg))
16.                                 if err != nil {
17.                                         fmt.Println("conn Write err:",err)
18.                                         break
19.                                 }
20.                         }
21.                         chatMsg = ""
22.                         fmt.Println(">>>>>请输入消息内容，exit退出")
23.                         fmt.Scanln(&chatMsg)
24.                 }
25.                 remoteName = ""
26.                 client.SelectUsers()
27.                 fmt.Println(">>>>>请输入聊天对象[用户名],exit退出：")
28.                 fmt.Scanln(&remoteName)
29.         }
30. }

复制代码

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。