```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
}
```

websocket关闭

直接用BeforCreate Hook就行
```go
// pkg/snowflake/snowflake.go
package snowflake

import (
	"net"

	"github.com/sony/sonyflake"
)

var sf *sonyflake.Sonyflake

func init() {
	// 初始化雪花算法生成器
	var err error
	// 手动设置机器 ID
	var machineID uint16
	if ip, err := getPrivateIP(); err == nil {
		machineID = getMachineIDFromIP(ip)
	} else {
		machineID = 1 // 如果找不到私有 IP，使用默认值
	}

	settings := sonyflake.Settings{
		MachineID: func() (uint16, error) {
			return machineID, nil
		},
	}

	sf, err = sonyflake.New(settings)

	if err != nil {
		panic(err)
	}

	if sf == nil {
		panic("sonyflake not initialized")
	}
}

// NextID 生成一个新的雪花 ID
func NextID() (uint64, error) {
	return sf.NextID()
}

// getPrivateIP 获取私有 IP 地址
func getPrivateIP() (net.IP, error) {
	addrs, err := net.InterfaceAddrs()
	if err != nil {
		return nil, err
	}

	for _, addr := range addrs {
		ipNet, ok := addr.(*net.IPNet)
		if ok && !ipNet.IP.IsLoopback() && ipNet.IP.To4() != nil {
			return ipNet.IP, nil
		}
	}

	return nil, nil
}

// getMachineIDFromIP 从 IP 地址生成机器 ID
func getMachineIDFromIP(ip net.IP) uint16 {
	ip = ip.To4()
	return uint16(ip[2])<<8 + uint16(ip[3])
}

```
先安装库封装一个包，我的个人PC获取不到私人IP,所以要手动获取

```go
package user

import (
	"time"

	sf "mim/pkg/snowflake"

	"gorm.io/gorm"
)

// User represents the user model.
// @Description User model
// @Description This model represents a user in the system.
type User struct {
	UserName      string         `gorm:"type:varchar(20)" json:"username" swagger:"username,required"`
	PassWord      string         `gorm:"type:varchar(20)" json:"password" swagger:"password,required"`
	LastLoginTime time.Time      `gorm:"type:timestamp" json:"last_login_time" swaggertype:"primitive,string" format:"date-time"`
	Id            uint64         `gorm:"primaryKey" json:"id" swaggerignore:"true"`
	CreateAt      time.Time      `gorm:"type:timestamp" json:"create_at" swaggertype:"primitive,string" format:"date-time"`
	UpdateAt      time.Time      `gorm:"type:timestamp" json:"update_at" swaggertype:"primitive,string" format:"date-time"`
	DeleteAt      gorm.DeletedAt `gorm:"index" json:"delete_at" swaggerignore:"true"` // 忽略 DeleteAt 字段
}


// BeforeCreate is a GORM hook that is called before creating a new user.
func (u *User) BeforeCreate(tx *gorm.DB) (err error) {
	id, err := sf.NextID()
	if err != nil {
		return err
	}
	u.Id = id
	u.CreateAt = time.Now()
	u.UpdateAt = time.Now()
	return
}

// BeforeUpdate is a GORM hook that is called before updating a user.
func (u *User) BeforeUpdate(tx *gorm.DB) (err error) {
	u.UpdateAt = time.Now()
	return
}
```




Gorm使用雪花算法生成ID

本教程主要演示使用srs搭建一个流媒体服务器然后用OBS进行rtmp推流，转码成flv给前端访问

## 启动一个srs服务器

我们使用下面的镜像源拉取SRS4.0的镜像
```sh
docker pull docker.unsee.tech/ossrs/srs:4
```
创建配置文件`srs.conf`
```srs.conf
listen              1935;
max_connections     999;
srs_log_tank        file;
srs_log_file        ./objs/srs.log;
http_server {
    enabled         on;
    listen          8080;
    dir             ./objs/nginx/html;
}
vhost __defaultVhost__ {
    hls {
        enabled         on;
    }
    http_remux {
        enabled     on;
        mount       [vhost]/[app]/[stream].flv;
    }
}
```
直接运行
```sh
 docker run --rm -p 1935:1935 -p 8090:8080 -v $(pwd)/srs.conf:/usr/local/srs/conf/srs.conf ossrs/srs:4
```
然后通过8090端口访问srs web服务


![image.png](/static/img/67bd9597cc1d379b121a74a8eee80d16.image.webp)
## OBS推流

这样就运行成功了

接下来去OBS官网下载[OBS](https://obsproject.com/zh-cn)

安装好后进去设置


![image.png](/static/img/c80eac556d52242b37fee65c053d9cca.image.webp)

配置一下推流地址


![image.png](/static/img/c6035a099bc8deb6237639f08135e609.image.webp)
接下来就能通过你的服务地址+推流码访问，推流码随便设置，地址默认是 rtmp://ip/live

## 访问直播流
然后回到上面的srs主页



![image.png](/static/img/49a97c8bfa9fae37d17a64a644a50889.image.webp)

将url改成你的推流地址+推流码

![image.png](/static/img/561dc6334bc6e237dea8b7f1af4b5f59.image.webp)

## 前端页面访问

当然我们要实现的是通过前端访问，这样才能实现直播的需求

下面是一个demo页面，使用flv.js，不管是用vue还是react都能用这种方式

```html
<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>FLV Streaming Test</title>
    <style>
		.back {
			width: 100%;
			height: 100vh;
			display: flex;
			justify-content: center;
			align-items: center;

		}
		.content {
			width: 800px;
			height: 600px;
		}

		#video {
			width: 100%;
			height: 100%;
		}


    </style>
</head>
<body>
	<div class="back">
		<div class="content">

			<video id="video" controls></video>

		</div>
	</div>
    

    <script src="https://cdn.jsdelivr.net/npm/flv.js/dist/flv.min.js"></script>
    <script>
        if (flvjs.isSupported()) {
            const videoElement = document.getElementById('video');
            const flvPlayer = flvjs.createPlayer({
                type: 'flv',
                url: 'http://117.164.186.124:8090/live/01.flv'
            });
            flvPlayer.attachMediaElement(videoElement);
            flvPlayer.load();
            flvPlayer.play();
        }
    </script>
</body>
</html>
```


![image.png](/static/img/181eddbcb4cbf266f5c4de08b86df617.image.webp)