首页 > 代码库 > Github开源:Sheng.RabbitMQ.CommandExecuter (RabbitMQ 的命令模式实现)

Github开源:Sheng.RabbitMQ.CommandExecuter (RabbitMQ 的命令模式实现)

[Github]:https://github.com/iccb1013/Sheng.RabbitMQ.CommandExecuter

Sheng.RabbitMQ.CommandExecuter 是使用 .Net 对 RabbitMQ 的一个简单封装。

它通过XML配置文件定义Exchange及队列等信息,根据此配置文件自动声明及初始化相关队列信息,方便 .Net 开发人员使用 RabbitMQ。

并实现了一个基于 MQ 的命令执行器,将 MQ 消息抽象化为命令,发布端和订阅端通过命令进行交互。默认实现了两个命令:
1)HTTP请求转发,将收到的MQ消息的指定内容转发到指定URL上;
2)数据库同步,通过预先定义的配置文件,指明不同数据库和表之间的关联关系,发送端向 MQ 中发布数据库同步命令后,订阅方(可作为 windows 服务部署,已在工程中实现)负责解析并执行数据库同步工作。

你可以直接使用基本的 RabbitMQ 封装,也可以在此命令模式的基础上实现你自己的命令。

 

技术分享

Sheng.RabbitMQ.CommandExecuter.Contract:命令模式中的命令契约,命令的发布方和接收方通过此契约类库共享已知类型。

Sheng.RabbitMQ.CommandExecuter.Core:命令模式的核心实现,命令的解释和执行在此实现,并默认实现了两个命令。

Sheng.RabbitMQ.CommandExecuter.RabbitMQ:RabbitMQ 的简单封装,通过XML配置文件初始化 RabbitMQ,可单独使用此工程方便在 .Net 工程中使用 RabbitMQ。

Sheng.RabbitMQ.CommandExecuter.Service:Windows 服务,可做为命令模式中的接收端。

Sheng.RabbitMQ.CommandExecuter.WindowsForm:没有实际功能,方便开发阶段调试程序。

 

RabbitMQConfig.xml 用于配置 RabbitMQ 信关信息:

<?xml version="1.0" encoding="utf-8" ?><rabbitMQ>  <connectionFactory hostName="192.168.100.100" userName="user" password="1234"></connectionFactory>  <exchangeList>    <exchange name="exchangeName_A" type="direct">      <queueList>        <queue name="queue_test_A" durable="true" exclusive="false" autoDelete="false" routingKey="routingKey_A" type="send"></queue>        <queue name="queue_test_B" durable="true" exclusive="false" autoDelete="false" routingKey="routingKey_B" type="receive"></queue>      </queueList>    </exchange>  </exchangeList></rabbitMQ>

 

 

相关属性与 RabbitMQ 名称意义皆相同,只是注意一点,queue 节点中的 type 有 send 和 receive 两种,send 表示此队列只用于发送消息,receive 表示此队列只用于接收消息。RabbitMQService 不会订阅 type 为 send 的队列的消息。

 

代码简要说明:

声明核心类:RabbitMQService ,然后通过此类订阅消息,发送消息。

RabbitMQService _rabbitMQService = RabbitMQService.Instance;

通过 Subscribe 方法订阅消息:

private void btnStartService_Click(object sender, EventArgs e)       {           _rabbitMQService.Subscribe("routingKey_B", RabbitMQCallback);            _rabbitMQService.Start();            MessageBox.Show("RabbitMQService 已启动。");       }        private void RabbitMQCallback(ulong deliveryTag, string routingKey, string body)       {           _rabbitMQService.Ack(deliveryTag, false);            Debug.WriteLine(routingKey + Environment.NewLine + body);            MessageBox.Show(routingKey + Environment.NewLine + body);       }

_rabbitMQService.Start() 用于启动 RabbitMQ 的监听。

通过 Send 方法发送消息:

_rabbitMQService.Send("exchangeName_A", "routingKey_B", "123");

 有关命令模式的实现说明:

在契约类库中,所有的命令均继承自 Command 类,并提供一个 CommandType 属性

public abstract class Command    {        public string CommandType        {            get;            protected set;        }    }

简单的 URL 转发命令的定义如下:

public class DatabaseSyncForwardCommand: Command   {       public const string CommandTypeName = "DatabaseSyncForwardCommand";        public DatabaseSyncForwardCommand()       {           base.CommandType = CommandTypeName;       }        public string Url       {           get;set;       }        public string CommandContent       {           get;set;       }   }

命令的解析和执行在 Sheng.RabbitMQ.CommandExecuter.Core 中

技术分享

CommandExecuterService 用于向调用者提供命令执行服务的整体封装,通过它启动命令的监听和执行。

CommandReceiver 目录下是命令的接收和执行程序,每个命令都有一个对应的 *Receiver 类,这些类通过 CommandReceiverFactory 这个工厂类统一实例化和调用。

 

有关数据库同步命令的说明

数据库同步命令可以实现简单的,多个数据库之间的表数据同步。可以定义多个消费者,并为每个消费者定义多个数据提供者,并定义它们之间的表结构对应关系。在文件 DatabaseSyncConfig.xml 中定义:

<?xml version="1.0" encoding="utf-8" ?><databaseSync>  <connectionList>    <connection name="connection1" connectionString=""/>    <connection name="connection2" connectionString=""/>  </connectionList>  <consumerList>    <consumer name="crm" connection="connection1">      <producerList>        <producer name="erp" routingKey="routingKey_A" connection="connection2">          <tableDefinition>            <table name="Customers" primaryKey="Id" consumerTable="Customers" consumerTablePrimaryKey="Id">              <Field name="Id" consumerField="Id" ></Field>              <Field name="CustomerName" consumerField="CustomerName"></Field>              <Field name="NopUserName" consumerField="NopUserName"></Field>            </table>          </tableDefinition>        </producer>      </producerList>    </consumer>  </consumerList></databaseSync>

connectionList 节点用于定义数据库连接,可以定义多个不同的数据库连接,并在下面的配置环节中引用。

consumerList 用于配置消费者,consumer 下的 producer 表示针对这一消费者的数据提供者。
tableDefinition 用于定义消费者和数据提供者之间的表结构对应关系。
相关程序在收到数据库同步命令时,通过数据提供者定义中的 routingKey 来判断数据同步命令来自哪个数据提供者。

对于数据的提供者,通过类似如下代码,向 MQ 中发送数据同步命令:

DatabaseSyncCommand cmd = new DatabaseSyncCommand(); DatabaseSyncItem item1 = new DatabaseSyncItem(){    Action = DatabaseSyncAction.Add,    Table = "Customers",    PrimaryKeyValue = "062B54F5-69AA-A108-09F8-39DB9C2F58C4"}; cmd.SyncItemList.Add(item1); string json = JsonConvert.SerializeObject(cmd); _rabbitMQService.Send("exchangeName_A", "routingKey_A", json);

以上。

QQ: 279060597    @南京

引用请注明原文出处: http://sheng.city/post/sheng-rabbitmq-commandexecuter

 

Github开源:Sheng.RabbitMQ.CommandExecuter (RabbitMQ 的命令模式实现)