合并Service中的Event、RpcResponse、RpcRequest管道

service/module.go

@@ -35,8 +35,6 @@ type IModuleTimer interface {
 	NewTicker(d time.Duration, cb func(*timer.Ticker)) *timer.Ticker
 }
 
-//1.管理各模块树层关系
-//2.提供定时器常用工具
 type Module struct {
 	rpcHandle.IRpcHandler
 	moduleId int64                              //模块Id
@@ -80,8 +78,9 @@ func (m *Module) OnInit() error{
 func (m *Module) AddModule(module IModule) (int64,error){
 	//没有事件处理器不允许加入其他模块
 	if m.GetEventProcessor() == nil {
-		return 0,fmt.Errorf("module %+v is not Event Processor is nil", m.self)
+		return 0,fmt.Errorf("module %+v Event Processor is nil", m.self)
 	}
+
 	pAddModule := module.getBaseModule().(*Module)
 	if pAddModule.GetModuleId()==0 {
 		pAddModule.moduleId = m.NewModuleId()
@@ -92,7 +91,7 @@ func (m *Module) AddModule(module IModule) (int64,error){
 	}
 	_,ok := m.child[module.GetModuleId()]
 	if ok == true {
-		return 0,fmt.Errorf("Exists module id %d",module.GetModuleId())
+		return 0,fmt.Errorf("exists module id %d",module.GetModuleId())
 	}
 	pAddModule.IRpcHandler = m.IRpcHandler
 	pAddModule.self = module
@@ -118,14 +117,14 @@ func (m *Module) ReleaseModule(moduleId int64){
 	pModule := m.GetModule(moduleId).getBaseModule().(*Module)
 
 	//释放子孙
-	for id,_ := range pModule.child {
+	for id := range pModule.child {
 		m.ReleaseModule(id)
 	}
 
 	pModule.GetEventHandler().Destroy()
 	pModule.self.OnRelease()
 	log.SDebug("Release module ", pModule.GetModuleName())
-	for pTimer,_ := range pModule.mapActiveTimer {
+	for pTimer := range pModule.mapActiveTimer {
 		pTimer.Cancel()
 	}
 

service/service.go

@@ -1,14 +1,16 @@
 package service
 
 import (
-	"fmt"
+	"errors"
 	"github.com/duanhf2012/origin/event"
 	"github.com/duanhf2012/origin/log"
 	"github.com/duanhf2012/origin/profiler"
 	"github.com/duanhf2012/origin/rpc"
+	originSync "github.com/duanhf2012/origin/util/sync"
 	"github.com/duanhf2012/origin/util/timer"
 	"reflect"
 	"runtime"
+	"strconv"
 	"sync"
 	"sync/atomic"
 )
@@ -33,6 +35,12 @@ type IService interface {
 	GetProfiler() *profiler.Profiler
 }
 
+// eventPool的内存池,缓存Event
+var maxServiceEventChannel = 2000000
+var eventPool = originSync.NewPoolEx(make(chan originSync.IPoolData, maxServiceEventChannel), func() originSync.IPoolData {
+	return &event.Event{}
+})
+
 type Service struct {
 	Module
 	rpcHandler rpc.RpcHandler           //rpc
@@ -44,14 +52,23 @@ type Service struct {
 	eventProcessor event.IEventProcessor
 	profiler *profiler.Profiler //性能分析器
 	rpcEventLister rpc.IRpcListener
+	chanEvent chan event.IEvent
 }
 
-type RpcEventData struct{
+// RpcConnEvent Node结点连接事件
+type RpcConnEvent struct{
 	IsConnect bool
 	NodeId int
 }
 
-func (rpcEventData *RpcEventData) GetEventType() event.EventType{
+func SetMaxServiceChannel(maxEventChannel int){
+	maxServiceEventChannel = maxEventChannel
+	eventPool = originSync.NewPoolEx(make(chan originSync.IPoolData, maxServiceEventChannel), func() originSync.IPoolData {
+		return &event.Event{}
+	})
+}
+
+func (rpcEventData *RpcConnEvent) GetEventType() event.EventType{
 	return event.Sys_Event_Rpc_Event
 }
 
@@ -70,8 +87,8 @@ func (s *Service) OpenProfiler()  {
 
 func (s *Service) Init(iService IService,getClientFun rpc.FuncRpcClient,getServerFun rpc.FuncRpcServer,serviceCfg interface{}) {
 	s.dispatcher =timer.NewDispatcher(timerDispatcherLen)
-
-	s.rpcHandler.InitRpcHandler(iService.(rpc.IRpcHandler),getClientFun,getServerFun)
+	s.chanEvent = make(chan event.IEvent,maxServiceEventChannel)
+	s.rpcHandler.InitRpcHandler(iService.(rpc.IRpcHandler),getClientFun,getServerFun,iService.(rpc.IRpcHandlerChannel))
 	s.IRpcHandler = &s.rpcHandler
 	s.self = iService.(IModule)
 	//初始化祖先
@@ -81,25 +98,14 @@ func (s *Service) Init(iService IService,getClientFun rpc.FuncRpcClient,getServe
 	s.serviceCfg = serviceCfg
 	s.goroutineNum = 1
 	s.eventProcessor = event.NewEventProcessor()
+	s.eventProcessor.Init(s)
 	s.eventHandler =  event.NewEventHandler()
 	s.eventHandler.Init(s.eventProcessor)
 }
 
-func (s *Service) SetGoRoutineNum(goroutineNum int32) bool {
-	//已经开始状态不允许修改协程数量,打开性能分析器不允许开多线程
-	if s.startStatus == true || s.profiler!=nil {
-		log.SError("open profiler mode is not allowed to set Multi-coroutine.")
-		return false
-	}
-
-	s.goroutineNum = goroutineNum
-	return true
-}
 
 func (s *Service) Start() {
 	s.startStatus = true
-	s.eventProcessor.SetEventChannel(0)
-	
 	for i:=int32(0);i< s.goroutineNum;i++{
 		s.wg.Add(1)
 		go func(){
@@ -114,41 +120,64 @@ func (s *Service) Run() {
 	var bStop = false
 	s.self.(IService).OnStart()
 	for{
-		rpcRequestChan := s.GetRpcRequestChan()
-		rpcResponseCallBack := s.GetRpcResponseChan()
-		eventChan := s.eventProcessor.GetEventChan()
 		var analyzer *profiler.Analyzer
 		select {
 		case <- closeSig:
 			bStop = true
-		case rpcRequest :=<- rpcRequestChan:
-			if s.profiler!=nil {
-				analyzer = s.profiler.Push("[Req]"+rpcRequest.RpcRequestData.GetServiceMethod())
+		case ev := <- s.chanEvent:
+			switch ev.GetEventType() {
+			case event.ServiceRpcRequestEvent:
+				cEvent,ok := ev.(*event.Event)
+				if ok == false {
+					log.SError("Type event conversion error")
+					break
+				}
+				rpcRequest,ok := cEvent.Data.(*rpc.RpcRequest)
+				if ok == false {
+					log.SError("Type *rpc.RpcRequest conversion error")
+					break
+				}
+				if s.profiler!=nil {
+					analyzer = s.profiler.Push("[Req]"+rpcRequest.RpcRequestData.GetServiceMethod())
+				}
+
+				s.GetRpcHandler().HandlerRpcRequest(rpcRequest)
+				if analyzer!=nil {
+					analyzer.Pop()
+					analyzer = nil
+				}
+				eventPool.Put(cEvent)
+			case event.ServiceRpcResponseEvent:
+				cEvent,ok := ev.(*event.Event)
+				if ok == false {
+					log.SError("Type event conversion error")
+					break
+				}
+				rpcResponseCB,ok := cEvent.Data.(*rpc.Call)
+				if ok == false {
+					log.SError("Type *rpc.Call conversion error")
+					break
+				}
+				if s.profiler!=nil {
+					analyzer = s.profiler.Push("[Res]" + rpcResponseCB.ServiceMethod)
+				}
+				s.GetRpcHandler().HandlerRpcResponseCB(rpcResponseCB)
+				if analyzer!=nil {
+					analyzer.Pop()
+					analyzer = nil
+				}
+				eventPool.Put(cEvent)
+			default:
+				if s.profiler!=nil {
+					analyzer = s.profiler.Push("[SEvent]"+strconv.Itoa(int(ev.GetEventType())))
+				}
+				s.eventProcessor.EventHandler(ev)
+				if analyzer!=nil {
+					analyzer.Pop()
+					analyzer = nil
+				}
 			}
 
-			s.GetRpcHandler().HandlerRpcRequest(rpcRequest)
-			if analyzer!=nil {
-				analyzer.Pop()
-				analyzer = nil
-			}
-		case rpcResponseCB := <-rpcResponseCallBack:
-			if s.profiler!=nil {
-				analyzer = s.profiler.Push("[Res]" + rpcResponseCB.ServiceMethod)
-			}
-			s.GetRpcHandler().HandlerRpcResponseCB(rpcResponseCB)
-			if analyzer!=nil {
-				analyzer.Pop()
-				analyzer = nil
-			}
-		case ev := <- eventChan:
-			if s.profiler!=nil {
-				analyzer = s.profiler.Push(fmt.Sprintf("[Event]%d", int(ev.GetEventType())))
-			}
-			s.eventProcessor.EventHandler(ev)
-			if analyzer!=nil {
-				analyzer.Pop()
-				analyzer = nil
-			}
 		case t := <- s.dispatcher.ChanTimer:
 			if s.profiler != nil {
 				analyzer = s.profiler.Push("[timer]"+t.GetName())
@@ -211,11 +240,11 @@ func (s *Service) GetProfiler() *profiler.Profiler{
 }
 
 func (s *Service) RegEventReceiverFunc(eventType event.EventType, receiver event.IEventHandler,callback event.EventCallBack){
-	s.eventProcessor.RegEventReciverFunc(eventType, receiver,callback)
+	s.eventProcessor.RegEventReceiverFunc(eventType, receiver,callback)
 }
 
 func (s *Service) UnRegEventReceiverFunc(eventType event.EventType, receiver event.IEventHandler){
-	s.eventProcessor.UnRegEventReciverFun(eventType, receiver)
+	s.eventProcessor.UnRegEventReceiverFun(eventType, receiver)
 }
 
 func (s *Service) IsSingleCoroutine() bool {
@@ -230,7 +259,7 @@ func (s *Service) OnStart(){
 }
 
 func (s *Service) OnRpcEvent(ev event.IEvent){
-	event := ev.(*RpcEventData)
+	event := ev.(*RpcConnEvent)
 	if event.IsConnect {
 		s.rpcEventLister.OnNodeConnected(event.NodeId)
 	}else{
@@ -249,3 +278,46 @@ func (s *Service) UnRegRpcListener(rpcLister rpc.IRpcListener) {
 	RegRpcEventFun(s.GetName())
 }
 
+
+func (s *Service) PushRpcRequest(rpcRequest *rpc.RpcRequest) error{
+	ev := eventPool.Get().(*event.Event)
+	ev.Type = event.ServiceRpcRequestEvent
+	ev.Data = rpcRequest
+
+	return s.pushEvent(ev)
+}
+
+func (s *Service) PushRpcResponse(call *rpc.Call) error{
+	ev := eventPool.Get().(*event.Event)
+	ev.Type = event.ServiceRpcResponseEvent
+	ev.Data = call
+
+	return s.pushEvent(ev)
+}
+
+func (s *Service) PushEvent(ev event.IEvent) error{
+	return s.pushEvent(ev)
+}
+
+func (s *Service) pushEvent(ev event.IEvent) error{
+	if len(s.chanEvent) >= maxServiceEventChannel {
+		err := errors.New("The event channel in the service is full")
+		log.SError(err.Error())
+		return err
+	}
+
+	s.chanEvent <- ev
+	return nil
+}
+
+
+func (s *Service) SetGoRoutineNum(goroutineNum int32) bool {
+	//已经开始状态不允许修改协程数量,打开性能分析器不允许开多线程
+	if s.startStatus == true || s.profiler!=nil {
+		log.SError("open profiler mode is not allowed to set Multi-coroutine.")
+		return false
+	}
+
+	s.goroutineNum = goroutineNum
+	return true
+}