### Go 语言中的 singleflight：避免重复请求的利器

在现代的分布式系统中，高并发是一个常见的挑战。当多个请求同时到达时，可能会导致资源浪费或系统过载。为了解决这个问题，Go 语言提供了一个非常实用的工具——`singleflight`。本文将深入探讨 `singleflight` 的工作原理、使用场景，并通过一个具体的示例来展示它的强大功能。

#### 什么是 singleflight？

`singleflight` 是 Go 语言标准库中的一个包，位于 `golang.org/x/sync/singleflight`。它的主要作用是确保在同一时间只有一个 goroutine 去执行某个操作，而其他 goroutine 则等待结果。这种机制可以有效地避免重复请求，尤其是在面对高并发场景时，能够显著减少对后端服务的压力。

#### 为什么需要 singleflight？

想象一下，你的系统中有一个缓存层，当缓存失效时，多个请求同时到达，可能会导致数据库或后端服务压力过大。使用 `singleflight` 可以确保只有一个请求去加载数据，其他请求等待结果，从而避免重复的资源消耗。

#### singleflight 的核心功能

`singleflight` 包的核心是 `Group` 类型，它提供了以下几个方法：

1. **`Do(key string, fn func() (interface{}, error)) (v interface{}, err error, shared bool)`**:
   - 这个方法会确保在同一时间只有一个 goroutine 执行 `fn`，其他 goroutine 会等待第一个 goroutine 的结果。
   - `key` 是用于标识请求的唯一键。
   - `fn` 是实际执行的函数，返回值和错误会被传递给等待的 goroutine。
   - `v` 是 `fn` 的返回值。
   - `err` 是 `fn` 的错误。
   - `shared` 表示结果是否被多个 goroutine 共享。

2. **`DoChan(key string, fn func() (interface{}, error)) <-chan Result`**:
   - 类似于 `Do`，但返回一个 `chan Result`，允许异步获取结果。
   - `Result` 是一个包含 `v`、`err` 和 `shared` 的结构体。

3. **`Forget(key string)`**:
   - 这个方法会移除 `key` 对应的请求，允许后续的请求重新执行 `fn`。

#### 使用场景

`singleflight` 通常用于以下场景：

- **缓存击穿防护**：当缓存失效时，多个请求同时到达，可能会导致数据库或后端服务压力过大。使用 `singleflight` 可以确保只有一个请求去加载数据，其他请求等待结果。
- **并发控制**：在某些情况下，多个 goroutine 可能需要执行相同的操作，但这些操作是幂等的。使用 `singleflight` 可以避免重复执行。

#### 示例代码：缓存击穿防护

让我们通过一个具体的示例来展示 `singleflight` 的使用。假设我们有一个简单的缓存系统，当缓存失效时，我们需要从数据库中加载数据。为了避免多个请求同时到达数据库，我们可以使用 `singleflight`。

```go
package main

import (
	"fmt"
	"golang.org/x/sync/singleflight"
	"time"
)

// 模拟一个耗时的数据库操作
func fetchFromDB(key string) (string, error) {
	time.Sleep(2 * time.Second)
	return fmt.Sprintf("Data for key: %s", key), nil
}

func main() {
	var g singleflight.Group

	// 模拟多个 goroutine 同时请求
	for i := 0; i < 5; i++ {
		go func(id int) {
			v, err, shared := g.Do("key", func() (interface{}, error) {
				return fetchFromDB("key")
			})
			fmt.Printf("Goroutine %d: result=%v, err=%v, shared=%v\n", id, v, err, shared)
		}(i)
	}

	// 等待所有 goroutine 完成
	time.Sleep(3 * time.Second)
}
```

#### 输出结果

```
Goroutine 4: result=Data for key: key, err=<nil>, shared=true
Goroutine 0: result=Data for key: key, err=<nil>, shared=true
Goroutine 1: result=Data for key: key, err=<nil>, shared=true
Goroutine 2: result=Data for key: key, err=<nil>, shared=true
Goroutine 3: result=Data for key: key, err=<nil>, shared=true
```

在这个例子中，所有 goroutine 都共享了同一个结果，因为 `singleflight` 确保了只有一个 goroutine 执行了 `fetchFromDB` 函数。

#### 其他应用场景

除了缓存击穿防护，`singleflight` 还可以用于以下场景：

- **幂等操作**：在某些情况下，多个 goroutine 可能需要执行相同的操作，但这些操作是幂等的。使用 `singleflight` 可以避免重复执行。
- **资源加载**：在加载资源（如配置文件、静态文件等）时，可以使用 `singleflight` 确保只有一个 goroutine 去加载资源，其他 goroutine 等待结果。

#### 总结

`singleflight` 是一个非常有用的工具，特别是在需要减少重复请求或控制并发访问的场景中。它可以帮助你避免不必要的资源消耗，提高系统的性能和稳定性。通过本文的介绍和示例，相信你已经对 `singleflight` 有了更深入的了解。希望你在实际项目中能够灵活运用这一强大的工具，提升系统的并发处理能力。

Go 语言中的 singleflight：避免重复请求的利器


#### 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  # 硬限制，允许每个用户创建的最大进程数。

```

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

