C#使用Socket实现分布式事件总线,不依赖第三方MQ

简介: `CodeWF.EventBus.Socket` 是一个轻量级的、基于Socket的分布式事件总线系统,旨在简化分布式架构中的事件通信。它允许进程之间通过发布/订阅模式进行通信,无需依赖外部消息队列服务。

使用 Socket 实现的分布式事件总线,支持 CQRS,不依赖第三方 MQ。

CodeWF.EventBus.Socket 是一个轻量级的、基于 Socket 的分布式事件总线系统,旨在简化分布式架构中的事件通信。它允许进程之间通过发布/订阅模式进行通信,无需依赖外部消息队列服务。

Command

Command

Query

Query

特性

  • 轻量级:不依赖任何外部 MQ 服务,减少了系统复杂性和依赖。

  • 高性能:基于 Socket 的直接通信,提供低延迟、高吞吐量的消息传递。

  • 灵活性:支持自定义事件类型和消息处理器,易于集成到现有系统中。

  • 可扩展性:支持多客户端连接,适用于分布式系统环境。

通信协议

通过TCP 协议进行数据交互,协议包结构如下:

0.0.8@2x

安装

通过NuGet包管理器安装CodeWF.EventBus.Socket

Install-Package CodeWF.EventBus.Socket

服务端使用

运行事件服务

在服务端代码中,创建并启动EventServer实例以监听客户端连接和事件:

using CodeWF.EventBus.Socket;

// 创建事件服务器实例
IEventServer eventServer = new EventServer();

// 启动事件服务器,监听指定IP和端口
eventServer.Start("127.0.0.1", 9100);

停止事件服务

当不再需要事件服务时,调用Stop方法以优雅地关闭服务器:

eventServer.Stop();

客户端使用

连接事件服务

在客户端代码中,创建EventClient实例并连接到事件服务器:

using CodeWF.EventBus.Socket;

// 创建事件客户端实例
IEventClient eventClient = new EventClient();

// 连接到事件服务器,使用eventClient.ConnectStatus检查连接状态
eventClient.Connect("127.0.0.1", 9100));

订阅事件

订阅特定类型的事件,并指定事件处理函数:

eventClient.Subscribe<NewEmailCommand>("event.email.new", ReceiveNewEmailCommand);

private void ReceiveNewEmail(NewEmailCommand command)
{
   
    // 处理新邮件通知
    Console.WriteLine($"收到新邮件,主题是{message.Subject}");
}

发布命令(Command)

发布事件到指定的主题,供已订阅的客户端处理:

// 发布新邮件通知事件
eventClient.Publish("event.email.new", new NewEmailCommand {
    Subject = "恭喜您中Github一等奖", Content = "我们很开心,您在2024年7月...", SendTime = new DateTime(2024, 7, 27) });

查询(Query)

查询指定主题,需要有接收查询端订阅相同的主题(即生产者),收到请求后,再以相同的主题发布查询结果:

eventClient.Subscribe<EmailQuery>("event.email.query", ReceiveEmailQuery);

private void ReceiveEmailQuery(EmailQuery query)
{
   
    // 执行查询请求,准备查询结果
    var response = new EmailQueryResponse {
    Emails = EmailManager.QueryEmail(request.Subject) };

    // 以相同的主题,发布查询结果
    if (_eventClient!.Publish("event.email.query", response,
        out var errorMessage))
    {
   
        Logger.Info($"Response query result: {response}");
    }
    else
    {
   
        Logger.Error($"Response query failed: {errorMessage}");
    }
}

其他端可使用相同的主题查询(即消费者):

var response = _eventClient!.Query<EmailQuery, EmailQueryResponse>("event.email.query",
    new EmailQuery() {
    Subject = "Account" },
    out var errorMessage);
if (string.IsNullOrWhiteSpace(errorMessage) && response != null)
{
   
    Logger.Info($"Query event.email.query, result: {response}");
}
else
{
   
    Logger.Error(
        $"Query event.email.query failed: [{errorMessage}]");
}

取消订阅事件

不再需要接收某类事件时,可以取消订阅:

eventClient.Unsubscribe<NewEmailNotification>("event.email.new", ReceiveNewEmail);

断开事件服务

完成事件处理或需要断开与服务器的连接时,调用Disconnect方法:

