今天压测一天的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)

SRS流媒体服务器搭建

## 项目结构

在Spring和Spring Boot项目中，项目结构通常遵循一定的分层设计原则，以保持代码的清晰和可维护性。常见的项目结构如下：

```
src
├── main
│   ├── java
│   │   └── com
│   │       └── example
│   │           ├── controller  // 控制层，处理HTTP请求
│   │           ├── service     // 服务层，处理业务逻辑
│   │           ├── mapper      // MyBatis Mapper接口
│   │           ├── entity      // 实体类，对应数据库表
│   │           ├── config      // 配置类，如Spring配置、MyBatis配置等
│   │           ├── interceptor // 拦截器
│   │           ├── filter      // 过滤器
│   │           └── Application.java // 程序入口
│   └── resources
│       ├── application.yml     // 配置文件
│       ├── mapper              // MyBatis XML映射文件
│       └── static              // 静态资源
└── test
    └── java
        └── com
            └── example
                └── test         // 测试类
```

## 配置文件

Spring Boot使用`application.yml`或`application.properties`作为主要的配置文件。常见的配置项包括：

- **服务器配置**：如端口号、上下文路径等。
- **数据库配置**：如数据库URL、用户名、密码等。
- **日志配置**：如日志级别、日志文件路径等。
- **缓存配置**：如Redis配置。

示例：

```yaml
server:
  port: 8080

spring:
  datasource:
    url: jdbc:mysql://localhost:3306/mydb
    username: root
    password: root

logging:
  level:
    com.example: DEBUG
```

## 程序入口

Spring Boot项目的入口通常是一个带有`@SpringBootApplication`注解的类，该类会启动Spring Boot应用。

```java
package com.example;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication
public class Application {
    public static void main(String[] args) {
        SpringApplication.run(Application.class, args);
    }
}
```

## 路由

在Spring Boot中，路由通常由控制器（Controller）类来处理。使用`@RequestMapping`、`@GetMapping`、`@PostMapping`等注解来定义路由。

```java
package com.example.controller;

import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

@RestController
@RequestMapping("/api")
public class UserController {

    @GetMapping("/users")
    public String getUsers() {
        return "List of users";
    }
}
```

## 传参

Spring Boot支持多种传参方式：

- **路径参数**：使用`@PathVariable`注解。
- **查询参数**：使用`@RequestParam`注解。
- **请求体参数**：使用`@RequestBody`注解。

示例：

```java
@GetMapping("/users/{id}")
public String getUserById(@PathVariable Long id) {
    return "User ID: " + id;
}

@GetMapping("/users")
public String getUsers(@RequestParam String name) {
    return "User name: " + name;
}

@PostMapping("/users")
public String createUser(@RequestBody User user) {
    return "Created user: " + user.getName();
}
```

## 拦截器

拦截器（Interceptor）用于在请求处理前后执行一些操作，如权限验证、日志记录等。通过实现`HandlerInterceptor`接口来创建拦截器。

```java
package com.example.interceptor;

import org.springframework.web.servlet.HandlerInterceptor;
import org.springframework.web.servlet.ModelAndView;

import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;

public class MyInterceptor implements HandlerInterceptor {

    @Override
    public boolean preHandle(HttpServletRequest request, HttpServletResponse response, Object handler) throws Exception {
        // 请求处理前的逻辑
        return true;
    }

    @Override
    public void postHandle(HttpServletRequest request, HttpServletResponse response, Object handler, ModelAndView modelAndView) throws Exception {
        // 请求处理后的逻辑
    }

    @Override
    public void afterCompletion(HttpServletRequest request, HttpServletResponse response, Object handler, Exception ex) throws Exception {
        // 请求完成后的逻辑
    }
}
```

在配置类中注册拦截器：

```java
package com.example.config;

import com.example.interceptor.MyInterceptor;
import org.springframework.context.annotation.Configuration;
import org.springframework.web.servlet.config.annotation.InterceptorRegistry;
import org.springframework.web.servlet.config.annotation.WebMvcConfigurer;

@Configuration
public class WebConfig implements WebMvcConfigurer {

    @Override
    public void addInterceptors(InterceptorRegistry registry) {
        registry.addInterceptor(new MyInterceptor()).addPathPatterns("/api/**");
    }
}
```

## 过滤器

过滤器（Filter）用于在请求到达Servlet之前或之后执行一些操作。通过实现`Filter`接口来创建过滤器。

```java
package com.example.filter;

import javax.servlet.*;
import java.io.IOException;

public class MyFilter implements Filter {

    @Override
    public void doFilter(ServletRequest request, ServletResponse response, FilterChain chain) throws IOException, ServletException {
        // 请求处理前的逻辑
        chain.doFilter(request, response);
        // 请求处理后的逻辑
    }
}
```

