分布式EventBus的Socket实现

￡神魔★判官ぃ 2021-11-26 12:32 207阅读 0赞

在这篇文章中，[EventBus实现 - 发布订阅 - XML加载][EventBus_ - _ - XML] 所适用的范围只是本机的事件传播，要是牵涉到多台服务器之间的事件传播就不行了，解决办法有用msmq解决的，Node.js解决的，也有用redis的发布订阅解决的，这次用C\# socket来实现，能实现立刻推送事件到所有订阅了相关event的server上。

这次的子系统适用的场景如下：

![SouthEast][]

主要分2个部分：各个server使用的Event Bus Broker以及Event Bus Server。

Broker与Server之间的通信协议就3个：ME、Subscribe、Publish。分别代表：我的名字是、我要订阅的事件是、我触发事件。

Event Bus Server是基于SuperSocket开源组件写的，非常方便。实现了3个对应的命令类（建议大家先看看[SuperSocket][]的文档），如下：

public class ME : CommandBase<EventDispatcherBusSession, StringRequestInfo>
        {
            public override void ExecuteCommand(EventDispatcherBusSession session, StringRequestInfo requestInfo)
            {
                session.AssociatedServerIdentity = EventBase.Helper.RestoreSpecialCharacters(requestInfo.Parameters[0]);
                Console.WriteLine(string.Format("[{0}]: connected.", session.AssociatedServerIdentity));
            }
        }
    
    public class Publish : CommandBase<EventDispatcherBusSession, StringRequestInfo>
        {
            public override void ExecuteCommand(EventDispatcherBusSession session, StringRequestInfo requestInfo)
            {
                string evtClassPath = EventBase.Helper.RestoreSpecialCharacters(requestInfo.Parameters[0]);
                string evtXml = EventBase.Helper.RestoreSpecialCharacters(requestInfo.Parameters[1]);
    
                foreach(EventDispatcherBusSession s in session.AppServer.GetAllSessions())
                {
                    if (s.AssociatedServerIdentity == session.AssociatedServerIdentity)
                        continue;
    
                    if (s.SubscribedEventClasses.Contains(evtClassPath))
                    {
                        s.Send("Publish " + requestInfo.Body);//forward only
                        Console.WriteLine(string.Format("Forwarding publish command from {1} to server: [{0}]", s.AssociatedServerIdentity, session.AssociatedServerIdentity));
                    }
                }
            }
        }
    
    public class Subscribe : CommandBase<EventDispatcherBusSession, StringRequestInfo>
        {
            public override void ExecuteCommand(EventDispatcherBusSession session, StringRequestInfo requestInfo)
            {
                string classPath = EventBase.Helper.RestoreSpecialCharacters(requestInfo.Parameters[0]);
                if (session.SubscribedEventClasses.Contains(classPath))
                    return;
    
                session.SubscribedEventClasses.Add(classPath);
    
                string msg = "";
                msg+=string.Format("[{0}], now subscribed these events:\r\n", session.AssociatedServerIdentity);
                foreach(string evtClass in session.SubscribedEventClasses)
                    msg += string.Format("              {0}\r\n", evtClass);
                Console.WriteLine(msg);
            }
        }

启动server代码：

static void Main(string[] args)
            {
                var appServer = new EventDispatcherBusServer();
    
                if (!appServer.Setup(2012))
                {
                    Console.WriteLine("Failed to setup!");
                    Console.ReadKey();
                    return;
                }
    
                Console.WriteLine();
    
                if (!appServer.Start())
                {
                    Console.WriteLine("Failed to start!");
                    Console.ReadKey();
                    return;
                }
    
                Console.WriteLine("The server started successfully, press key 'q' to stop it!");
    
                Console.ReadKey();
    
                appServer.Stop();
            }

server端没多少好说的，都是很简单的，下面来看看客户端：

static void Main(string[] args)
            {
                EventBusClientBroker busBroker = new EventBusClientBroker("127.0.0.1", 2012, "App server 2", "AppEvents.dll");
                busBroker.NewEventReceived += new EventReceived(busBroker_NewEventReceived);//新event被推送过来的事件
                busBroker.Connect();
    
                busBroker.Subscribe<NewUserRegisteredEvent>();//订阅Event
    
                //busBroker.Close();
            }
    
            static void busBroker_NewEventReceived(string evtClass, object evt)
            {
                //这里会将event对象发过来
                //evtClass是evt的class全路径
                if (evt is NewUserRegisteredEvent)
                    Console.WriteLine(((NewUserRegisteredEvent)evt).UserName);
                else if (evt is UserProfileUpdatedEvent)
                    Console.WriteLine(((UserProfileUpdatedEvent)evt).UserID);
            }

如果要触发一个事件，则：

NewUserRegisteredEvent evt = new NewUserRegisteredEvent();
                        evt.RegisterDate = DateTime.Now;
                        evt.UserName = "aaron";
                        busBroker.Publish<NewUserRegisteredEvent>(evt);

我们来看看效果图：

![SouthEast 1][]

搞定，[代码下载][Link 1]。

[EventBus_ - _ - XML]: http://www.cnblogs.com/aarond/p/EventBus.html
[SouthEast]: /images/20211126/9f3f04af325d4f84b2c57d3620d9267e.png
[SuperSocket]: http://docs.supersocket.net/v1-5/zh-CN
[SouthEast 1]: /images/20211126/a098f48010c44dff9cbc7cd8aac95beb.png
[Link 1]: http://files.cnblogs.com/aarond/EventBus2.rar