2023-04-28

ws推送优化

1、写多读少的情况，使用epoll 避免创建读的协程。

2、分片存储连接，降低锁的粒度，提高推送的并发。

3、写的时候使用缓存，批量定时写入，减少系统调用、协程的调度。

1、epoll

1、实现参考，https://github.com/eranyanay/1m-go-websockets 只适用在写多读少的情况下，并且读之后的业务耗时的操作不要太频繁。

https://github.com/eranyanay/1m-go-websockets/blob/master/3_optimize_ws_goroutines/server.go 这里读的时候要考虑到半消息的情况，https://studygolang.com/topics/13377 当收到websocket数据包不全的情况下，go net官方库当不可读的时候是会阻塞的。

我自己写的一个测试 github.com/gobwas/ws 提供了非常灵活的用法我们可以可以直接将一个websocket数据包转为字节。

func main() { {
     dialer := gws.DefaultDialer
     conn, _, _, err := dialer.Dial(context.Background(), "ws://10.18.13.129:8989/ws")
     if err != nil {
        log.Fatalln("err", err)
    }

     message := NewMessage(gws.OpText, []byte("abcdefjhigklmnopqrusasdflsdfasdl"))
     b, _ := message.ToBytes()
     conn.Write(b[:5])
     log.Println("发送前5个字节")
     time.Sleep(time.Second * 3)
     conn.Write(b[5:])
     log.Println("发送后面的字节")

     select {}
 }

   type Message struct {
    messageType gws.OpCode
    data        []byte
}

func NewMessage(code gws.OpCode, data []byte) Message {
    return Message{
        messageType: code,
        data:        data,
    }
}
func (m Message) ToBytes()([]byte, error) {
    var res bytes.Buffer
    frame := gws.NewFrame(m.messageType, true, m.data)
    if err := gws.WriteFrame(&res, frame); err != nil {
        return nil, err
    }
    return res.Bytes(), nil
}
2023/04/18 18:09:55 发送前5个字节
2023/04/18 18:09:58 发送后面的字节

func main() {
    engine := gin.New()
    engine.GET("/ws", Connect)
    engine.Run(":8989")
}
func Connect(c *gin.Context) {
    var httpUpgrade gws.HTTPUpgrader
    conn, _, _, err := httpUpgrade.Upgrade(c.Request, c.Writer)
    if err != nil {
       return
   }

    if err != nil {
       return
   }
     frame, err := gws.ReadFrame(conn)
       if err != nil {
          log.Println("err", err)
          return
      }
    log.Println(string(frame.Payload))
    //最终要等到数据全部到，go net官方库当不可读的时候是阻塞的。
   2023/04/18 18:09:58 abcdefjhigklmnopqrusasdflsdfasdl

}

解决方法：将数据读出来，如果io.ReadFull如果没有读完的，就回退到上次读出完整数据的位置，这里需要自己diy bufio.Reader ,当没有读出“半消息的时候缓存起来”。

2、使用写缓存

如果在小包特别多得情况下，直接发送会有很多系统调用,gmp模型，每一次系统调用go都会有一次调度会消耗cpu。在实时性要求不高的情况下，我们可以批量定时写入。

得益于github.com/gobwas/ws 提供的非常灵活的用法，我们可以暂时将我们要写的数据转为websocket数据包缓存起来，批量定时写入。此外提前encode websocket数据包，也可以避免每一个连接都去encode 一遍websocket数据包,提高效率。

 type Message struct {
    messageType gws.OpCode
    data        []byte
}
func (m Message) ToBytes()([]byte, error) {
    var res bytes.Buffer
    frame := gws.NewFrame(m.messageType, true, m.data)
    if err := gws.WriteFrame(&res, frame); err != nil {
        return nil, err
    }
    return res.Bytes(), nil
}
for {
    select {
       case data := <-conn.write:
       if _, err := conn.writeBuf.Write(data); err != nil {
          return
      }
      //writeBuf 为*bufio.Writer。
       case <-conn.writeRate.C:

       //写到连接中
       if conn.writeBuf.Available() > 0 {
          if err := conn.writeBuf.Flush(); err != nil {
             return
         }
      }

   }
}

3、分片存储连接

主要是参考https://github.com/owenliang/go-push

主要的思路就是分区map,连接分散到到不同的map中，提高锁的粒度和推送的并发读。同时如果是读多写少的情况可以使用sync.Map提高效率

实现功能

推送数据到公共频道

推送数据到所有连接

推送数据到私有频道

订阅操作

客户端请求数据的基本结构

{
    "code":1,
    "topic":"test",
    "data":"any"
}

code:表示操作的类型

1订阅公共频道。
2取消订阅公共频道。
3登录操作，需要你自己实现登录的逻辑。
4登出操作,需要你自己实现登出的逻辑，会退出你订阅的所有私有频道。
5订阅私有频道，需要登录。
6取消订阅私有频道，需要登录。

topic:你订阅的频道的名称

data:任意类型，保留字段，如果是登录操作你可以用这个字段传你的登录凭证信息。

后台推送

后台推送提供http接口和grpc接口

http 接口：使用grpc-gateway 提供

POST /v1/pushData

json格式数据

{
    "uid":"123456",
    "topic":"test",
    "data":"5pWw5o2u"
}

uid : 用户id当推送给私有频道指定用户的id，否则为空。

topic : 频道的名称，当你想推送给所有的时候为空。

data : 具体的推送的内容，推送的数据要经过base64编码

grpc接口 github.com/luxun9527/gpush/proto

protobuf 格式

message Data{
  string uid =1;
  string topic=2;
  bytes data=3;
}

代码

具体代码仓库地址 https://github.com/luxun9527/gpush 练习时长还没到两年半，还有很多不完善的地方主要是提供一个实现的思路如果对您有帮助，请帮我点一下star。