在配置类中注册过滤器：

```java
package com.example.config;

import com.example.filter.MyFilter;
import org.springframework.boot.web.servlet.FilterRegistrationBean;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class FilterConfig {

    @Bean
    public FilterRegistrationBean<MyFilter> myFilter() {
        FilterRegistrationBean<MyFilter> registrationBean = new FilterRegistrationBean<>();
        registrationBean.setFilter(new MyFilter());
        registrationBean.addUrlPatterns("/api/*");
        return registrationBean;
    }
}
```

## 数据库操作

### 基本增删查改

MyBatis-Plus提供了非常便捷的CRUD操作，通过继承`BaseMapper`接口即可实现。

```java
package com.example.mapper;

import com.baomidou.mybatisplus.core.mapper.BaseMapper;
import com.example.entity.User;
import org.apache.ibatis.annotations.Mapper;

@Mapper
public interface UserMapper extends BaseMapper<User> {
}
```

在Service层中使用：

```java
package com.example.service;

import com.example.entity.User;
import com.example.mapper.UserMapper;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

@Service
public class UserService {

    @Autowired
    private UserMapper userMapper;

    public User getUserById(Long id) {
        return userMapper.selectById(id);
    }

    public void createUser(User user) {
        userMapper.insert(user);
    }

    public void updateUser(User user) {
        userMapper.updateById(user);
    }

    public void deleteUser(Long id) {
        userMapper.deleteById(id);
    }
}
```

### 条件查询

MyBatis-Plus提供了强大的条件构造器`QueryWrapper`，用于构建复杂的查询条件。

```java
package com.example.service;

import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.example.entity.User;
import com.example.mapper.UserMapper;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.List;

@Service
public class UserService {

    @Autowired
    private UserMapper userMapper;

    public List<User> getUsersByName(String name) {
        QueryWrapper<User> queryWrapper = new QueryWrapper<>();
        queryWrapper.eq("name", name);
        return userMapper.selectList(queryWrapper);
    }
}
```

### 事务

Spring提供了声明式事务管理，通过`@Transactional`注解来实现。

```java
package com.example.service;

import com.example.entity.User;
import com.example.mapper.UserMapper;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

@Service
public class UserService {

    @Autowired
    private UserMapper userMapper;

    @Transactional
    public void createUserWithTransaction(User user) {
        userMapper.insert(user);
        // 如果发生异常，事务会回滚
        if (true) {
            throw new RuntimeException("Transaction will rollback");
        }
    }
}
```

### 钩子

MyBatis-Plus提供了一些钩子方法，如`@TableField`注解中的`fill`属性，用于在插入或更新时自动填充字段。

```java
package com.example.entity;

import com.baomidou.mybatisplus.annotation.TableField;
import com.baomidou.mybatisplus.annotation.TableId;
import com.baomidou.mybatisplus.annotation.TableName;
import com.baomidou.mybatisplus.core.handlers.MetaObjectHandler;
import org.apache.ibatis.reflection.MetaObject;
import org.springframework.stereotype.Component;

import java.util.Date;

@TableName("user")
public class User {

    @TableId
    private Long id;

    private String name;

    @TableField(fill = FieldFill.INSERT)
    private Date createTime;

    @TableField(fill = FieldFill.INSERT_UPDATE)
    private Date updateTime;
}

@Component
public class MyMetaObjectHandler implements MetaObjectHandler {

    @Override
    public void insertFill(MetaObject metaObject) {
        this.strictInsertFill(metaObject, "createTime", Date.class, new Date());
        this.strictInsertFill(metaObject, "updateTime", Date.class, new Date());
    }

    @Override
    public void updateFill(MetaObject metaObject) {
        this.strictUpdateFill(metaObject, "updateTime", Date.class, new Date());
    }
}
```

## 中间件集成

### Redis集成

Spring Boot提供了对Redis的集成支持，通过`spring-boot-starter-data-redis`依赖即可实现。

在`application.yml`中配置Redis：

```yaml
spring:
  redis:
    host: localhost
    port: 6379
```

使用RedisTemplate进行操作：

```java
package com.example.service;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;

@Service
public class RedisService {

    @Autowired
    private RedisTemplate<String, String> redisTemplate;

    public void setValue(String key, String value) {
        redisTemplate.opsForValue().set(key, value);
    }

    public String getValue(String key) {
        return redisTemplate.opsForValue().get(key);
    }
}
```

通过以上配置和代码示例，你可以快速搭建一个基于Spring Boot和MyBatis-Plus的项目，并集成Redis等中间件。