liwei_3d/go_dreamfactory
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

上传会话对象消息队列优化

Browse Source
This commit is contained in:
liwei1dao 2022-06-28 17:09:32 +08:00
parent 6098f3edda
commit 50bb5a92a8
9 changed files with 42 additions and 45 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
comm/core.go
View File

@ -16,7 +16,7 @@ import (
type ISC_GateRouteComp interface { type ISC_GateRouteComp interface {
core.IServiceComp core.IServiceComp
ReceiveMsg(ctx context.Context, args *pb.AgentMessage, reply *pb.RPCMessageReply) error ReceiveMsg(ctx context.Context, args *pb.AgentMessage, reply *pb.RPCMessageReply) error
RegisterRoute(methodName string, comp reflect.Value, msg reflect.Type, reply reflect.Type, check, handle reflect.Method) RegisterRoute(methodName string, comp reflect.Value, msg reflect.Type, check, handle reflect.Method)
} }
//游戏类资源类型 //游戏类资源类型

2
comm/usersession.go
View File

@ -22,6 +22,7 @@ func NewUserSession(service base.IRPCXService, ip, sessionId, gatewayServiceId s
SessionId: sessionId, SessionId: sessionId,
GatewayServiceId: gatewayServiceId, GatewayServiceId: gatewayServiceId,
UserId: uid, UserId: uid,
msgqueue: make([]*pb.UserMessage, 0),
service: service, service: service,
} }
} }
@ -32,6 +33,7 @@ type UserSession struct {
GatewayServiceId string //用户所在网关服务 GatewayServiceId string //用户所在网关服务
UserId string UserId string
service base.IRPCXService service base.IRPCXService
msgqueue []*pb.UserMessage
} }
//获取用户的会话id //获取用户的会话id

11
modules/comp_gate.go
View File

