mmm-go / gocomm · 提交

转到一个项目

作者 yangfu
提交 a0c1b8efbd411f24dbc5018ba0097360101c2f68 1 个父辈 a6c6e89a

ws 链路管理修改

内嵌 并排对比
正在显示 4 个修改的文件 包含 261 行增加 和 43 行删除
package websocket
import (
"encoding/json"
"fmt"
"github.com/astaxie/beego"
"github.com/gorilla/websocket"
"gitlab.fjmaimaimai.com/mmm-go/gocomm/pkg/mybeego"
"reflect"
"sync"
)
var (
ErrorNotFound =fmt.Errorf("conn not exist")
)
var DefaultConnmgrs Connmgrs
//初始化websocket链路管理器数量,通过uid取模放入不同的管理器
func InitWebsocketConnmgrs(mgrsSize int){
connmgrs :=make(map[int]IConnmgr,mgrsSize)
for i:=0;i<mgrsSize;i++{
connmgrs[i] = NewMemoryConnmgr()
}
DefaultConnmgrs = Connmgrs(connmgrs)
}
type IConnmgr interface {
Put(key,value interface{})(bool)
Remove(key interface{}) error
Get(key interface{})(value interface{},err error)
}
//连接管理器
type Connmgrs map[int]IConnmgr
//将连接从指定连接管理器中移除
func (m Connmgrs)Remove(connmgrId int,key interface{})(err error){
//删除特定链接管理的连接
if mgr,ok:= m[connmgrId];ok{
err =mgr.Remove(key)
if err!=nil{
return err
}
}
return
}
//将连接装载到指定 连接管理器
func (m Connmgrs)Put(connmgrId int,key,value interface{})(result bool){
result =false
if mgr,ok:= m[connmgrId];ok{
result =mgr.Put(key,value)
if !result{
return
}
}
return
}
type MemoryConnmgr struct {
mutex sync.RWMutex
Connections *JMap //conn
Clients *JMap // key=uid(int64) value(*WebsocketConnection)
//rooms //房间
}
func NewMemoryConnmgr()*MemoryConnmgr{
keyType := reflect.TypeOf(&websocket.Conn{})
valueType := reflect.TypeOf(&WebsocketConnection{})
return &MemoryConnmgr{
Connections:NewJMap(keyType, valueType),
Clients:NewJMap(reflect.TypeOf("1:1"), valueType),
}
}
func(m *MemoryConnmgr)Put(key,value interface{})(result bool){
m.mutex.Lock()
defer m.mutex.Unlock()
if c,ok :=value.(*WebsocketConnection);ok{
idKey := fmt.Sprintf("%d:%d", c.Uid, c.AppId)
return m.Connections.Put(c.Conn,c) && m.Clients.Put(idKey,c)
}
return false
}
func(m *MemoryConnmgr)Get(key interface{})(value interface{},err error){
var ok bool
switch reflect.TypeOf(key).Kind() {
case reflect.String:
if value,ok = m.Clients.Get(key);!ok{
err = ErrorNotFound
return
}
case reflect.Struct:
if value,ok = m.Connections.Get(key);!ok{
err = ErrorNotFound
return
}
default:
err = ErrorNotFound
}
return
}
func(m *MemoryConnmgr)Remove(key interface{})(err error){
m.mutex.Lock()
defer m.mutex.Unlock()
if c,ok :=key.(*WebsocketConnection);ok{
key := fmt.Sprintf("%d:%d", c.Uid, c.AppId)
m.Connections.Remove(c.Conn)
m.Clients.Remove(key)
}
return
}
//发送数据
func SendDataByConnmgr(uid int64, appId int, sendMsg interface{}) bool {
if sendMsg == nil || uid < 1 || appId < 1 {
return false
}
var mgrId int =int(uid % int64(len(DefaultConnmgrs)))
connmgr,ok :=(DefaultConnmgrs)[mgrId]
if !ok{
return false
}
msg := &mybeego.Message{
Errno: 0,
Errmsg: mybeego.NewMessage(0).Errmsg,
Data: sendMsg,
}
msgByte, err := json.Marshal(msg)
if err != nil {
beego.Error(err)
return false
}
key := fmt.Sprintf("%d:%d", uid, appId)
if connI,err := connmgr.Get(key); err==nil {
if conn, ok := connI.(*WebsocketConnection); ok {
conn.Send(string(msgByte))
return true
}
}
return false
}
... ...
package websocket
import (
"github.com/gorilla/websocket"
"testing"
)
func TestNewMemoryConnmgr(t *testing.T){
connmgr :=NewMemoryConnmgr()
num:=100
var listConn []*WebsocketConnection
for i:=0;i<num;i++{
listConn = append(listConn,&WebsocketConnection{
Uid:int64(i),
AppId:1,
Conn:&websocket.Conn{},
})
}
for i:=range listConn{
connmgr.Put(listConn[i],listConn[i])
}
if connmgr.Clients.Size() !=num && connmgr.Connections.Size()!=num{
t.Fatal("size error :",connmgr.Clients.Size(),connmgr.Connections.Size())
}
for i:=range listConn{
connmgr.Remove(listConn[i])
}
if connmgr.Clients.Size() !=0 && connmgr.Connections.Size()!=0{
t.Fatal("size error :",connmgr.Clients.Size(),connmgr.Connections.Size())
}
}
func BenchmarkNewMemoryConnmgr(b *testing.B) {
connmgr :=NewMemoryConnmgr()
var listConn []*WebsocketConnection
for i:=0;i<b.N;i++{
listConn = append(listConn,&WebsocketConnection{
Uid:int64(i),
AppId:1,
Conn:&websocket.Conn{},
})
}
b.ResetTimer()
for i:=0;i<b.N;i++{
connmgr.Put(listConn[i],listConn[i])
}
for i:=0;i<b.N;i++{
connmgr.Remove(listConn[i])
}
if connmgr.Clients.Size() !=0 && connmgr.Connections.Size()!=0{
b.Fatal("size error :",connmgr.Clients.Size(),connmgr.Connections.Size())
}
}
... ...
... ... @@ -17,12 +17,12 @@ const (
Connected
)
func init() {
keyType := reflect.TypeOf(&websocket.Conn{})
valueType := reflect.TypeOf(&WebsocketConnection{})
Connections = NewJMap(keyType, valueType)
Clients = NewJMap(reflect.TypeOf("1:1"), valueType)
}
//func init() {
// keyType := reflect.TypeOf(&websocket.Conn{})
// valueType := reflect.TypeOf(&WebsocketConnection{})
// Connections = NewJMap(keyType, valueType)
// Clients = NewJMap(reflect.TypeOf("1:1"), valueType)
//}
type ReceiveHandler (func([]byte) *mybeego.Message)
... ... @@ -35,6 +35,7 @@ type WebsocketConnection struct {
State ConnState
OnReceive ReceiveHandler
OnceClose sync.Once
Connmgrg *Connmgrs
}
func NewWebsocketConnection(conn *websocket.Conn,head *mybeego.RequestHead,recv ReceiveHandler)*WebsocketConnection{
... ... @@ -46,14 +47,15 @@ func NewWebsocketConnection(conn *websocket.Conn,head *mybeego.RequestHead,recv
Wchan: make(chan string, 10),
State: Connected,
OnReceive: recv,
Connmgrg:&DefaultConnmgrs,
}
}
//声明了两个cliets 管理 一个通过uid 一个通过conn管理
// key(*websocket.Conn) value(*WebsocketConnection)
var Connections *JMap
//var Connections *JMap
// key=uid(int64) value(*WebsocketConnection)
var Clients *JMap
//var Clients *JMap
type JMap struct {
sync.RWMutex
... ... @@ -107,13 +109,12 @@ func (this *JMap) Put(k interface{}, v interface{}) bool {
if !this.acceptable(k, v) {
return false
}
if connI, ok := Clients.Get(k); ok {
if connI, ok := this.m[k]; ok {
beego.Debug("key:", k, "已经连接,先剔除下线")
if conn, ok := connI.(*WebsocketConnection); ok {
//conn.Conn.WriteMessage(websocket.TextMessage, []byte("您的帐号在其它地方登录,您被剔除下线"))
conn.Close()
}
}
this.Lock()
this.m[k] = v
... ... @@ -152,14 +153,18 @@ func (this *JMap) Contains(k interface{}) bool {
func (c *WebsocketConnection) Serve() {
c.State = Connected
Connections.Put(c.Conn, c)
key := fmt.Sprintf("%d:%d", c.Uid, c.AppId)
Clients.Put(key, c)
//Connections.Put(c.Conn, c)
//key := fmt.Sprintf("%d:%d", c.Uid, c.AppId)
//Clients.Put(key, c)
c.Connmgrg.Put(c.GetConnmgrId(c.Uid),c,c)
go doWrite(c)
doRead(c)
}
func(c *WebsocketConnection)GetConnmgrId(uid int64)int{
return (int)(uid % int64(len(*c.Connmgrg)))
}
func (c *WebsocketConnection) Send(msg string) {
//panic("panic in websocket.send...")
c.Wchan <- msg
... ... @@ -167,14 +172,15 @@ func (c *WebsocketConnection) Send(msg string) {
func (c *WebsocketConnection) Close() {
c.OnceClose.Do(func(){
beego.Info("ws:close----uid:", c.Uid, "appid:", c.AppId, "state:", c.State)
beego.Info("ws:close----uid:", c.Uid, "appid:", c.AppId, "state:", c.State ,"connmgr-id:",c.GetConnmgrId(c.Uid))
if c.State == Disconnected {
return
}
Connections.Remove(c.Conn)
key := fmt.Sprintf("%d:%d", c.Uid, c.AppId)
Clients.Remove(key)
//Connections.Remove(c.Conn)
//key := fmt.Sprintf("%d:%d", c.Uid, c.AppId)
//Clients.Remove(key)
c.Connmgrg.Remove(c.GetConnmgrId(c.Uid),c)
c.State = Disconnected
close(c.Echan)
... ... @@ -236,27 +242,27 @@ func doWrite(c *WebsocketConnection) {
}
}
func SendDataByWs(uid int64, appId int, sendMsg interface{}) bool {
if sendMsg == nil || uid < 1 || appId < 1 {
return false
}
msg := &mybeego.Message{
Errno: 0,
Errmsg: mybeego.NewMessage(0).Errmsg,
Data: sendMsg,
}
msgByte, err := json.Marshal(msg)
if err != nil {
beego.Error(err)
return false
}
key := fmt.Sprintf("%d:%d", uid, appId)
if connI, ok := Clients.Get(key); ok {
beego.Debug(ok)
if conn, ok := connI.(*WebsocketConnection); ok {
conn.Send(string(msgByte))
return true
}
}
return false
}
//func SendDataByWs(uid int64, appId int, sendMsg interface{}) bool {
// if sendMsg == nil || uid < 1 || appId < 1 {
// return false
// }
// msg := &mybeego.Message{
// Errno: 0,
// Errmsg: mybeego.NewMessage(0).Errmsg,
// Data: sendMsg,
// }
// msgByte, err := json.Marshal(msg)
// if err != nil {
// beego.Error(err)
// return false
// }
// key := fmt.Sprintf("%d:%d", uid, appId)
// if connI, ok := Clients.Get(key); ok {
// beego.Debug(ok)
// if conn, ok := connI.(*WebsocketConnection); ok {
// conn.Send(string(msgByte))
// return true
// }
// }
// return false
//}
... ...
... ... @@ -2,20 +2,37 @@ package websocket
import (
"github.com/gorilla/websocket"
"gitlab.fjmaimaimai.com/mmm-go/gocomm/pkg/log"
"gitlab.fjmaimaimai.com/mmm-go/gocomm/pkg/mybeego"
"html/template"
"log"
"net/http"
"strconv"
"testing"
"time"
)
func Test_RunWebSocket(t *testing.T) {
//InitWebsocketConnmgrs(10)
//http.HandleFunc("/join",join)
//http.HandleFunc("/",home)
//
//go TimerWork()
//log.Fatal(http.ListenAndServe(":8080",nil))
}
func TimerWork(){
t :=time.NewTicker(10*time.Second)
ch :=make(chan int,1)
for {
select {
case <-t.C:
//log.Info(DefaultConnmgrs[0])
SendDataByConnmgr(0,0,time.Now())
}
}
<-ch
}
var upgrader = websocket.Upgrader{}
func join(w http.ResponseWriter, r *http.Request) {
... ...