From 09d742e74039aba94f3af15dc7fd0317568986f8 Mon Sep 17 00:00:00 2001 From: liuxiaobo <1224730913@qq.com> Date: Thu, 29 May 2025 11:49:24 +0800 Subject: [PATCH] =?UTF-8?q?=E5=B0=86=E5=86=85=E9=83=A8=E6=B6=88=E6=81=AF?= =?UTF-8?q?=E9=9B=86=E6=88=90=E5=88=B0=E6=A1=86=E6=9E=B6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- internalPb/gene_proto.sh | 7 + internalPb/internal.proto | 28 ++++ ipb/helper.go | 39 ++++++ ipb/internal.pb.go | 280 ++++++++++++++++++++++++++++++++++++++ service/baseService.go | 9 +- service/iservice.go | 9 +- service/natsService.go | 55 +++++--- service/topic.go | 14 +- 8 files changed, 415 insertions(+), 26 deletions(-) create mode 100644 internalPb/gene_proto.sh create mode 100644 internalPb/internal.proto create mode 100644 ipb/helper.go create mode 100644 ipb/internal.pb.go diff --git a/internalPb/gene_proto.sh b/internalPb/gene_proto.sh new file mode 100644 index 0000000..1690a35 --- /dev/null +++ b/internalPb/gene_proto.sh @@ -0,0 +1,7 @@ +#!/bin/bash + + + +protoc -I=. --proto_path=./ --go_out=../../ *.proto + +echo "success" diff --git a/internalPb/internal.proto b/internalPb/internal.proto new file mode 100644 index 0000000..bb4edf8 --- /dev/null +++ b/internalPb/internal.proto @@ -0,0 +1,28 @@ +syntax = "proto3"; +package ipb; +option go_package = "fox/ipb"; + +enum MsgId +{ + Unknown = 0; + Internal = -1; // 内部消息id +} + +enum MsgType +{ + NormalMsg = 0; + RpcMsg = 1; +} + +message InternalMsg +{ + string service_name = 1; // 该服务类型下的具体的服务节点名，需要保证该消息是该服务节点发的。否则可能会导致客户端出现路由错误 + uint32 conn_id = 2; // 刚登陆时没有user_id，只有conn_id + int64 user_id = 3; // 玩家id + int32 msg_id = 4; // 消息id + bytes msg = 5; // 消息 + MsgType type = 6; // 消息类型 + int32 ret_rpc_msg_id = 7; // 依赖本字段把数据回传给service_name节点的rpc processor去处理 +} + + diff --git a/ipb/helper.go b/ipb/helper.go new file mode 100644 index 0000000..6805ad6 --- /dev/null +++ b/ipb/helper.go @@ -0,0 +1,39 @@ +package ipb + +import "sync/atomic" + +func MakeMsg(serviceName string, connId uint32, userId int64, msgId int32, msg []byte) *InternalMsg { + return &InternalMsg{ + ServiceName: serviceName, + ConnId: connId, + UserId: userId, + MsgId: msgId, + Msg: msg, + } +} + +func MakeRpcMsg(serviceName string, connId uint32, userId int64, msgId int32, msg []byte) *InternalMsg { + return &InternalMsg{ + ServiceName: serviceName, + ConnId: connId, + UserId: userId, + MsgId: msgId, + Msg: msg, + Type: MsgType_RpcMsg, + RetRpcMsgId: genRpcId(), + } +} + +const ( + rpcBeginId = -500000 + rpcEndId = -100000 +) + +var rpcId int32 + +func genRpcId() int32 { + if atomic.LoadInt32(&rpcId) > rpcEndId { + atomic.StoreInt32(&rpcId, rpcBeginId) + } + return atomic.AddInt32(&rpcId, 1) +} diff --git a/ipb/internal.pb.go b/ipb/internal.pb.go new file mode 100644 index 0000000..e99ffc9 --- /dev/null +++ b/ipb/internal.pb.go @@ -0,0 +1,280 @@ +// Code generated by protoc-gen-go. DO NOT EDIT. +// versions: +// protoc-gen-go v1.36.6 +// protoc v6.31.0 +// source: internal.proto + +package ipb + +import ( + protoreflect "google.golang.org/protobuf/reflect/protoreflect" + protoimpl "google.golang.org/protobuf/runtime/protoimpl" + reflect "reflect" + sync "sync" + unsafe "unsafe" +) + +const ( + // Verify that this generated code is sufficiently up-to-date. + _ = protoimpl.EnforceVersion(20 - protoimpl.MinVersion) + // Verify that runtime/protoimpl is sufficiently up-to-date. + _ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20) +) + +type MsgId int32 + +const ( + MsgId_Unknown MsgId = 0 + MsgId_Internal MsgId = -1 // 内部消息id +) + +// Enum value maps for MsgId. +var ( + MsgId_name = map[int32]string{ + 0: "Unknown", + -1: "Internal", + } + MsgId_value = map[string]int32{ + "Unknown": 0, + "Internal": -1, + } +) + +func (x MsgId) Enum() *MsgId { + p := new(MsgId) + *p = x + return p +} + +func (x MsgId) String() string { + return protoimpl.X.EnumStringOf(x.Descriptor(), protoreflect.EnumNumber(x)) +} + +func (MsgId) Descriptor() protoreflect.EnumDescriptor { + return file_internal_proto_enumTypes[0].Descriptor() +} + +func (MsgId) Type() protoreflect.EnumType { + return &file_internal_proto_enumTypes[0] +} + +func (x MsgId) Number() protoreflect.EnumNumber { + return protoreflect.EnumNumber(x) +} + +// Deprecated: Use MsgId.Descriptor instead. +func (MsgId) EnumDescriptor() ([]byte, []int) { + return file_internal_proto_rawDescGZIP(), []int{0} +} + +type MsgType int32 + +const ( + MsgType_NormalMsg MsgType = 0 + MsgType_RpcMsg MsgType = 1 +) + +// Enum value maps for MsgType. +var ( + MsgType_name = map[int32]string{ + 0: "NormalMsg", + 1: "RpcMsg", + } + MsgType_value = map[string]int32{ + "NormalMsg": 0, + "RpcMsg": 1, + } +) + +func (x MsgType) Enum() *MsgType { + p := new(MsgType) + *p = x + return p +} + +func (x MsgType) String() string { + return protoimpl.X.EnumStringOf(x.Descriptor(), protoreflect.EnumNumber(x)) +} + +func (MsgType) Descriptor() protoreflect.EnumDescriptor { + return file_internal_proto_enumTypes[1].Descriptor() +} + +func (MsgType) Type() protoreflect.EnumType { + return &file_internal_proto_enumTypes[1] +} + +func (x MsgType) Number() protoreflect.EnumNumber { + return protoreflect.EnumNumber(x) +} + +// Deprecated: Use MsgType.Descriptor instead. +func (MsgType) EnumDescriptor() ([]byte, []int) { + return file_internal_proto_rawDescGZIP(), []int{1} +} + +type InternalMsg struct { + state protoimpl.MessageState `protogen:"open.v1"` + ServiceName string `protobuf:"bytes,1,opt,name=service_name,json=serviceName,proto3" json:"service_name,omitempty"` // 该服务类型下的具体的服务节点名，需要保证该消息是该服务节点发的。否则可能会导致客户端出现路由错误 + ConnId uint32 `protobuf:"varint,2,opt,name=conn_id,json=connId,proto3" json:"conn_id,omitempty"` // 刚登陆时没有user_id，只有conn_id + UserId int64 `protobuf:"varint,3,opt,name=user_id,json=userId,proto3" json:"user_id,omitempty"` // 玩家id + MsgId int32 `protobuf:"varint,4,opt,name=msg_id,json=msgId,proto3" json:"msg_id,omitempty"` // 消息id + Msg []byte `protobuf:"bytes,5,opt,name=msg,proto3" json:"msg,omitempty"` // 消息 + Type MsgType `protobuf:"varint,6,opt,name=type,proto3,enum=ipb.MsgType" json:"type,omitempty"` // 消息类型 + RetRpcMsgId int32 `protobuf:"varint,7,opt,name=ret_rpc_msg_id,json=retRpcMsgId,proto3" json:"ret_rpc_msg_id,omitempty"` // 依赖本字段把数据回传给service_name节点的rpc processor去处理 + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *InternalMsg) Reset() { + *x = InternalMsg{} + mi := &file_internal_proto_msgTypes[0] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *InternalMsg) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*InternalMsg) ProtoMessage() {} + +func (x *InternalMsg) ProtoReflect() protoreflect.Message { + mi := &file_internal_proto_msgTypes[0] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use InternalMsg.ProtoReflect.Descriptor instead. +func (*InternalMsg) Descriptor() ([]byte, []int) { + return file_internal_proto_rawDescGZIP(), []int{0} +} + +func (x *InternalMsg) GetServiceName() string { + if x != nil { + return x.ServiceName + } + return "" +} + +func (x *InternalMsg) GetConnId() uint32 { + if x != nil { + return x.ConnId + } + return 0 +} + +func (x *InternalMsg) GetUserId() int64 { + if x != nil { + return x.UserId + } + return 0 +} + +func (x *InternalMsg) GetMsgId() int32 { + if x != nil { + return x.MsgId + } + return 0 +} + +func (x *InternalMsg) GetMsg() []byte { + if x != nil { + return x.Msg + } + return nil +} + +func (x *InternalMsg) GetType() MsgType { + if x != nil { + return x.Type + } + return MsgType_NormalMsg +} + +func (x *InternalMsg) GetRetRpcMsgId() int32 { + if x != nil { + return x.RetRpcMsgId + } + return 0 +} + +var File_internal_proto protoreflect.FileDescriptor + +const file_internal_proto_rawDesc = "" + + "\n" + + "\x0einternal.proto\x12\x03ipb\"\xd2\x01\n" + + "\vInternalMsg\x12!\n" + + "\fservice_name\x18\x01 \x01(\tR\vserviceName\x12\x17\n" + + "\aconn_id\x18\x02 \x01(\rR\x06connId\x12\x17\n" + + "\auser_id\x18\x03 \x01(\x03R\x06userId\x12\x15\n" + + "\x06msg_id\x18\x04 \x01(\x05R\x05msgId\x12\x10\n" + + "\x03msg\x18\x05 \x01(\fR\x03msg\x12 \n" + + "\x04type\x18\x06 \x01(\x0e2\f.ipb.MsgTypeR\x04type\x12#\n" + + "\x0eret_rpc_msg_id\x18\a \x01(\x05R\vretRpcMsgId*+\n" + + "\x05MsgId\x12\v\n" + + "\aUnknown\x10\x00\x12\x15\n" + + "\bInternal\x10\xff\xff\xff\xff\xff\xff\xff\xff\xff\x01*$\n" + + "\aMsgType\x12\r\n" + + "\tNormalMsg\x10\x00\x12\n" + + "\n" + + "\x06RpcMsg\x10\x01B\tZ\afox/ipbb\x06proto3" + +var ( + file_internal_proto_rawDescOnce sync.Once + file_internal_proto_rawDescData []byte +) + +func file_internal_proto_rawDescGZIP() []byte { + file_internal_proto_rawDescOnce.Do(func() { + file_internal_proto_rawDescData = protoimpl.X.CompressGZIP(unsafe.Slice(unsafe.StringData(file_internal_proto_rawDesc), len(file_internal_proto_rawDesc))) + }) + return file_internal_proto_rawDescData +} + +var file_internal_proto_enumTypes = make([]protoimpl.EnumInfo, 2) +var file_internal_proto_msgTypes = make([]protoimpl.MessageInfo, 1) +var file_internal_proto_goTypes = []any{ + (MsgId)(0), // 0: ipb.MsgId + (MsgType)(0), // 1: ipb.MsgType + (*InternalMsg)(nil), // 2: ipb.InternalMsg +} +var file_internal_proto_depIdxs = []int32{ + 1, // 0: ipb.InternalMsg.type:type_name -> ipb.MsgType + 1, // [1:1] is the sub-list for method output_type + 1, // [1:1] is the sub-list for method input_type + 1, // [1:1] is the sub-list for extension type_name + 1, // [1:1] is the sub-list for extension extendee + 0, // [0:1] is the sub-list for field type_name +} + +func init() { file_internal_proto_init() } +func file_internal_proto_init() { + if File_internal_proto != nil { + return + } + type x struct{} + out := protoimpl.TypeBuilder{ + File: protoimpl.DescBuilder{ + GoPackagePath: reflect.TypeOf(x{}).PkgPath(), + RawDescriptor: unsafe.Slice(unsafe.StringData(file_internal_proto_rawDesc), len(file_internal_proto_rawDesc)), + NumEnums: 2, + NumMessages: 1, + NumExtensions: 0, + NumServices: 0, + }, + GoTypes: file_internal_proto_goTypes, + DependencyIndexes: file_internal_proto_depIdxs, + EnumInfos: file_internal_proto_enumTypes, + MessageInfos: file_internal_proto_msgTypes, + }.Build() + File_internal_proto = out.File + file_internal_proto_goTypes = nil + file_internal_proto_depIdxs = nil +} diff --git a/service/baseService.go b/service/baseService.go index 88c8bca..c056e47 100644 --- a/service/baseService.go +++ b/service/baseService.go @@ -3,6 +3,7 @@ package service import ( "context" "fmt" + "github.com/fox/fox/ipb" "github.com/fox/fox/ksync" "github.com/fox/fox/log" "github.com/fox/fox/safeChan" @@ -90,17 +91,17 @@ func (s *BaseService) RunWait(cb func() (retValue any)) (retValue any, err error } } -func (s *BaseService) Send(topic string, msg []byte) error { +func (s *BaseService) Send(topic string, msg *ipb.InternalMsg) error { if s.sender != nil { return s.sender.Send(topic, msg) } return s.Err("send is nil") } -func (s *BaseService) Call(topic string, timeout time.Duration, msg []byte) ([]byte, error) { +func (s *BaseService) Call(rpcTopic string, msg *ipb.InternalMsg, cb func(msg *ipb.InternalMsg)) error { if s.sender != nil { - return s.sender.Call(topic, timeout, msg) + return s.sender.Call(rpcTopic, msg, cb) } - return nil, s.Err("call is nil") + return s.Err("call is nil") } func (s *BaseService) WaitStop() { diff --git a/service/iservice.go b/service/iservice.go index f916a6a..4e0b8f3 100644 --- a/service/iservice.go +++ b/service/iservice.go @@ -1,6 +1,7 @@ package service import ( + "github.com/fox/fox/ipb" "time" ) @@ -16,16 +17,16 @@ type IService interface { // 向服务内部消息管道写入消息 Write(msg []byte) error - Send(topic string, msg []byte) error - Call(topic string, timeout time.Duration, msg []byte) ([]byte, error) + Send(topic string, msg *ipb.InternalMsg) error + Call(rpcTopic string, msg *ipb.InternalMsg, cb func(msg *ipb.InternalMsg)) error WaitStop() NotifyStop() } type ISender interface { - Send(topic string, msg []byte) error - Call(topic string, timeout time.Duration, msg []byte) ([]byte, error) + Send(topic string, msg *ipb.InternalMsg) error + Call(rpcTopic string, msg *ipb.InternalMsg, cb func(msg *ipb.InternalMsg)) error } type IOnFunc interface { diff --git a/service/natsService.go b/service/natsService.go index 5070994..8027d02 100644 --- a/service/natsService.go +++ b/service/natsService.go @@ -3,12 +3,14 @@ package service import ( "encoding/json" "github.com/fox/fox/etcd" + "github.com/fox/fox/ipb" "github.com/fox/fox/log" "github.com/fox/fox/nat" + "github.com/fox/fox/processor" + "github.com/golang/protobuf/proto" "github.com/nats-io/nats.go" "os" "sync" - "time" ) const ( @@ -30,9 +32,10 @@ type InitNatsServiceParams struct { type NatsService struct { *BaseService - nats *nat.Nats - registry *etcd.Registry[etcd.ServiceNode] - node etcd.ServiceNode // 本服务节点信息 + nats *nat.Nats + registry *etcd.Registry[etcd.ServiceNode] + rpcProcessor *processor.Processor + node etcd.ServiceNode // 本服务节点信息 } func NewNatsService(param *InitNatsServiceParams) (*NatsService, error) { @@ -52,8 +55,6 @@ func NewNatsService(param *InitNatsServiceParams) (*NatsService, error) { return nil, err } s.registry.WatchServices() - // 广播服务上线 - s.publishUpdateService() s.nats = nat.NewNats(param.ServiceName, param.NatsAddress...) if err = s.nats.Connect(); err != nil { @@ -64,6 +65,12 @@ func NewNatsService(param *InitNatsServiceParams) (*NatsService, error) { if err = s.Subscribe(Topic(s)); err != nil { log.Error(err.Error()) } + // 订阅广播服务上线 + _ = s.subscribeUpdateService() + s.publishUpdateService() + + // 订阅rpc回调 + _ = s.subscribeRpc() return s, nil } @@ -73,7 +80,7 @@ func (n *NatsService) publishUpdateService() { } func (n *NatsService) subscribeUpdateService() error { - return n.nats.SubscribeCb(updateTopic, func(m *nats.Msg) { + return n.SubscribeCb(updateTopic, func(m *nats.Msg) { var node = &etcd.ServiceNode{} _ = json.Unmarshal(m.Data, node) // 不是同类服务不处理，是自己发出来的更新，也不处理 @@ -90,9 +97,21 @@ func (n *NatsService) subscribeUpdateService() error { }) } -// 订阅回调，极少用。调用方自行保证并发性问题 +func (n *NatsService) subscribeRpc() error { + return n.SubscribeCb(RpcTopic(n), func(m *nats.Msg) { + var iMsg = &ipb.InternalMsg{} + _ = proto.Unmarshal(m.Data, iMsg) + _ = n.rpcProcessor.Dispatch(iMsg.RetRpcMsgId, iMsg) + }) +} + +// 订阅回调 func (n *NatsService) SubscribeCb(topic string, cb func(m *nats.Msg)) error { - return n.nats.SubscribeCb(topic, cb) + return n.nats.SubscribeCb(topic, func(m *nats.Msg) { + n.RunOnce(func() { + cb(m) + }) + }) } func (n *NatsService) Subscribe(topic string) error { @@ -108,15 +127,19 @@ func (s *NatsService) OnStop() { s.nats.Close() } -func (s *NatsService) Send(topic string, msg []byte) error { - return s.nats.Publish(topic, msg) +func (s *NatsService) Send(topic string, msg *ipb.InternalMsg) error { + data, _ := proto.Marshal(msg) + return s.nats.Publish(topic, data) } -func (s *NatsService) Call(topic string, timeout time.Duration, msg []byte) ([]byte, error) { - _ = topic - _ = timeout - _ = msg - return nil, nil +func (s *NatsService) Call(rpcTopic string, msg *ipb.InternalMsg, cb func(msg *ipb.InternalMsg)) error { + data, _ := proto.Marshal(ipb.MakeRpcMsg(msg.ServiceName, msg.ConnId, msg.UserId, msg.MsgId, msg.Msg)) + err := s.nats.Publish(rpcTopic, data) + if err != nil { + return err + } + s.rpcProcessor.RegisterMessage(msg.RetRpcMsgId, ipb.InternalMsg{}, cb) + return nil } func (s *NatsService) ServiceEtcd() *etcd.Registry[etcd.ServiceNode] { diff --git a/service/topic.go b/service/topic.go index fdbbdb5..6d19865 100644 --- a/service/topic.go +++ b/service/topic.go @@ -1,8 +1,9 @@ package service const ( - extTopic = ".topic" - extGroup = ".group" + extTopic = ".topic" + extGroup = ".group" + extRpcTopic = ".rpc.topic" ) // 每个服务都有自己的服务topic @@ -14,6 +15,15 @@ func TopicEx(serviceName string) string { return serviceName + extTopic } +// 每个服务都有自己的rpc服务topic +func RpcTopic(s IService) string { + return s.Name() + extRpcTopic +} + +func RpcTopicEx(serviceName string) string { + return serviceName + extRpcTopic +} + func GroupTopic(s IService) string { return s.Type() + extTopic }