@ -77,7 +77,7 @@ func (this *MCompGate) reflectionRouteHandle(typ reflect.Type, method reflect.Me
if method.PkgPath != "" { if method.PkgPath != "" {
return return
} }
if mtype.NumIn() != 5 { if mtype.NumIn() != 4 {
return return
} }
ctxType := mtype.In(1) ctxType := mtype.In(1)
@ -95,13 +95,6 @@ func (this *MCompGate) reflectionRouteHandle(typ reflect.Type, method reflect.Me
if !this.isExportedOrBuiltinType(agrType) { if !this.isExportedOrBuiltinType(agrType) {
return return
} }
replyType := mtype.In(4)
if replyType.Kind() != reflect.Ptr {
return
}
if !this.isExportedOrBuiltinType(replyType) {
return
}
if mtype.NumOut() != 1 { if mtype.NumOut() != 1 {
return return
@ -118,7 +111,7 @@ func (this *MCompGate) reflectionRouteHandle(typ reflect.Type, method reflect.Me
return return
} }
if err := this.reflectionRouteCheck(check, agrType); err == nil { if err := this.reflectionRouteCheck(check, agrType); err == nil {
this.scomp.RegisterRoute(fmt.Sprintf("%s.%s", this.module.GetType(), strings.ToLower(mname)), reflect.ValueOf(this.comp), agrType, replyType, check, method) this.scomp.RegisterRoute(fmt.Sprintf("%s.%s", this.module.GetType(), strings.ToLower(mname)), reflect.ValueOf(this.comp), agrType, check, method)
} else { } else {
log.Panicf("反射注册用户处理函数错误 [%s-%s]校验函数格式异常:%v", this.module.GetType(), mname, err) log.Panicf("反射注册用户处理函数错误 [%s-%s]校验函数格式异常:%v", this.module.GetType(), mname, err)
return return

0
modules/comp_configure.go → modules/compconfigure.go
View File

15
modules/equipment/api_getlist.go
View File

@ -13,21 +13,20 @@ func (this *apiComp) Getlist_Check(session comm.IUserSession, req *pb.Equipment_
} }
///获取用户装备列表 ///获取用户装备列表
func (this *apiComp) Getlist(session comm.IUserSession, agrs map[string]interface{}, req *pb.Equipment_GetList_Req, resp *pb.Equipment_GetList_Resp) (code pb.ErrorCode) { func (this *apiComp) Getlist(session comm.IUserSession, agrs map[string]interface{}, req *pb.Equipment_GetList_Req) (code pb.ErrorCode) {
var ( var (
err error err error
items []*pb.DB_Equipment items []*pb.DB_Equipment
) )
// defer func() { defer func() {
// if code == pb.ErrorCode_Success { if code == pb.ErrorCode_Success {
// session.SendMsg(string(this.module.GetType()), "", &pb.Equipment_GetList_Resp{Equipments: items}) session.SendMsg(string(this.module.GetType()), "", &pb.Equipment_GetList_Resp{Equipments: items})
// } }
// }() }()
if items, err = this.module.modelEquipment.QueryUserEquipments(session.GetUserId()); err != nil { if items, err = this.module.modelequipment.QueryUserEquipments(session.GetUserId()); err != nil {
log.Errorf("QueryUserPackReq err:%v", err) log.Errorf("QueryUserPackReq err:%v", err)
code = pb.ErrorCode_CacheReadError code = pb.ErrorCode_CacheReadError
return return
} }
resp.Equipments = items
return return
} }

16
modules/gateway/agent.go
View File

@ -208,7 +208,7 @@ func (this *Agent) Close() {
} }
//分发用户消息 //分发用户消息
func (this *Agent) messageDistribution(msg *pb.UserMessage) error { func (this *Agent) messageDistribution(msg *pb.UserMessage) (err error) {
reply := &pb.RPCMessageReply{} reply := &pb.RPCMessageReply{}
log.Debugf("agent:%s uId:%s MessageDistribution msg:%s.%s", this.sessionId, this.uId, msg.MainType, msg.SubType) log.Debugf("agent:%s uId:%s MessageDistribution msg:%s.%s", this.sessionId, this.uId, msg.MainType, msg.SubType)
servicePath := comm.Service_Worker servicePath := comm.Service_Worker
@ -219,7 +219,7 @@ func (this *Agent) messageDistribution(msg *pb.UserMessage) error {
servicePath = fmt.Sprintf("%s/%s", comm.Service_Worker, this.wId) servicePath = fmt.Sprintf("%s/%s", comm.Service_Worker, this.wId)
} }
} }
if err := this.gateway.Service().RpcCall(context.Background(), servicePath, string(comm.Rpc_GatewayRoute), &pb.AgentMessage{ if err = this.gateway.Service().RpcCall(context.Background(), servicePath, string(comm.Rpc_GatewayRoute), &pb.AgentMessage{
Ip: this.IP(), Ip: this.IP(),
UserSessionId: this.sessionId, UserSessionId: this.sessionId,
UserId: this.uId, UserId: this.uId,
@ -229,16 +229,22 @@ func (this *Agent) messageDistribution(msg *pb.UserMessage) error {
Message: msg.Data, Message: msg.Data,
}, reply); err != nil { }, reply); err != nil {
log.Errorf("agent:%s uId:%s MessageDistribution err:%v", this.sessionId, this.uId, err) log.Errorf("agent:%s uId:%s MessageDistribution err:%v", this.sessionId, this.uId, err)
return err return
} }
if reply.Code != pb.ErrorCode_Success { if reply.Code != pb.ErrorCode_Success {
data, _ := anypb.New(&pb.ErrorNotify{ReqMainType: msg.MainType, ReqSubType: msg.SubType, Code: pb.ErrorCode(reply.Code.Number())}) data, _ := anypb.New(&pb.ErrorNotify{ReqMainType: msg.MainType, ReqSubType: msg.SubType, Code: pb.ErrorCode(reply.Code.Number())})
err := this.WriteMsg(&pb.UserMessage{ err = this.WriteMsg(&pb.UserMessage{
MainType: comm.MainType_Notify, MainType: comm.MainType_Notify,
SubType: comm.SubType_ErrorNotify, SubType: comm.SubType_ErrorNotify,
Data: data, Data: data,
}) })
return err return
} else {
for _, v := range reply.Reply {
if err = this.WriteMsg(v); err != nil {
return
}
}
} }
return nil return nil
} }

12
pb/comm.pb.go
View File

@ -252,10 +252,10 @@ type RPCMessageReply struct {
sizeCache protoimpl.SizeCache sizeCache protoimpl.SizeCache
unknownFields protoimpl.UnknownFields unknownFields protoimpl.UnknownFields
Code ErrorCode `protobuf:"varint,1,opt,name=Code,proto3,enum=ErrorCode" json:"Code"` Code ErrorCode `protobuf:"varint,1,opt,name=Code,proto3,enum=ErrorCode" json:"Code"`
ErrorMessage string `protobuf:"bytes,2,opt,name=ErrorMessage,proto3" json:"ErrorMessage"` ErrorMessage string `protobuf:"bytes,2,opt,name=ErrorMessage,proto3" json:"ErrorMessage"`
ErrorData string `protobuf:"bytes,3,opt,name=ErrorData,proto3" json:"ErrorData"` ErrorData string `protobuf:"bytes,3,opt,name=ErrorData,proto3" json:"ErrorData"`
Reply *UserMessage `protobuf:"bytes,4,opt,name=Reply,proto3" json:"Reply"` Reply []*UserMessage `protobuf:"bytes,4,rep,name=Reply,proto3" json:"Reply"`
} }
func (x *RPCMessageReply) Reset() { func (x *RPCMessageReply) Reset() {
@ -311,7 +311,7 @@ func (x *RPCMessageReply) GetErrorData() string {
return "" return ""
} }
func (x *RPCMessageReply) GetReply() *UserMessage { func (x *RPCMessageReply) GetReply() []*UserMessage {
if x != nil { if x != nil {
return x.Reply return x.Reply
} }
@ -771,7 +771,7 @@ var file_comm_proto_rawDesc = []byte{
0x20, 0x01, 0x28, 0x09, 0x52, 0x0c, 0x45, 0x72, 0x72, 0x6f, 0x72, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0c, 0x45, 0x72, 0x72, 0x6f, 0x72, 0x4d, 0x65, 0x73, 0x73, 0x61,
0x67, 0x65, 0x12, 0x1c, 0x0a, 0x09, 0x45, 0x72, 0x72, 0x6f, 0x72, 0x44, 0x61, 0x74, 0x61, 0x18, 0x67, 0x65, 0x12, 0x1c, 0x0a, 0x09, 0x45, 0x72, 0x72, 0x6f, 0x72, 0x44, 0x61, 0x74, 0x61, 0x18,
0x03, 0x20, 0x01, 0x28, 0x09, 0x52, 0x09, 0x45, 0x72, 0x72, 0x6f, 0x72, 0x44, 0x61, 0x74, 0x61, 0x03, 0x20, 0x01, 0x28, 0x09, 0x52, 0x09, 0x45, 0x72, 0x72, 0x6f, 0x72, 0x44, 0x61, 0x74, 0x61,
0x12, 0x22, 0x0a, 0x05, 0x52, 0x65, 0x70, 0x6c, 0x79, 0x18, 0x04, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x12, 0x22, 0x0a, 0x05, 0x52, 0x65, 0x70, 0x6c, 0x79, 0x18, 0x04, 0x20, 0x03, 0x28, 0x0b, 0x32,
0x0c, 0x2e, 0x55, 0x73, 0x65, 0x72, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x52, 0x05, 0x52, 0x0c, 0x2e, 0x55, 0x73, 0x65, 0x72, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x52, 0x05, 0x52,
0x65, 0x70, 0x6c, 0x79, 0x22, 0x69, 0x0a, 0x0d, 0x41, 0x67, 0x65, 0x6e, 0x74, 0x42, 0x75, 0x69, 0x65, 0x70, 0x6c, 0x79, 0x22, 0x69, 0x0a, 0x0d, 0x41, 0x67, 0x65, 0x6e, 0x74, 0x42, 0x75, 0x69,
0x6c, 0x64, 0x52, 0x65, 0x71, 0x12, 0x24, 0x0a, 0x0d, 0x55, 0x73, 0x65, 0x72, 0x53, 0x65, 0x73, 0x6c, 0x64, 0x52, 0x65, 0x71, 0x12, 0x24, 0x0a, 0x0d, 0x55, 0x73, 0x65, 0x72, 0x53, 0x65, 0x73,

2
pb/proto/comm.proto
View File

@ -28,7 +28,7 @@ message RPCMessageReply {
ErrorCode Code = 1; ErrorCode Code = 1;
string ErrorMessage = 2; string ErrorMessage = 2;
string ErrorData = 3; string ErrorData = 3;
UserMessage Reply = 4; repeated UserMessage Reply = 4;
} }
//用户代理绑定Uid请求 //用户代理绑定Uid请求

27
services/comp_gateroute.go
View File

@ -33,11 +33,10 @@ func NewGateRouteComp() comm.ISC_GateRouteComp {
//用户协议处理函数注册的反射对象 //用户协议处理函数注册的反射对象
type msghandle struct { type msghandle struct {
rcvr reflect.Value rcvr reflect.Value
msgType reflect.Type //消息请求类型 msgType reflect.Type //消息请求类型
replyType reflect.Type //消息回应类型 check reflect.Method //校验函数
check reflect.Method //校验函数 handle reflect.Method //处理函数
handle reflect.Method //处理函数
} }
//服务网关组件 //服务网关组件
@ -70,7 +69,7 @@ func (this *SCompGateRoute) Start() (err error) {
} }
//业务模块注册用户消息处理路由 //业务模块注册用户消息处理路由
func (this *SCompGateRoute) RegisterRoute(methodName string, comp reflect.Value, msg reflect.Type, replyType reflect.Type, check, handele reflect.Method) { func (this *SCompGateRoute) RegisterRoute(methodName string, comp reflect.Value, msg reflect.Type, check, handele reflect.Method) {
log.Debugf("注册用户路由【%s】", methodName) log.Debugf("注册用户路由【%s】", methodName)
this.mrlock.RLock() this.mrlock.RLock()
_, ok := this.msghandles[methodName] _, ok := this.msghandles[methodName]
@ -81,11 +80,10 @@ func (this *SCompGateRoute) RegisterRoute(methodName string, comp reflect.Value,
} }
this.mrlock.Lock() this.mrlock.Lock()
this.msghandles[methodName] = &msghandle{ this.msghandles[methodName] = &msghandle{
rcvr: comp, rcvr: comp,
msgType: msg, msgType: msg,
replyType: replyType, check: check,
check: check, handle: handele,
handle: handele,
} }
this.mrlock.Unlock() this.mrlock.Unlock()
} }
@ -117,9 +115,6 @@ func (this *SCompGateRoute) ReceiveMsg(ctx context.Context, args *pb.AgentMessag
return err return err
} }
//序列化用户消息对象
replymsg := reflect.New(msghandle.replyType.Elem()).Interface()
//调用校验接口 //调用校验接口
checkreturn := msghandle.check.Func.Call([]reflect.Value{msghandle.rcvr, reflect.ValueOf(session), reflect.ValueOf(msg)}) checkreturn := msghandle.check.Func.Call([]reflect.Value{msghandle.rcvr, reflect.ValueOf(session), reflect.ValueOf(msg)})
//读取校验结果 有错误直接返回错误码给用户 //读取校验结果 有错误直接返回错误码给用户
@ -141,7 +136,7 @@ func (this *SCompGateRoute) ReceiveMsg(ctx context.Context, args *pb.AgentMessag
//校验结果成功 处理临时数据转移 //校验结果成功 处理临时数据转移
result := checkreturn[0].Interface().(map[string]interface{}) result := checkreturn[0].Interface().(map[string]interface{})
//调用用户处理函数 //调用用户处理函数
handlereturn := msghandle.handle.Func.Call([]reflect.Value{msghandle.rcvr, reflect.ValueOf(session), reflect.ValueOf(result), reflect.ValueOf(msg), reflect.ValueOf(replymsg)}) handlereturn := msghandle.handle.Func.Call([]reflect.Value{msghandle.rcvr, reflect.ValueOf(session), reflect.ValueOf(result), reflect.ValueOf(msg)})
errcode := pb.ErrorCode(handlereturn[0].Int()) errcode := pb.ErrorCode(handlereturn[0].Int())
if errcode != pb.ErrorCode_Success { //处理返货错误码 返回用户错误信息 if errcode != pb.ErrorCode_Success { //处理返货错误码 返回用户错误信息
log.Errorf("HandleUserMsg:%s msg:%v code:%d", method, msg, code) log.Errorf("HandleUserMsg:%s msg:%v code:%d", method, msg, code)
@ -149,6 +144,7 @@ func (this *SCompGateRoute) ReceiveMsg(ctx context.Context, args *pb.AgentMessag
reply.ErrorMessage = pb.GetErrorCodeMsg(errcode) reply.ErrorMessage = pb.GetErrorCodeMsg(errcode)
return nil return nil
} }
} else { //未找到消息处理函数 } else { //未找到消息处理函数
reply.Code = pb.ErrorCode_ReqParameterError reply.Code = pb.ErrorCode_ReqParameterError
} }
@ -158,5 +154,6 @@ func (this *SCompGateRoute) ReceiveMsg(ctx context.Context, args *pb.AgentMessag
//RPC_NoticeUserClose 接收用户离线通知 //RPC_NoticeUserClose 接收用户离线通知
func (this *SCompGateRoute) NoticeUserClose(ctx context.Context, args *pb.NoticeUserCloseReq, reply *pb.RPCMessageReply) error { func (this *SCompGateRoute) NoticeUserClose(ctx context.Context, args *pb.NoticeUserCloseReq, reply *pb.RPCMessageReply) error {
event.TriggerEvent(comm.Event_UserOffline, args.UserId) event.TriggerEvent(comm.Event_UserOffline, args.UserId)
return nil return nil
} }