并发聊天服务器编写

并发聊天服务器

package main

import (
	"fmt"
	"net"
	"strings"
	"time"
)

// 结构体
type Client struct {
	C    chan string //用户发送数据的管道
	Name string      //用户名
	Addr string      //网络地址
}

// 保存在线用户 cliAddr -->client
var onlineMap map[string]Client

// 创建一个通信的管道
var message = make(chan string)

func MakeMsg(cli Client, msg string) (buf string) {
	buf = "[" + cli.Addr + "]" + cli.Name + ": " + msg
	return
}

// WriteMsgToClient 专门给当前客户端发送信息
func WriteMsgToClient(cli Client, conn net.Conn) {
	for msg := range cli.C { //给当前客户端发送信息
		_, err := conn.Write([]byte(msg + "\n"))
		if err != nil {
			fmt.Println("conn write err =", err)
			return
		}
	}
}

// HandleConn //处理用户链接
func HandleConn(conn net.Conn) {
	defer conn.Close()
	//获取客户端的网络地址
	cliAddr := conn.RemoteAddr().String()

	//创建一个结构体 创建一个管道 默认用户名和网络地址一样
	cli := Client{make(chan string), cliAddr, cliAddr}
	//把结构体添加到map
	onlineMap[cliAddr] = cli

	//新开一个协程，专门给当前客户端发送信息
	go WriteMsgToClient(cli, conn)

	//广播某个人在线
	//message <- "[" + cli.Addr + "]" + cli.Name + ": login"
	message <- MakeMsg(cli, "login")
	//提示，我是谁
	cli.C <- MakeMsg(cli, "I am hello")

	isQuit := make(chan bool) //对方是否主动退出
	isData := make(chan bool) //对方是否有数据发送
	//新建一个协程，接收用户发送过来的数据
	go func() {
		buf := make([]byte, 2048)
		for {
			readSize, err := conn.Read(buf)
			//if err != nil {
			//	if err == io.EOF {
			//		fmt.Println("写入完毕")
			//	}
			//	return
			//}

			if readSize == 0 { //对象断开，或者出问题
				isQuit <- true
				fmt.Println("conn read err = ", err)
				return
			}

			msg := string(buf[:readSize-1])

			if len(msg) == 3 && msg == "who" { //查询在线用户
				//遍历map,get当前用户发送所有成员
				conn.Write([]byte("user list:\n"))
				for _, tmp := range onlineMap {
					msg = tmp.Addr + ":" + tmp.Name + "\n"
					conn.Write([]byte(msg))
				}
			} else if len(msg) >= 8 && msg[:6] == "rename" {//修改用户名
				name := strings.Split(msg, "|")[1]
				cli.Name = name
				onlineMap[cliAddr] = cli
				conn.Write([]byte("rename ok\n"))
			} else {
				//转发此内容
				fmt.Println("成朋")
				message <- MakeMsg(cli, msg)
			}

			isData <- true //代表有数据
		}
	}()

	for true {
		//通过select检测channel的流动
		select {
		case <-isQuit:
			delete(onlineMap, cliAddr)           //当前用户从map移除
			message <- MakeMsg(cli, "login out") //广播谁下线了
			return
		case <-isData:
		case <-time.After(30 * time.Second):
			delete(onlineMap, cliAddr)
			message <- MakeMsg(cli, "time out leave out") //超时了
			return
		}
	}

}

// Manager 新开一个协程 转发消息,只要有消息来了，遍历map，给map每个成员都发送消息
func Manager() {
	//给map分配空间
	onlineMap = make(map[string]Client)
	for true {
		meg := <-message //没有消息前，这里会阻塞
		//遍历map，给map每个成员都发送消息
		for _, client := range onlineMap {
			client.C <- meg //把消息给这个管道
		}
	}
}
func main() {

	//监听
	listener, err := net.Listen("tcp", ":8000")
	if err != nil {
		fmt.Println("listener err = ", err)
		return
	}

	defer listener.Close()

	//新开一个协程 转发消息,只要有消息来了，遍历map，给map每个成员都发送消息
	go Manager()
	//主协程。循环阻塞等待用户连接
	for true {
		conn, err := listener.Accept() //建立连接
		if err != nil {
			fmt.Println("conn err = ", err)
			continue
		}

		go HandleConn(conn) //处理用户链接
	}
}

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

调试工具netcat