Distributed WebSocket

特性

架构

┌──────────────────┐ ┌───────────────────┐ ┌───────────────┐
│ WebSocket 客户端 │────▶│ WebSocket 服务节点 │◀───▶│ Redis/Etcd │
└──────────────────┘ └───────────────────┘ └───────────────┘
 │ ▲さんかく
 ▼ │
 ┌──────────────────┐
 │ gRPC │
 │ 服务接口 │
 └──────────────────┘

安装

go get github.com/jayecc/go-websocket

依赖

快速开始

单节点模式

package main
import (
 "log"
 "net/http"
 "time"
 
 "github.com/gin-gonic/gin"
 websocket "github.com/jayecc/go-websocket"
)
func main() {
 gin.SetMode(gin.ReleaseMode)
 gin.DisableConsoleColor()
 app := gin.Default()
 // 创建WebSocket Hub
 hub := websocket.NewHubRun()
 defer hub.Close()
 // 注册WebSocket路由
 app.GET("/ws", func(ctx *gin.Context) {
 client := websocket.NewClient(hub, websocket.WithId("xxxx"))
 // 设置连接回调
 client.OnConnect(func(conn *websocket.Client) {
 log.Printf("Client %s connected", conn.GetID())
 })
 // 设置消息处理回调
 client.OnEvent(func(conn *websocket.Client, messageType int, message []byte) {
 log.Printf("Received message from client %s: %s", conn.GetID(), string(message))
 // 发送服务器时间作为响应
 response := time.Now().Format(time.RFC3339)
 conn.Emit([]byte(response))
 })
 // 设置断开连接回调
 client.OnDisconnect(func(id string) {
 log.Printf("Client %s disconnected", id)
 })
 // 设置错误处理回调
 client.OnError(func(id string, err error) {
 log.Printf("Error from client %s: %v", id, err)
 })
 // 建立WebSocket连接
 if err := client.Conn(ctx.Writer, ctx.Request); err != nil {
 log.Printf("Failed to establish WebSocket connection: %v", err)
 ctx.String(http.StatusInternalServerError, "Failed to establish WebSocket connection")
 return
 }
 })
 log.Fatal(app.Run(":8080"))
}

分布式模式

package main
import (
 "context"
 "log"
 "net"
 "net/http"
 
 "github.com/gin-gonic/gin"
 "github.com/go-redis/redis/v8"
 "github.com/grpc-ecosystem/go-grpc-middleware/recovery"
 "golang.org/x/sync/errgroup"
 "google.golang.org/grpc"
 
 websocket "github.com/jayecc/go-websocket"
 "github.com/jayecc/go-websocket/websocketpb"
)
func main() {
 serverGroup := errgroup.Group{}
 grpcAddr := ":8081"
 httpAddr := ":8082"
 grpcHost := websocket.IP().String() + grpcAddr
 
 // 创建Redis客户端
 redisClient := redis.NewClient(&redis.Options{
 Addr: "localhost:6379", 
 Password: "", 
 DB: 0,
 })
 
 // 创建Redis存储
 storage := websocket.NewRedisStorage(redisClient, "websocket")
 
 // 创建WebSocket Hub
 websocketHub := websocket.NewHubRun()
 defer websocketHub.Close()
 
 // 创建分布式WebSocket客户端
 websocketClient := websocket.NewDistClient(storage)
 // 启动 gRPC 服务器
 serverGroup.Go(func() error {
 lis, err := net.Listen("tcp", grpcAddr)
 if err != nil {
 return err
 }
 
 grpcServer := grpc.NewServer(
 grpc.UnaryInterceptor(grpcrecovery.UnaryServerInterceptor()),
 )
 
 // 注册 gRPC 服务
 websocketpb.RegisterWebsocketServer(grpcServer, websocket.NewDistServer(websocketHub))
 return grpcServer.Serve(lis)
 })
 // 启动 HTTP 服务器
 serverGroup.Go(func() error {
 gin.SetMode(gin.ReleaseMode)
 gin.DisableConsoleColor()
 app := gin.Default()
 
 // 注册WebSocket路由
 app.GET("/ws", func(ctx *gin.Context) {
 session := websocket.NewDistSession(websocketHub, storage, grpcHost, websocket.WithId("xxxx"))
 
 session.OnError(func(id string, err error) {
 log.Printf("OnError: %v\n", err)
 })
 
 session.OnEvent(func(conn *websocket.Client, messageType int, message []byte) {
 log.Printf("OnEvent: %s\n", string(message))
 // 广播消息到所有节点
 log.Println(websocketClient.Broadcast(context.Background(), []byte("grpc广播消息")))
 })
 
 session.OnConnect(func(conn *websocket.Client) {
 log.Printf("OnConnect: %s\n", conn.GetID())
 })
 
 session.OnDisconnect(func(id string) {
 log.Printf("OnDisconnect: %s\n", id)
 })
 
 if err := session.Conn(ctx.Writer, ctx.Request); err != nil {
 ctx.String(http.StatusInternalServerError, "Failed to establish WebSocket connection")
 return
 }
 })
 
 return app.Run(httpAddr)
 })
 
 log.Println(serverGroup.Wait())
}

API 参考

核心类型

Hub

Client

DistSession

DistClient

Storage

gRPC 服务

配置

Redis 配置

redisClient := redis.NewClient(&redis.Options{
 Addr: "localhost:6379",
 Password: "", 
 DB: 0,
})

存储前缀

storage := websocket.NewRedisStorage(redisClient, "websocket")

最佳实践

许可证