# WebSocket

**Repository Path**: vipkwds/ws

## Basic Information

- **Project Name**: WebSocket
- **Description**: Go 语言实现的高可用、高复用 WebSocket 组件库，支持客户端和服务端
- **Primary Language**: Go
- **License**: MIT
- **Default Branch**: master
- **Homepage**: None
- **GVP Project**: No

## Statistics

- **Stars**: 0
- **Forks**: 0
- **Created**: 2026-06-22
- **Last Updated**: 2026-06-22

## Categories & Tags

**Categories**: Uncategorized

**Tags**: None

## README

# WebSocket 高可用通用组件库

Go 语言实现的高可用、高复用 WebSocket 组件库，支持客户端和服务端。

## 特性

| 特性 | 说明 |
|------|------|
| **高可用** | 心跳检测、断线重连、熔断器、消息确认重发 |
| **多连接池** | Least-Pending 调度策略，支持背压 |
| **双协议支持** | JSON 和 Protobuf 编解码 |
| **At-Least-Once** | Ack 确认机制，保证消息可靠传递 |
| **最小依赖** | 仅依赖 `gorilla/websocket` + `google.golang.org/protobuf` |

## 架构

```
ws/
├── client/              # 客户端实现
│   ├── client.go        # 客户端主入口
│   ├── pool.go          # 连接池（Least-Pending 调度）
│   ├── session.go       # 客户端会话
│   └── event.go         # 事件回调
│
├── server/              # 服务端实现
│   ├── server.go        # 服务端主入口
│   ├── session.go       # 服务端会话
│   ├── handler.go       # 业务 Handler
│   └── upgrader.go      # HTTP Upgrade
│
├── core/                # 核心抽象层
│   ├── session.go       # Session 接口
│   ├── message.go       # Message 结构
│   ├── state.go         # 状态机
│   └── errors.go        # 统一错误
│
├── protocol/            # 协议层
│   ├── codec.go         # Codec 接口
│   ├── json.go          # JSON Codec
│   ├── protobuf.go      # Protobuf Codec
│   └── packet.go        # 协议包设计
│
├── reliability/         # 可靠性机制
│   ├── ack.go           # Ack 确认 + 重发
│   ├── retry.go         # 重试策略
│   ├── circuitbreaker.go# 熔断器
│   └── heartbeat.go     # 双向心跳
│
├── flowctrl/            # 流量控制
│   ├── queue.go         # 发送队列（背压）
│   └── window.go        # 滑动窗口
│
└── util/                # 工具
    └── id.go            # 自增 ID 生成器
```

## 核心设计

### 协议包格式

```
┌──────────────────────────────────────────────────┐
│  Version (1) │  Type (1)  │  Seq (8) │  Length (4)│  <- 固定头 14 字节
├──────────────────────────────────────────────────┤
│                  Payload (JSON/PB)               │
└──────────────────────────────────────────────────┘

字段大小（字节）：Version=1 + Type=1 + Seq=8 + Length=4 = 14
Type: Handshake | Data | Ack | Ping | Pong | Close
Seq:  消息序列号（用于 Ack 匹配）
Length: Payload 长度（4字节 uint32）
```

### Ack 确认机制

- 发送消息时带 Seq，存入 pending 队列
- 接收方回复 Ack(Seq) 表示已收到
- 超时未收到 Ack 则重发（最大 3 次）
- 接收方根据 Seq 做去重处理

### 心跳机制

- 双向心跳：应用层 Ping/Pong + TCP Keep-Alive
- 心跳间隔：10s，超时判定：30s
- 连续 3 次超时触发断连

### 熔断器

- 连续 5 次失败触发熔断
- 熔断持续 30s
- 之后进入 HalfOpen 状态探测恢复

### 连接池调度

- Least-Pending 策略：选择待确认消息最少的连接
- 背压机制：队列满时发送方阻塞等待
- 定期健康检查，自动剔除不健康连接

## 核心接口

### Session 接口

```go
type Session interface {
    ID() string
    State() State
    Send(ctx context.Context, msg any) error
    Receive(ctx context.Context) (any, error)
    Close(err error) error
}
```

### Codec 接口

```go
type Codec interface {
    Marshal(any) ([]byte, error)
    Unmarshal([]byte, any) error
    Name() string
}
```

### Handler 接口（服务端）

```go
type Handler interface {
    Handle(ctx context.Context, sess Session, msg any)
}
```

## 快速开始

### 客户端示例

```go
import (
    wsclient "ws/client"
    "ws/core"
    "ws/protocol"
)

eventHandler := &wsclient.ClientEventHandler{
    OnConnect: func(sess core.Session) {
        fmt.Println("connected:", sess.ID())
    },
    OnDisconnect: func(sess core.Session, err error) {
        fmt.Println("disconnected:", err)
    },
    OnMessage: func(sess core.Session, msg any) {
        fmt.Printf("received: %v\n", msg)
    },
}

client := wsclient.NewClient(
    "ws://localhost:8080/ws",
    &protocol.JSONCodec{},
    3,                  // poolSize
    eventHandler,
)

ctx := context.Background()
if err := client.Dial(ctx); err != nil {
    log.Fatal(err)
}
defer client.Close()

// 发送消息（可靠传递，At-Least-Once）
if err := client.Send(ctx, &MyMessage{Data: "hello"}); err != nil {
    log.Printf("send error: %v", err)
}
```

### 服务端示例

```go
import (
    "ws/core"
    "ws/protocol"
    "ws/server"
)

type MyHandler struct{}

func (h *MyHandler) Handle(ctx context.Context, sess core.Session, msg any) {
    fmt.Printf("received from %s: %v\n", sess.ID(), msg)
    sess.Send(ctx, &Response{Result: "ok"})
}

type Response struct {
    Result string
}

server := server.NewServer(
    ":8080",
    &protocol.JSONCodec{},
    &MyHandler{},
)

fmt.Println("server started on :8080")
if err := server.ListenAndServe(); err != nil {
    log.Fatal(err)
}
```

## 状态机

```
            ┌──────────┐
            │  Init    │
            └────┬─────┘
                 │ Dial
                 ▼
          ┌─────────────┐
    ┌───►│  Connecting │◄────────┐
    │    └──────┬──────┘         │
    │           │ Connect OK     │ Reconnecting
    │           ▼                │
    │    ┌────────────┐          │
    │    │ Connected  │──────────┘
    │    └──────┬─────┘
    │           │ Heartbeat Timeout / Error
    │           ▼
    │    ┌──────────────┐
    │    │ Reconnecting │
    │    └──────┬───────┘
    │           │ Max retries exceeded
    │           ▼
    │    ┌──────────┐
    └────│  Closed  │
         └──────────┘
```

## 依赖

| 依赖 | 版本 | 用途 |
|------|------|------|
| gorilla/websocket | v1.5.x | WebSocket 底层实现 |
| google.golang.org/protobuf | v1.33.x | Protobuf 编解码 |

## 注意事项

- TLS 由 Nginx 等代理层处理，应用层使用 WS 而非 WSS
- Session ID 采用自增整数生成
- 消息可靠性级别为 At-Least-Once
- 连接池大小由 `poolSize` 参数指定（最小连接数 = 最大连接数 = poolSize）
- Ack 超时时间 3 秒，最大重试 3 次
- 滑动窗口大小 100，发送队列大小 1000