分布式 EventBus：基於 Socket 的發布訂閱實現

背景

本地 EventBus 僅限進程內，跨服務器事件傳播需分佈式方案。選用 C# Socket + SuperSocket 框架，實現即時推送到所有訂閱相同事件的服務器。

架構

• EventBus Server：中心節點，基於 SuperSocket，處理三條命令
• EventBus Client Broker：各應用服務器嵌入，連接 Server，訂閱/發布事件

通信協議 (三條命令)

1. ME [ServerIdentity]：客戶端註冊身份
2. Subscribe [EventClassFullName]：訂閱事件類型
3. Publish [EventClass] [EventXml]：發布事件，Server 轉發給所有訂閱該類型的其他客戶端

Server 端命令實現

public class ME : CommandBase {
    public override void ExecuteCommand(...) {
        session.AssociatedServerIdentity = RestoreSpecialCharacters(requestInfo.Parameters[0]);
        Console.WriteLine($"[{session.AssociatedServerIdentity}] connected.");
    }
}

public class Subscribe : CommandBase<...> {
    public override void ExecuteCommand(...) {
        var classPath = RestoreSpecialCharacters(requestInfo.Parameters[0]);
        if (!session.SubscribedEventClasses.Contains(classPath))
            session.SubscribedEventClasses.Add(classPath);
        // 日誌輸出訂閱列表
    }
}

public class Publish : CommandBase<...> {
    public override void ExecuteCommand(...) {
        var evtClass = RestoreSpecialCharacters(requestInfo.Parameters[0]);
        foreach (var s in session.AppServer.GetAllSessions()) {
            if (s.AssociatedServerIdentity == session.AssociatedServerIdentity) continue; // 不回發給自己
            if (s.SubscribedEventClasses.Contains(evtClass)) {
                s.Send("Publish " + requestInfo.Body); // 透傳原始消息體
                Console.WriteLine($"Forward {evtClass} from {session.AssociatedServerIdentity} to {s.AssociatedServerIdentity}");
            }
        }
    }
}

客戶端 Broker 使用

var broker = new EventBusClientBroker("127.0.0.1", 2012, "AppServer2", "AppEvents.dll");
broker.NewEventReceived += (evtClass, evtObj) => {
    if (evtObj is NewUserRegisteredEvent e) Console.WriteLine(e.UserName);
    // ...
};
broker.Connect();
broker.Subscribe();

// 發布事件
broker.Publish(new NewUserRegisteredEvent { UserName = "aaron", RegisterDate = DateTime.Now });

效果演示

Server 啟動後，多客戶端連接、訂閱、發布，事件即時轉發到訂閱者。

代碼下載

EventBus2.rar