
#### 1. 微信的消息存储机制

微信的消息存储机制是基于写扩散的，每个用户都有自己的收信箱。这意味着每条消息都会被复制到每个相关用户的收信箱中，而不是在发送者和接收者之间直接传递。这种机制使得消息的传递更加可靠，但也增加了存储和同步的复杂性。

#### 2. 消息的有序性

微信的消息在会话内是有序的，即同一个会话中的消息按照发送顺序排列。然而，全局消息是无序的，因为用户的收信箱中包含了来自多个会话的混合消息。这种设计使得消息同步变得复杂，因为无法简单地通过时间戳来同步消息。

#### 3. 用户唯一序列号（seq）

为了解决消息同步的问题，微信为每个用户分配了一个唯一的发号器，用于生成递增的序列号（seq）。当用户收到一条消息时，会请求发号器生成一个新的递增的seq，并将其作为用户信箱的唯一递增seq。服务端会保存用户已经同步到的seq位置，下次启动时只会拉取这之后的数据。

#### 4. 消息同步机制

微信的消息同步机制依赖于用户唯一seq。具体步骤如下：

1. **用户上线**：用户上线时，客户端会请求服务端获取当前用户的最新seq。
2. **消息同步**：客户端根据获取到的seq，从服务端拉取seq之后的所有消息。
3. **更新同步点位**：当客户端成功同步消息后，会更新服务端中用户的同步点位，即最新的seq。

#### 5. 消息保留策略

微信只保留7天的未读数据。如果用户在7天内未读消息，这些消息会被清除，无法再进行同步。这种策略确保了消息存储的效率，但也带来了一些限制，例如用户无法同步超过7天的消息。

#### 6. 实际测试结果

通过实际测试，我们可以验证微信的消息同步机制：

1. **新设备登录**：新设备登录时，客户端没有拉取到同步消息，说明服务端维护的用户同步seq已是最大，没有消息需要同步。
2. **设备A离线，设备B登录**：设备A离线时发送消息，设备B登录后成功收到消息，重新登录A设备后没有同步到消息，说明B在同步到消息后更新了用户同步点位。
3. **设备A在线，设备B登录**：设备A在线时发送消息，设备B登录后同步不到消息，说明A在线发送消息后，用户同步到最新消息，服务端更新了用户同步点位。
4. **两台设备在线**：两台设备都在线时，都能收到消息，说明在线直接推送消息。



微信的消息同步机制通过用户唯一seq实现了高效的消息同步。尽管这种机制在某些情况下可能会导致消息无法同步（例如超过7天未读消息被清除），但它确保了消息的可靠传递和存储效率。

微信消息同步机制分析

### 使用 Go 语言和 KCP 进行高效通信

在网络通信中，传输速度和可靠性是两个关键因素。传统的 TCP 协议在某些情况下可能会因为拥塞控制和重传机制而导致延迟增加。为了解决这个问题，KCP 协议应运而生。KCP 是一个快速可靠的传输协议，旨在提高 TCP 的性能。本文将通过一个简单的示例，展示如何使用 Go 语言和 KCP 进行高效的客户端-服务器通信。

#### 1. 安装 KCP 库

安装 `kcp-go` 库。你可以通过以下命令来安装：

```bash
go get -u github.com/xtaci/kcp-go
```

#### 2. 服务器端代码
我们编写服务器端的代码。服务器将监听 `12345` 端口，实现一个回声服务器

```go
package main

import (
	"fmt"
	"log"
	"net"

	"github.com/xtaci/kcp-go"
)

func handleClient(conn net.Conn) {
	defer conn.Close()

	buf := make([]byte, 4096)
	for {
		n, err := conn.Read(buf)
		if err != nil {
			log.Println("Read error:", err)
			return
		}

		fmt.Printf("Received: %s\n", string(buf[:n]))

		// Echo back the message
		_, err = conn.Write(buf[:n])
		if err != nil {
			log.Println("Write error:", err)
			return
		}
	}
}

func main() {
	listener, err := kcp.ListenWithOptions(":12345", nil, 10, 3)
	if err != nil {
		log.Fatal("Listen error:", err)
	}
	defer listener.Close()

	fmt.Println("KCP server is listening on :12345")

	for {
		conn, err := listener.Accept()
		if err != nil {
			log.Println("Accept error:", err)
			continue
		}

		go handleClient(conn)
	}
}
```

