2、创建服务总线,发送一个命令
static void Main(string[] args) { Console.WriteLine("Press 'Enter' to send a message.To exit, Ctrl + C"); var bus = BusCreator.CreateBus(); var sendToUri = new Uri($"{RabbitMqConstants.RabbitMqUri}{RabbitMqConstants.GreetingQueue}"); while (Console.ReadLine()!=null) { Task.Run(() => SendCommand(bus, sendToUri)).Wait(); } Console.ReadLine(); } private static async void SendCommand(IBusControl bus,Uri sendToUri) { var endPoint =await bus.GetSendEndpoint(sendToUri); var command = new GreetingCommand() { Id = Guid.NewGuid(), DateTime = DateTime.Now }; await endPoint.Send(command); Console.WriteLine($"send command:id={command.Id},{command.DateTime}"); }这一段代码隐藏了众多关于消息队列的细节,将我们的注意力集中在发送消息上,同时ServiceBus提供的API也更接近业务,我们虽然发送的是一个消息,但是在这种场景下体现出来是一个命令,Send(command)这一API描述了我们的意图。
3、服务端接收这一命令
新建一个命令台控制程序:Masstransit.RabbitMQ.GreetingServer
var bus = BusCreator.CreateBus((cfg, host) => { cfg.ReceiveEndpoint(host, RabbitMqConstants.GreetingQueue, e => { e.Consumer<GreetingConsumer>(); }); });这一代码可以理解为服务端在监听消息,我们在服务端注册了一个名为“GreetingConsumer”的消费者,GreetingConsumer的定义:
public class GreetingConsumer :IConsumer<GreetingCommand> { public async Task Consume(ConsumeContext<GreetingCommand> context) { await Console.Out.WriteLineAsync($"receive greeting commmand: {context.Message.Id},{context.Message.DateTime}"); } }该consumer可以消费类型为GreetingCommand的消息。这一实例几乎隐藏了有关RabbitMQ的技术细节,将代码中心放在了业务中,将这两个控制台应用跑起来试试:
五、实现Publish/Subscribe模式
发布/订阅模式使得基于消息传递的软件架构成为可能,这一能力表现为ClientA发送消息X,ClientB和ClientC都可以订阅消息X。
1、我们在上面的例子中改造一下,当GreetingConsumer收到GreetingCommand后发送一个GreetingEvent:
var greetingEvent = new GreetingEvent() { Id = context.Message.Id, DateTime = DateTime.Now }; await context.Publish(greetingEvent);2、新建控制台程序Masstransit.RabbitMQ.GreetingEvent.SubscriberA用来订阅GreetingEvent消息:
var bus = BusCreator.CreateBus((cfg, host) => { cfg.ReceiveEndpoint(host, RabbitMqConstants.GreetingEventSubscriberAQueue, e => { e.Consumer<GreetingEventConsumer>(); }); }); bus.Start();定义GreetingEventConsumer:
public class GreetingEventConsumer:IConsumer<Greeting.Message.GreetingEvent> { public async Task Consume(ConsumeContext<Greeting.Message.GreetingEvent> context) { await Console.Out.WriteLineAsync($"receive greeting event: id {context.Message.Id}"); } }这一代码跟Masstransit.RabbitMQ.GreetingServer接受一个命令几乎一模一样,唯一的区别在于:
在Send/Receive模式中Client首先要获得对方(Server)的终结点(endpoint),直接向该终结点发送命令。Server方监听自己的终结点并消费命令。
而Publish/Subscribe模式中Client publish一个事件,SubscriberA在自己的终结点(endpointA)监听事件,SubscriberB在自己的终结点(endpointB)监听事件。
3、根据上面的分析再定义一个Masstransit.RabbitMQ.GreetingEvent.SubscriberB
4、将4个控制台应用程序跑起来看看
六、实现RPC模式
这一模式在Masstransit中被称作Request/Response模式,通过IRequestClient<IRequest, IResponse> 接口来实现相关操作。一个相关的例子在官方的github。
结束语:本篇文章分析了如何使用Masstransit来抽象业务,避免直接使用具体的消息队列,当然本文提到的众多服务总线机制,如“重试、熔断等”并没有在该文中出现,需要大家进一步去了解该项目。
通过对Masstransit的一些试用和NServiceBus的对比,Masstransit在实际项目中很容易上手并且免费,各种API定义的也非常清晰,但是官方的文档有点过于简单,实际使用中还需要去做深入的研究。作为.NET平台下为数不多的ESB开源产品,其关注程度还是不够,期待大家为开源项目做出贡献。
本文例子提供下载: