Golang网络编程：即时通讯系统Instance Messaging System

系统基本架构

版本迭代

项目改造

• 无人机是client，我们是server，提供注册登入，场景选择等。
• 信道模拟器是server，我们是client，我们向信道模拟器发送数据，等待信道模拟器计算结果，返回给无人机。

1. 一台无人机上线通知所有无人机（广播）
2. 在线无人机查询
3. 超时踢出
4. 无人机点对点通信（全连通和星形中的星形连接）

共性知识点

1. 使用mutex互斥锁解决用户在线表的并发操作问题

在 Go 语言中，并发操作一个 map 需要采取一定的措施来确保并发安全。因为 map 在并发读写时会引发数据竞争，可能导致程序运行不正确或崩溃。

下面介绍两种常见的在并发环境下安全地操作 map 的方式：

1. 使用互斥锁（Mutex）：
   最常见的方式是使用互斥锁来保护 map 的读写操作。互斥锁能够确保同一时间只有一个 goroutine 可以访问被锁定的代码块。

下面是使用互斥锁来并发安全地操作 map 的示例：

import (
    "sync"
)

var mu sync.Mutex
var m = make(map[string]int)

func writeToMap(key string, value int) {
    mu.Lock()
    defer mu.Unlock()
    m[key] = value
}

