GWS
简å•, 快速, å¯é çš„ WebSocket æœåŠ¡å™¨å’Œå®¢æˆ·ç«¯
[](https://github.com/avelino/awesome-go#networking) [](https://codecov.io/gh/lxzan/gws) [](https://github.com/lxzan/gws/actions/workflows/go.yml) [](https://goreportcard.com/report/github.com/lxzan/gws) [](LICENSE) [](https://github.com/lxzan/gws)
### 介ç»
GWS（Go WebSocket）是一个用 Go 编写的éžå¸¸ç®€å•ã€å¿«é€Ÿã€å¯é 且功能丰富的 WebSocket 实现。它设计用于高并å‘环境，构建"接å£", "代ç†", "游æˆ", "æµåª’体", "消æ¯å‘布订阅"ç‰æœåŠ¡ã€‚它æä¾›éžå¸¸ç®€å•çš„ API,
您å¯ä»¥è½»æ¾ç¼–写自己的æœåŠ¡å™¨æˆ–客户端。
### 为什么选择 GWS
- 简å•æ˜“用
- **用户å‹å¥½**: 简æ´æ˜Žäº†çš„ `WebSocket` 事件接å£è®¾è®¡ï¼Œè®©æœåŠ¡å™¨å’Œå®¢æˆ·ç«¯çš„交互å˜å¾—è½»æ¾ç®€å•.
- **ç¼–ç 效率**: 最大é™åº¦åœ°å‡å°‘实施å¤æ‚的解决方案所需的代ç é‡.
- 性能出众
- **高åžå低延迟**: ä¸“ä¸ºå¿«é€Ÿä¼ è¾“å’ŒæŽ¥æ”¶æ•°æ®è€Œè®¾è®¡ï¼Œæ˜¯æ—¶é—´æ•æ„Ÿåž‹åº”用的ç†æƒ³ä¹‹é€‰.
- **低内å˜å 用**: 高度优化的内å˜å¤ç”¨ç³»ç»Ÿ, 最大é™åº¦é™ä½Žå†…å˜ä½¿ç”¨é‡ï¼Œé™ä½Žæ‚¨çš„æˆæœ¬.
- 稳定å¯é
- **å¥å£®çš„错误处ç†**: 管ç†å’Œå‡å°‘错误的先进机制，确ä¿æŒç»è¿è¡Œ.
- **完善的测试用例**: 通过了所有 `Autobahn` 测试用例, ç¬¦åˆ `RFC 7692` æ ‡å‡†. å•å…ƒæµ‹è¯•è¦†ç›–çŽ‡å‡ ä¹Žè¾¾åˆ° 100%, 覆盖所有æ¡ä»¶åˆ†æ”¯.
### 基准测试
#### IOPS (Echo Server)
GOMAXPROCS=4, Connection=1000, CompressEnabled=false
#### GoBench
```go
go test -benchmem -run=^$ -bench . github.com/lxzan/gws
goos: linux
goarch: amd64
pkg: github.com/lxzan/gws
cpu: AMD Ryzen 5 PRO 4650G with Radeon Graphics
BenchmarkConn_WriteMessage/compress_disabled-12 5263632 232.3 ns/op 24 B/op 1 allocs/op
BenchmarkConn_WriteMessage/compress_enabled-12 99663 11265 ns/op 386 B/op 1 allocs/op
BenchmarkConn_ReadMessage/compress_disabled-12 7809654 152.4 ns/op 8 B/op 0 allocs/op
BenchmarkConn_ReadMessage/compress_enabled-12 326257 3133 ns/op 81 B/op 1 allocs/op
PASS
ok github.com/lxzan/gws 17.231s
```
### Index
- [介ç»](#介ç»)
- [为什么选择 GWS](#为什么选择-gws)
- [基准测试](#基准测试)
- [IOPS (Echo Server)](#iops-echo-server)
- [GoBench](#gobench)
- [Index](#index)
- [特性](#特性)
- [注æ„](#注æ„)
- [安装](#安装)
- [事件](#事件)
- [快速上手](#快速上手)
- [最佳实践](#最佳实践)
- [更多用例](#更多用例)
- [KCP](#kcp)
- [代ç†](#代ç†)
- [广æ’](#广æ’)
- [写入超时](#写入超时)
- [å‘布/订阅](#å‘布订阅)
- [Autobahn 测试](#autobahn-测试)
- [交æµ](#交æµ)
- [致谢](#致谢)
### 特性
- [x] äº‹ä»¶é©±åŠ¨å¼ API
- [x] 广æ’
- [x] 代ç†æ‹¨å·
- [x] 上下文接管
- [x] 支æŒå¹¶å‘和异æ¥éžé˜»å¡žå†™å…¥
- [x] 通过所有 Autobahn 测试用例 [Server](https://lxzan.github.io/gws/reports/servers/) / [Client](https://lxzan.github.io/gws/reports/clients/)
### 注æ„
- 所有 gws.Conn 导出方法返回的错误都是å¯å¿½ç•¥çš„, 它们在内部已ç»è¢«å¦¥å–„处ç†äº†
- ä¼ è¾“å¤§æ–‡ä»¶æœ‰é˜»å¡žè¿žæŽ¥çš„é£Žé™©
- 如果å¤ç”¨ HTTP æœåŠ¡å™¨, 建议开å¯æ–°çš„ Goroutine æ¥è°ƒç”¨ ReadLoop, 以é¿å…请求上下文内å˜ä¸èƒ½åŠæ—¶å›žæ”¶.
### 安装
```bash
go get -v github.com/lxzan/gws@latest
```
### 事件
```go
type Event interface {
OnOpen(socket *Conn) // connection is established
OnClose(socket *Conn, err error) // received a close frame or input/output error occurs
OnPing(socket *Conn, payload []byte) // received a ping frame
OnPong(socket *Conn, payload []byte) // received a pong frame
OnMessage(socket *Conn, message *Message) // received a text/binary frame
}
```
### 快速上手
```go
package main
import "github.com/lxzan/gws"
func main() {
gws.NewServer(&gws.BuiltinEventHandler{}, nil).Run(":6666")
}
```
### 最佳实践
```go
package main
import (
"net/http"
"time"
"github.com/lxzan/gws"
)
const (
PingInterval = 5 * time.Second
PingWait = 10 * time.Second
)
func main() {
upgrader := gws.NewUpgrader(&Handler{}, &gws.ServerOption{
ParallelEnabled: true, // å¼€å¯å¹¶è¡Œæ¶ˆæ¯å¤„ç†
Recovery: gws.Recovery, // å¼€å¯å¼‚常æ¢å¤
PermessageDeflate: gws.PermessageDeflate{Enabled: true}, // å¼€å¯åŽ‹ç¼©
})
http.HandleFunc("/connect", func(writer http.ResponseWriter, request *http.Request) {
socket, err := upgrader.Upgrade(writer, request)
if err != nil {
return
}
go func() {
socket.ReadLoop() // æ¤å¤„阻塞会使请求上下文ä¸èƒ½é¡ºåˆ©è¢«GC
}()
})
http.ListenAndServe(":6666", nil)
}
type Handler struct{}
func (c *Handler) OnOpen(socket *gws.Conn) {
_ = socket.SetDeadline(time.Now().Add(PingInterval + PingWait))
}
func (c *Handler) OnClose(socket *gws.Conn, err error) {}
func (c *Handler) OnPing(socket *gws.Conn, payload []byte) {
_ = socket.SetDeadline(time.Now().Add(PingInterval + PingWait))
_ = socket.WritePong(nil)
}
func (c *Handler) OnPong(socket *gws.Conn, payload []byte) {}
func (c *Handler) OnMessage(socket *gws.Conn, message *gws.Message) {
defer message.Close()
socket.WriteMessage(message.Opcode, message.Bytes())
}
```
### 更多用例
#### KCP
- server
```go
package main
import (
"log"
"github.com/lxzan/gws"
kcp "github.com/xtaci/kcp-go"
)
func main() {
listener, err := kcp.Listen(":6666")
if err != nil {
log.Println(err.Error())
return
}
app := gws.NewServer(&gws.BuiltinEventHandler{}, nil)
app.RunListener(listener)
}
```
- client
```go
package main
import (
"github.com/lxzan/gws"
kcp "github.com/xtaci/kcp-go"
"log"
)
func main() {
conn, err := kcp.Dial("127.0.0.1:6666")
if err != nil {
log.Println(err.Error())
return
}
app, _, err := gws.NewClientFromConn(&gws.BuiltinEventHandler{}, nil, conn)
if err != nil {
log.Println(err.Error())
return
}
app.ReadLoop()
}
```
#### 代ç†
通过代ç†æ‹¨å·, 使用 socks5 åè®®.
```go
package main
import (
"crypto/tls"
"github.com/lxzan/gws"
"golang.org/x/net/proxy"
"log"
)
func main() {
socket, _, err := gws.NewClient(new(gws.BuiltinEventHandler), &gws.ClientOption{
Addr: "wss://example.com/connect",
TlsConfig: &tls.Config{InsecureSkipVerify: true},
NewDialer: func() (gws.Dialer, error) {
return proxy.SOCKS5("tcp", "127.0.0.1:1080", nil, nil)
},
PermessageDeflate: gws.PermessageDeflate{
Enabled: true,
ServerContextTakeover: true,
ClientContextTakeover: true,
},
})
if err != nil {
log.Println(err.Error())
return
}
socket.ReadLoop()
}
```
#### 广æ’
先创建一个 Broadcaster 实例，然åŽåœ¨å¾ªçŽ¯ä¸è°ƒç”¨ Broadcast 方法å‘æ¯ä¸ªå®¢æˆ·ç«¯å‘é€æ¶ˆæ¯ï¼Œæœ€åŽå…³é—
广æ’程åºä»¥å›žæ”¶å†…å˜ã€‚整个过程ä¸æ¶ˆæ¯åªä¼šè¢«åŽ‹ç¼©ä¸€æ¬¡ã€‚
```go
func Broadcast(conns []*gws.Conn, opcode gws.Opcode, payload []byte) {
var b = gws.NewBroadcaster(opcode, payload)
defer b.Close()
for _, item := range conns {
_ = b.Broadcast(item)
}
}
```
#### 写入超时
`SetDeadline` å¯ä»¥è¦†ç›–大部分使用场景, 想è¦ç²¾ç»†åœ°æŽ§åˆ¶æ¯ä¸€æ¬¡å†™å…¥çš„超时时间, 则需è¦è‡ªè¡Œå°è£…下 `WriteWithTimeout`
函数, `timer` 的创建和销æ¯ä¼šæœ‰ä¸€å®šé¢å¤–开销.
```go
func WriteWithTimeout(socket *gws.Conn, p []byte, timeout time.Duration) error {
var sig = atomic.Uint32{}
var timer = time.AfterFunc(timeout, func() {
if sig.CompareAndSwap(0, 1) {
socket.WriteClose(1000, []byte("write timeout"))
}
})
var err = socket.WriteMessage(gws.OpcodeText, p)
if sig.CompareAndSwap(0, 1) {
timer.Stop()
}
return err
}
```
#### å‘布/订阅
使用 event_emitter 包实现å‘布订阅模å¼ã€‚用结构体包装 `gws.Conn`，并实现 GetSubscriberID 方法以获å–订阅 ID，该 ID 必须是唯一的。订阅 ID 用于识别订阅者，订阅者åªèƒ½æŽ¥æ”¶å…¶è®¢é˜…主题的消æ¯ã€‚
æ¤ç¤ºä¾‹å¯¹äºŽä½¿ç”¨ gws 构建èŠå¤©å®¤æˆ–消æ¯æŽ¨é€éžå¸¸æœ‰ç”¨ã€‚è¿™æ„味ç€ç”¨æˆ·å¯ä»¥é€šè¿‡ websocket 订阅一个或多个主题，当å‘该主题å‘布消æ¯æ—¶ï¼Œæ‰€æœ‰è®¢é˜…用户都会收到消æ¯ã€‚
```go
package main
import (
"github.com/lxzan/event_emitter"
"github.com/lxzan/gws"
)
type Subscriber gws.Conn
func NewSubscriber(conn *gws.Conn) *Subscriber { return (*Subscriber)(conn) }
func (c *Subscriber) GetSubscriberID() int64 {
userId, _ := c.GetMetadata().Load("userId")
return userId.(int64)
}
func (c *Subscriber) GetMetadata() event_emitter.Metadata { return c.Conn().Session() }
func (c *Subscriber) Conn() *gws.Conn { return (*gws.Conn)(c) }
func Subscribe(em *event_emitter.EventEmitter[int64, *Subscriber], s *Subscriber, topic string) {
em.Subscribe(s, topic, func(msg any) {
_ = msg.(*gws.Broadcaster).Broadcast(s.Conn())
})
}
func Publish(em *event_emitter.EventEmitter[int64, *Subscriber], topic string, msg []byte) {
var broadcaster = gws.NewBroadcaster(gws.OpcodeText, msg)
defer broadcaster.Close()
em.Publish(topic, broadcaster)
}
```
### Autobahn 测试
```bash
cd examples/autobahn
mkdir reports
docker run -it --rm \
-v ${PWD}/config:/config \
-v ${PWD}/reports:/reports \
crossbario/autobahn-testsuite \
wstest -m fuzzingclient -s /config/fuzzingclient.json
```
### 交æµ
> 微信需è¦å…ˆæ·»åŠ 好å‹å†æ‹‰ç¾¤, 请注明æ¥è‡ª GitHub
### 致谢
- [crossbario/autobahn-testsuite](https://github.com/crossbario/autobahn-testsuite)
- [klauspost/compress](https://github.com/klauspost/compress)
- [lesismal/nbio](https://github.com/lesismal/nbio)