#### 3. 客户端代码

接下来，我们编写客户端的代码。

```go
/*
 * @Author: yowayimono
 * @Date: 2024-11-30 14:24:03
 * @LastEditors: yowayimono
 * @LastEditTime: 2024-11-30 14:26:43
 * @Description: nothing
 */
package main

import (
	"fmt"
	"log"

	"github.com/xtaci/kcp-go"
)

func main() {
	conn, err := kcp.DialWithOptions("127.0.0.1:12345", nil, 10, 3)
	if err != nil {
		log.Fatal("Dial error:", err)
	}
	defer conn.Close()

	for {
		var message string
		_, err := fmt.Scan(&message)

		if message == "exit" {
			break
		}

		if err != nil {
			log.Fatal("Scan error:", err)
		}

		_, err = conn.Write([]byte(message))
		if err != nil {
			log.Fatal("Write error:", err)
		}

		buf := make([]byte, 4096)
		n, err := conn.Read(buf)
		if err != nil {
			log.Fatal("Read error:", err)
		}

		fmt.Printf("Received from server: %s\n", string(buf[:n]))

	}

}

```

#### 4. 运行示例



KCP使用示例

```go
package health

import (
	"fmt"
	"html/template"
	"net/http"
	_ "net/http/pprof"
	"runtime"
	"sync"
	"time"

	"github.com/gorilla/websocket"
)

type MemUsage struct {
	Alloc      uint64
	TotalAlloc uint64
	Sys        uint64
	NumGC      uint32
	Timestamp  time.Time
}

var memUsageHistory []MemUsage
var mu sync.Mutex
var upgrader = websocket.Upgrader{
	CheckOrigin: func(r *http.Request) bool { return true },
}

func Run(addr string) {
	// 启动 pprof HTTP 服务器
	go func() {
		fmt.Println(http.ListenAndServe(addr, nil))
	}()

	// 启动内存监控 HTTP 服务器
	http.HandleFunc("/", memUsageHandler)
	http.HandleFunc("/ws", wsHandler)
	go func() {
		fmt.Println(http.ListenAndServe("localhost:8080", nil))
	}()

	// 定期记录内存使用情况
	go func() {
		for {
			mu.Lock()
			memUsageHistory = append(memUsageHistory, getMemUsage())
			mu.Unlock()
			time.Sleep(10 * time.Second)
		}
	}()
}

func memUsageHandler(w http.ResponseWriter, r *http.Request) {
	const htmlTemplate = `
	<!DOCTYPE html>
	<html lang="en">
	<head>
		<meta charset="UTF-8">
		<meta name="viewport" content="width=device-width, initial-scale=1.0">
		<title>Memory Usage</title>
		<style>
			body {
				margin: 0;
				font-family: Arial, Helvetica, sans-serif;
				background-color: #f8f9fa;
			}
			.header {
				position: fixed;
				top: 0;
				width: 100%;
				background-color: #007bff;
				color: white;
				padding: 10px 0;
				text-align: center;
				box-shadow: 0 2px 5px rgba(0, 0, 0, 0.2);
			}
			.container {
				margin-top: 70px;
				padding: 20px;
			}
			.table {
				width: 100%;
				border-collapse: collapse;
			}
			.table th, .table td {
				padding: 8px;
				border: 1px solid #dee2e6;
			}
			.table th {
				background-color: #007bff;
				color: white;
			}
			.table tr:nth-child(even) {
				background-color: #f2f2f2;
			}
		</style>
	</head>
	<body>
		<div class="header">
			<h1>Memory Usage</h1>
		</div>
		<div class="container">
			<table class="table">
				<thead>
					<tr>
						<th>Timestamp</th>
						<th>Alloc (MiB)</th>
						<th>TotalAlloc (MiB)</th>
						<th>Sys (MiB)</th>
						<th>NumGC</th>
					</tr>
				</thead>
				<tbody id="memUsageTable">
					{{range .}}
					<tr>
						<td>{{.Timestamp.Format "2006-01-02 15:04:05"}}</td>
						<td>{{.Alloc}}</td>
						<td>{{.TotalAlloc}}</td>
						<td>{{.Sys}}</td>
						<td>{{.NumGC}}</td>
					</tr>
					{{end}}
				</tbody>
			</table>
		</div>
		<script>
			var ws = new WebSocket("ws://localhost:8080/ws");
			ws.onmessage = function(event) {
				var data = JSON.parse(event.data);
				var row = "<tr>";
				row += "<td>" + new Date(data.Timestamp).toLocaleString() + "</td>";
				row += "<td>" + data.Alloc + "</td>";
				row += "<td>" + data.TotalAlloc + "</td>";
				row += "<td>" + data.Sys + "</td>";
				row += "<td>" + data.NumGC + "</td>";
				row += "</tr>";
				document.getElementById("memUsageTable").innerHTML += row;
			};
		</script>
	</body>
	</html>
	`

	tmpl := template.Must(template.New("memUsage").Parse(htmlTemplate))

	mu.Lock()
	defer mu.Unlock()
	tmpl.Execute(w, memUsageHistory)
}

func wsHandler(w http.ResponseWriter, r *http.Request) {
	conn, err := upgrader.Upgrade(w, r, nil)
	if err != nil {
		fmt.Println("Upgrade error:", err)
		return
	}
	defer conn.Close()

	for {
		mu.Lock()
		if len(memUsageHistory) > 0 {
			lastUsage := memUsageHistory[len(memUsageHistory)-1]
			err := conn.WriteJSON(lastUsage)
			if err != nil {
				fmt.Println("Write error:", err)
				break
			}
		}
		mu.Unlock()
		time.Sleep(1 * time.Second)
	}
}

func getMemUsage() MemUsage {
	var m runtime.MemStats
	runtime.ReadMemStats(&m)
	return MemUsage{
		Alloc:      bToMb(m.Alloc),
		TotalAlloc: bToMb(m.TotalAlloc),
		Sys:        bToMb(m.Sys),
		NumGC:      m.NumGC,
		Timestamp:  time.Now(),
	}
}

func bToMb(b uint64) uint64 {
	return b / 1024 / 1024
}

```

