首页 > 代码库 > 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 的命令模式实现）