eventClient.Disconnect();
Console.WriteLine("断开与事件服务的连接");

注意事项

  • 确保服务端和客户端使用的地址和端口号一致,并且端口未被其他服务占用。
  • 在生产环境中,服务端应配置为监听公共 IP 地址或适当的网络接口。
  • 考虑到网络异常和服务重启等情况,客户端可能需要实现重连逻辑。
  • 根据实际需求,可以扩展EventServerEventClient类以支持更复杂的功能,如消息加密、认证授权等。
相关实践学习
消息队列RocketMQ版:基础消息收发功能体验
本实验场景介绍消息队列RocketMQ版的基础消息收发功能,涵盖实例创建、Topic、Group资源创建以及消息收发体验等基础功能模块。
消息队列 MNS 入门课程
1、消息队列MNS简介 本节课介绍消息队列的MNS的基础概念 2、消息队列MNS特性 本节课介绍消息队列的MNS的主要特性 3、MNS的最佳实践及场景应用 本节课介绍消息队列的MNS的最佳实践及场景应用案例 4、手把手系列:消息队列MNS实操讲 本节课介绍消息队列的MNS的实际操作演示 5、动手实验:基于MNS,0基础轻松构建 Web Client 本节课带您一起基于MNS,0基础轻松构建 Web Client
相关文章
|
1月前
|
C# 开发者
C# 一分钟浅谈:Socket 编程基础
【10月更文挑战第7天】本文介绍了Socket编程的基础知识、基本操作及常见问题,通过C#代码示例详细展示了服务器端和客户端的Socket通信过程,包括创建、绑定、监听、连接、数据收发及关闭等步骤,帮助开发者掌握Socket编程的核心技术和注意事项。
92 3
C# 一分钟浅谈:Socket 编程基础
|
1月前
|
存储 开发框架 .NET
C#语言如何搭建分布式文件存储系统
C#语言如何搭建分布式文件存储系统
72 2
|
1月前
|
API C# Windows
【C#】在winform中如何实现嵌入第三方软件窗体
【C#】在winform中如何实现嵌入第三方软件窗体
95 0
|
1月前
|
存储 分布式计算 监控
C# 创建一个分布式文件存储系统需要怎么设计??
C# 创建一个分布式文件存储系统需要怎么设计??
37 0
|
3月前
|
消息中间件 存储 NoSQL
MQ的顺序性保证:顺序队列、消息编号、分布式锁,一文全掌握!
【8月更文挑战第24天】消息队列(MQ)是分布式系统的关键组件,用于实现系统解耦、提升可扩展性和可用性。保证消息顺序性是其重要挑战之一。本文介绍三种常用策略:顺序队列、消息编号与分布式锁,通过示例展示如何确保消息按需排序。这些方法各有优势,可根据实际场景灵活选用。提供的Java示例有助于加深理解与实践应用。
106 2
|
4月前
|
SQL 开发框架 前端开发
在C#开发中使用第三方组件LambdaParser、DynamicExpresso、Z.Expressions,实现动态解析/求值字符串表达式
在C#开发中使用第三方组件LambdaParser、DynamicExpresso、Z.Expressions,实现动态解析/求值字符串表达式
|
3月前
|
C#
C# WPF 将第三方DLL嵌入 exe
C# WPF 将第三方DLL嵌入 exe
82 0
|
5月前
|
消息中间件 中间件 程序员
分布式事务大揭秘:使用MQ实现最终一致性
本文由小米分享,介绍分布式事务中的MQ最终一致性实现,以RocketMQ为例。RocketMQ的事务消息机制包括准备消息、本地事务执行、确认/回滚消息及事务状态检查四个步骤。这种机制通过消息队列协调多系统操作,确保数据最终一致。MQ最终一致性具有系统解耦、提高可用性和灵活事务管理等优点,广泛应用于分布式系统中。文章还讨论了RocketMQ的事务消息处理流程和失败情况下的处理策略,帮助读者理解如何在实际应用中解决分布式事务问题。
409 6
|
5月前
|
存储 关系型数据库 Java
技术经验解读:三种分布式事务LCN、Seata、MQ
技术经验解读:三种分布式事务LCN、Seata、MQ
192 0
|
6月前
|
C#
C# 使用Socket对接
C# 使用Socket对接
40 1

热门文章

最新文章

下一篇
无影云桌面