go内存监控，主要使用runtime获取参数



go内存监控

今天压测一天的Mim
```go
[MIN] [2024-11-26 19:42:28] [ws/proc.go:53 -> handleOfflineMessages] [ERROR] get offline message failed
[MIN] [2024-11-26 19:42:28] [ws/handler.go:73 -> handlePrivateMessage] [ERROR] add offline 
msg failed: dial tcp 117.164.186.124:6379: bind: An operation on a socket could not be performed because the system lacked sufficient buffer space or because a queue was full.       
2024/11/26 19:42:42 get: dial tcp 117.164.186.124:6379: bind: An operation on a socket could not be performed because the system lacked sufficient buffer space or because a queue was full.
[MIN] [2024-11-26 19:42:28] [ws/handler.go:73 -> handlePrivateMessage] [ERROR] add offline 
msg failed: dial tcp 117.164.186.124:6379: bind: An operation on a socket could not be performed because the system lacked sufficient buffer space or because a queue was full.       
[MIN] [2024-11-26 19:42:28] [ws/proc.go:53 -> handleOfflineMessages] [ERROR] get offline message failed
```
压测到差不多一万个在线，且20分钟内用户间是相互发消息对话的

Tcp缓冲区和队列不够用了，就是短时间内大量连接建立，需要配置参数，这个时候需要配置，套接字上线和syn队列,tcp/ip相关参数