func readFromMap(key string) int {
    mu.Lock()
    defer mu.Unlock()
    return m[key]
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

2. 使用并发安全的 sync.Map：
   自 Go 1.9 起，标准库中引入了 sync.Map 类型，它提供了一种并发安全的 map 实现。与普通的 map 不同，sync.Map 的读写操作不需要额外加锁，因为它内部使用了一种高效的并发访问机制。

下面是使用 sync.Map 进行并发安全操作的示例：

import (
    "sync"
)

var m sync.Map

func writeToMap(key string, value int) {
    m.Store(key, value)
}

func readFromMap(key string) (int, bool) {
    return m.Load(key)
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14

2. golang中的chan操作

Go中的chan
在 Go 语言中，chan（通道）是用于协程之间进行通信的一种重要机制。可以通过 chan 进行数据的发送和接收，实现协程之间的同步和数据交换。
创建 chan 可以使用 make() 函数进行初始化，指定通道元素的类型。例如：

go ch := make(chan int)   // 创建一个 int 类型的通道 ch := make(chan string)  
1

创建一个 string 类型的通道 chan 的发送操作使用 <- 运算符，接收操作也使用 <- 运算符。例如：

 ch <- data     // 将 data 发送到 ch 通道
 data := <-ch   // 从 ch通道接收数据，并将其赋值给 data
1
2

发送和接收操作都会阻塞当前协程，直到发送或接收操作完成为止。这使得协程之间可以在通道上进行同步，确保数据按照特定顺序进行发送和接收。

对于无缓冲通道（即没有指定缓冲区大小的通道），发送和接收操作是同步的。发送操作会在有接收方准备好之前阻塞，而接收操作会在有消息可供接收之前阻塞。
使用带缓冲的通道时（通过指定缓冲区大小来初始化通道），当通道中已有缓冲的数据时，即使没有接收操作，发送操作也可以继续进行。只有当缓冲区已满时，发送操作才会阻塞。同样地，当缓冲区为空时，接收操作才会阻塞。

3. net.Conn 中的Read和Write方法

1. 参数和返回值源码定义

type Conn interface {
	// Read reads data from the connection.
	// Read can be made to time out and return an error after a fixed
	// time limit; see SetDeadline and SetReadDeadline.
	Read(b []byte) (n int, err error)

	// Write writes data to the connection.
	// Write can be made to time out and return an error after a fixed
	// time limit; see SetDeadline and SetWriteDeadline.
	Write(b []byte) (n int, err error)

1
2
3
4
5
6
7
8
9
10
11

2. read返回值n 如果是0 一般代表 client关闭了连接，如果成功从连接中读取数据，Read 方法会==返回读取的字节数 n ==和一个 nil 错误。此时，n 可能小于、等于或大于缓冲区的长度，取决于实际读取到的数据量。

在Go语言中，net.Conn接口表示一个通用的网络连接，用于进行数据的读写操作。Read方法是net.Conn接口的一部分，用于从连接中读取数据。下面是对Read方法的解读：

func (c net.Conn) Read(b []byte) (n int, err error)
1

函数签名中，net.Conn表示net.Conn类型的对象，b是用于存储读取数据的缓冲区（一个字节数组），函数的返回值为读取到的字节数和可能出现的错误。具体行为如下：

如果b的长度为0，那么Read会立即返回0和nil，不会阻塞并且不会进行任何读取。

如果b的长度大于0，但小于io.Reader接口内部的缓冲区大小（默认为4096字节），则Read尝试一次性读取全部的数据，不需要多次调用。

如果b的长度超过缓冲区的大小，那么Read会多次调用直到将b填满或者遇到错误为止。这种情况下，Read可能会阻塞等待更多的数据可供读取。

如果连接的对端关闭了连接，则Read会返回0和一个io.EOF错误。

需要注意的是，在网络传输中，读取操作可能会阻塞，并且如果没有数据可读，Read可能会一直阻塞等待。为了避免阻塞，可以使用超时控制、并发协程等方式来处理。另外，读取到的字节数n与缓冲区的长度不一定相等，这是因为网络中的数据可能会分块传输，需要多次读取才能获取完整的数据。因此，需要根据返回值的实际情况进行判断和处理。

4. 在Go语言中，select语句与case语句结合使用，用于处理并发操作。

1. select语句用于同时等待多个通道操作。它的基本语法如下：

select {
case <-ch1:
    // 当ch1通道有值可读时执行的代码
case data := <-ch2:
    // 当ch2通道有值可读时执行的代码，并将值赋给data变量
case ch3 <- value:
    // 将value的值发送到ch3通道中
default:
    // 当没有任何通道操作时执行的代码
}

1
2
3
4
5
6
7
8
9
10
11

2. select语句会阻塞，直到其中一个case语句能够执行。如果多个case同时满足条件，Go会随机选择一个执行。

以下是一些select语句的用法例子：

package main

import (
	"fmt"
	"time"
)

func main() {
	ch1 := make(chan int)
	ch2 := make(chan int)
	ch3 := make(chan int)

	go func() {
		time.Sleep(2 * time.Second)
		ch1 <- 1
	}()

	go func() {
		time.Sleep(3 * time.Second)
		ch2 <- 2
	}()

	go func() {
		time.Sleep(4 * time.Second)
		ch3 <- 3
	}()

	select {
	case <-ch1:
		fmt.Println("Received from ch1")
	case <-ch2:
		fmt.Println("Received from ch2")
	case <-ch3:
		fmt.Println("Received from ch3")
	case <-time.After(5 * time.Second):
		fmt.Println("Timeout")
	}
}

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

在上面的例子中，使用select同时等待三个通道操作。根据每个通道操作的延迟时间，可能会收到不同的结果。

请注意，在select语句中可以包含default分支，用于处理当没有任何通道操作可用时的情况。如果所有的通道都未准备好，且没有default分支，那么select语句将会阻塞，直到至少有一个通道准备好。使用select语句的default分支可以避免程序阻塞。
3. case <-time.After(time.Second * 10) 是 select 语句中的一个 case 分支，它使用了 time.After() 函数来创建一个定时器。

在这个例子中，time.After(time.Second * 10) 返回一个通道（channel），并在指定的时间间隔后向该通道发送一个值。通过将这个通道放在 select 语句中的 case 分支中，我们可以等待指定的时间后执行相应的代码。

以下是一个示例：

package main

import (
	"fmt"
	"time"
)

func main() {
	select {
	case <-time.After(time.Second * 10):
		fmt.Println("Timeout")
	}
}
1
2
3
4
5
6
7
8
9
10
11
12
13

在上面的示例中，我们使用 select 语句和 time.After() 创建了一个 10 秒的定时器。当 10 秒钟过去后，将会从 time.After() 返回的通道中接收到一个值，然后执行相应的代码，在这里是打印 “Timeout”。

这种用法常用于超时控制，例如在并发操作中设置一个操作的最大执行时间。如果在指定的时间内没有完成操作，就可以执行相应的超时逻辑。

1. 构建基础server

主要思路：

1. 使用socket listen监听，net.Listen(“tcp”,“ip+端口”)。这里的ip和端口使用一个server类（对应的指针）传进来。server这个类中最基本的肯定要有ip和端口。
2. 监听开启后会返回一个listener，listener有accept方法。 conn,err := listener.Accept(). 连接成功后，err==nill，conn有值。具体来说：listener.Accept() 是一个阻塞函数，用于等待客户端连接并接受连接。它会一直阻塞当前 goroutine 直到有新的客户端连接到服务器，然后返回一个表示客户端连接的 net.Conn 类型对象。
3. 由于 listener.Accept() 方法是阻塞调用，因此通常需要将其放在一个死循环中，并在每次接收到客户端连接后使用 goroutine 并发处理该连接。这样，服务器就可以同时处理多个客户端连接（死循环和并发协程结合在一起），而不必等待之前的连接处理完成。
   当客户端连接断开时，net.Conn 对象会自动关闭，因此我们不需要手动关闭连接。不过，为了保证资源能够被合理释放，我们可以使用 defer 语句在处理完连接后关闭连接
4. accept 之后，连接成功，就可以写handler处理server的业务，自己写的这个handler需要获取上面的连接conn
5. 面向对象的编程思想，start（）和handler都属于结构体server 的成员方法，可以在start（）中直接使用this调用handler（conn）。另外，新建一个server，自己将其封装为一个函数，方便main直接调用，得到一个server。
   

2. 用户上线，并广播通知其他已上线用户

• 用户client的模拟可以通过nc命令 ： 连接到远程主机的特定端口：nc <目标IP> <端口号> 如 nc 127.0.0.1 8888连接到了server。此时作为客户端，自己的ip就是主机ip，如果没有明确指定源端口，操作系统会自动分配一个可用的本地端口。

• 流程：
  简单说：server监听client，一旦有用户上线，开一个协程执行handler方法。handler就是先把上线的user存到map表中（或者存到数据库中），然后再给server自己的通道写一条当前这个用户上线的消息。由于之前开启了一个协程，定期的检查server的channel有没有东西，有东西就发给map表中每个在线的用户，当然，由于data := <-ch // 从 ch 通道接收数据 这个操作是阻塞的操作，因此通道为空的话会一直阻塞在这一行代码中。

  1. 开启服务器，listener, err := net.Listen("tcp", fmt.Sprintf("%s:%d", this.Ip, this.Port)) 监听自己的ip和端口有没有人连接
  2. 开启子协程，死循环不断监听 server自己的chan有没有消息写入，有的话向每个在线用户进行广播处理go this.ListenMessager()
  3. 第一步的listener不为空，说明有用户连上来，accept后得到conn连接对象conn, err := listener.Accept()
  4. server不停的accept（阻塞方法，配合循环和异步协程处理），只要有连接来，就会把和这个客户端口的连接对象conn传给handler进行业务处理，如向自己的chan中写入要广播的消息。
  5. 后面conn用于newUser对象，conn成为user对象的重要成员变量，当然，client和客户端完成交互后，要手动conn.Close()关闭连接，释放资源。
  6. conn用处是user和client之间通信：监听当前User channel 的时候,一旦有消息，就直接发送给对端客户端 this.conn.Write([]byte(msg + "\n"))，

• 外部流程：

  1. 客户端：nc 127.0.0.1：8888。 连接服务器，自己的ip比如是：127.0.0.1：45726
  2. server会生成与127.0.0.1：45726 的conn，这个conn用于new User对象。返回给127.0.0.1：45726 这个地址（即为client）对应的消息this.conn.Write([]byte(msg + "\n"))，nc的命令行就能收到对应的消息了。
     

2.1 先建立user结构体

• 需要注意的是newUser的方法需要传入net.Conn的类型参数，来源为服务器与当前用户accept之后的conn，创建一个user最重要的就是这个user与server之间的conn。user有了这个conn，可以直接写回客户端消息
• user其实除了将自己new出来，就做一件事情：监听自己的chan，有消息了就通过conn发给client。实际上user就是client，只不过抽象为一个对象比较好操作。

2.2 server中完善

1. server现在要做的事情主要两个：client来了之后先写入自己的chan相关消息（handler中调用），监听chan中的消息，发送给user 的chan
2. 之前的server只需要最基本的ip和port，传给net.Listen(“tcp”,“ip+port”)，满足socket监听和accept，得到conn。现在server还需要维护一个用户表、维护一个chan通道（用于广播给各个用户消息）

2.3 广播功能

1. 对于server来说，accept之后拿到conn对象，调用handler（conn），这时候handler之中需要写业务：通知其他在线用户新用户上线
2. handler：第一步，先将新用户注册到server的用户表中。写一个新的方法，广播到其他用户
3. 新方法实际上就是向server维护的通道中写入业务消息。this.Message <- 消息 Message是chan，是server的成员属性
4. 消息写到通道中去，需要开启子线程监听这个通道中，一有消息就广播给其他用户，思路也很简单，就是监听通道（阻塞的）：
   msg := -< range this.Message ,之后遍历map（使用range），拿到map中的value即为user对象
   

3. 用户消息广播

3.1 主要思路

之前的代码版本已经实现：只要serverChannel中一有消息，有一个GoRoutine 会马上将该消息broadCast给每个用户。现在需要将一个用户的消息进行广播，那只需要将该消息放到serverChannel中即可。而且根据连接的断开，net.Conn.Read() 会返回一个读到的字节数n，如果n==0，说明一个用户连接已经断开了，可以将该用户从列表中移除。
思路简单，有一个小问题就是会把消息发给每个用户，包括当前用户，这个应该也可以传入当前用户的相关信息进行改进。
1
2

4. 在线用户查询、修改用户名

4.1 主要思路

1. 之前的用户消息广播，就是接收到用户的消息（在server的handler中使用conn.Read(byteArrBuff)），将client的数据读到byteArrBuff中后，也就相当于得到了client发的消息的比特流。之前是直接将消息广播到所有用户，直接写入server的管道就好了，有goroutine读取chan的数据发送到每个用户的channel上
2. 现在不能将收到的所有消息直接全部发到serverChannel中，因为要细分功能。比如现在的在线用户查询、修改用户名。
3. 在线用户查询思路是，规定用户输入“who”，那就server这时候识别到是“who”，就查询onlineMap，将所有在线用户消息一条条地，直接通过conn.write()写回给client
4. 修改用户名思路是：如果收到client发送消息格式: rename|张三， 那就在map中修改名字即可。当然，需要判断用户的名字是否已经被占用等。

4.2 代码实现

func (this *User) DoMessage(msg string) {
	if msg == "who" {
		//查询当前在线用户都有哪些

		this.server.mapLock.Lock()
		for _, user := range this.server.OnlineMap {
			onlineMsg := "[" + user.Addr + "]" + user.Name + ":" + "在线...\n"
			this.SendMsg(onlineMsg)
		}
		this.server.mapLock.Unlock()

	} else if len(msg) > 7 && msg[:7] == "rename|" {
		//消息格式: rename|张三
		newName := strings.Split(msg, "|")[1]

		//判断name是否存在
		_, ok := this.server.OnlineMap[newName]
		if ok {
			this.SendMsg("当前用户名被使用\n")
		} else {
			this.server.mapLock.Lock()
			delete(this.server.OnlineMap, this.Name)
			this.server.OnlineMap[newName] = this
			this.server.mapLock.Unlock()

			this.Name = newName
			this.SendMsg("您已经更新用户名:" + this.Name + "\n")
		}

	} else {
		this.server.BroadCast(this, msg)
	}
}
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

5. 用户超时踢出

• 如果isLive这个chan中写入值了，有值可读，那就执行完毕，进行下一次for循环。
• 如果到了10s，isLive 还没有值可以读，time.After(time.Second * 10)会创建一个chan，并在一进入select开始计时10s，10s后向chan中发数据。这样一来就执行了它case中的内容。
• 总之， select 语句中的每个分支都只有在相应的通道操作可执行时才会被执行。而 time.After() 函数只是一个将指定时间间隔转换为可以等待的通道的辅助方法，用于在需要等待一段时间后才能执行某些操作的场景。

	//当前handler阻塞
	for {
		select {
		// 当isLive通道有值可读时执行的代码
		case <-isLive:
			//当前用户是活跃的，应该重置定时器
			//不做任何事情，为了激活select，更新下面的定时器

		case <-time.After(time.Second * 10):
			//已经超时
			//将当前的User强制的关闭

			user.SendMsg("你被踢了")

			//销毁用的资源
			close(user.C)

			//关闭连接
			conn.Close()

			//退出当前Handler
			return //runtime.Goexit()
		}
	}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

6. 用户单聊