```md

# 编辑 /etc/sysctl.conf 文件
net.core.somaxconn = 65535  # 限制了监听套接字的最大队列长度，增加此值可以处理更多的并发连接请求。
net.core.netdev_max_backlog = 65535  # 网络接口接收数据包的队列长度，增加此值可以处理更多的网络数据包。
net.ipv4.tcp_max_syn_backlog = 65535  # TCP连接的SYN队列长度，增加此值可以处理更多的半连接状态的TCP连接。
net.ipv4.tcp_max_tw_buckets = 2000000  # TIME_WAIT状态的套接字最大数量，增加此值可以处理更多的TIME_WAIT状态的连接。
net.ipv4.tcp_tw_reuse = 1  # 允许重用TIME_WAIT状态的套接字，减少TIME_WAIT状态的连接数量，提高连接复用率。
net.ipv4.tcp_tw_recycle = 1  # 允许快速回收TIME_WAIT状态的套接字，减少TIME_WAIT状态的连接数量，提高连接复用率。
net.ipv4.tcp_fin_timeout = 30  # 处于FIN_WAIT_2状态的连接的超时时间，减少此值可以更快地回收连接。
net.ipv4.tcp_keepalive_time = 600  # TCP连接的保活时间，减少此值可以更快地检测到死连接。
net.ipv4.tcp_keepalive_probes = 5  # 保活探测次数，增加此值可以更可靠地检测到死连接。
net.ipv4.tcp_keepalive_intvl = 15  # 保活探测间隔，减少此值可以更快地检测到死连接。
net.ipv4.ip_local_port_range = 1024 65535  # 本地端口范围，增加此范围可以提供更多的可用端口。
net.ipv4.tcp_mem = 786432 1048576 1572864  # TCP内存使用限制，增加此值可以处理更多的TCP连接。
net.ipv4.tcp_rmem = 4096 87380 16777216  # TCP接收缓冲区大小，增加此值可以处理更大的数据包。
net.ipv4.tcp_wmem = 4096 65536 16777216  # TCP发送缓冲区大小，增加此值可以处理更大的数据包。
net.core.rmem_max = 16777216  # 最大接收缓冲区大小，增加此值可以处理更大的数据包。
net.core.wmem_max = 16777216  # 最大发送缓冲区大小，增加此值可以处理更大的数据包。
net.core.rmem_default = 8388608  # 默认接收缓冲区大小，增加此值可以处理更大的数据包。
net.core.wmem_default = 8388608  # 默认发送缓冲区大小，增加此值可以处理更大的数据包。
net.core.optmem_max = 40960  # 最大选项缓冲区大小，增加此值可以处理更多的TCP选项。



# 编辑 /etc/security/limits.conf 文件
* soft nofile 1048576  # 软限制，允许每个用户打开的最大文件描述符数量。
* hard nofile 1048576  # 硬限制，允许每个用户打开的最大文件描述符数量。

# 编辑 /etc/security/limits.d/90-nproc.conf 文件
* soft nproc 1048576  # 软限制，允许每个用户创建的最大进程数。
* hard nproc 1048576  # 硬限制，允许每个用户创建的最大进程数。

```

配置上面的参数就可以了，中途还遇到很多其他问题，因为整个架构按我自己的来，没有怎么借鉴别人的代码，只能慢慢优化咯



websocket压测

最近在写一个即时通讯应用，需要用到websocket，用的库是`github.com/gofiber/contrib/websocket`,简单封装的`fasthttp.websocket`,然后发现`websocket.Conn.Close()`不能关闭连接，客户端还可以发信息

因为没有这个API不是关闭ws连接的，因该手动发送关闭消息

```go
func (s *Session) Close() error {

	if s.Ws == nil {
		return errors.New("WebSocket connection is nil")
	}

	// 发送 WebSocket 关闭消息
	deadline := time.Now().Add(time.Minute)
	err := s.Ws.WriteControl(
		websocket.CloseMessage,
		websocket.FormatCloseMessage(websocket.CloseNormalClosure, ""),
		deadline,
	)
	if err != nil {
		return err
	}

	// 设置读取下一条消息的截止时间
	err = s.Ws.SetReadDeadline(time.Now().Add(5 * time.Second))
	if err != nil {
		return err
	}

	// 读取消息，直到确认关闭消息
	for {
		_, _, err = s.Ws.NextReader()

		if websocket.IsCloseError(err, websocket.CloseNormalClosure) {
			break
		}
		if err != nil {
			break
		}
	}

	// 关闭 TCP 连接
	err = s.Ws.Close()
	if err != nil {
		return err
	}

	s.Ws = nil // 将连接置为 nil，避免重复关闭
	return nil
}
```