使用NServiceBus开发分布式应用

Posted 2020-06-09

转自：http://www.cnblogs.com/richieyang/p/5041883.html

系列主题：基于消息的软件架构模型演变

NServiceBus 是一个.Net平台下开源的消息服务框架，这类产品有时也被称作ESB(Enterprise Service Bus)——企业服务总线。
NServicebus官方地址：http://particular.net/
git: https://github.com/Particular/NServiceBus
NServiceBus原作者Udi Dahan，该产品最早于2006年发行了第一个版本，这是一个企业级的开源产品，企业开发需要购买License，参照：http://particular.net/licensing。

一、NServiceBus的特性

1、高性能和可扩展性

可以广泛应用于许多业务领域，可扩展性和性能都经过了实战检验。

2、具有自动重试的可靠性集成

通过配置机制提供基于消息通讯的的最佳实践方案，能够识别错误响应并自动重试。

3、工作流和后台任务调度

通过Saga来完成长时间运行的流程定义和管理功能，提供强大而灵活的工作流功能。

4、消息的集中审核流程

很容易将整个分布式系统聚集到一个中心位置配置消息审核。

5、通过发布/订阅来减少耦合

提供了发布/订阅机制。可扩展、可配置、易于理解和易于使用。

6、易于扩展和配置

多个灵活的扩展点和配置选项，NServieBus可以根据用户需求对各个特性进行自定义配置。

7、支持广泛的消息传输技术

提供了MSMQ, RabbitMQ, SQL Server, Windows Azure Queues,Windows AzureService Bus消息传输机制，当然你也可以自定义或者选择由社区开发的消息传输方案。

二、.NET平台下其他ESB介绍

1、Biztalk

在微软的世界里，BizTalk Server一直被用来解决异构平台上应用程序之间数据交换的复杂集成问题。BizTalk同样提供了发布/订阅模式实现松耦合的架构。有时候你需要将现有的代码和一个运行在不同技术和协议下的历史遗留程序集成，这是一个经典的企业应用程序集成(Enterprise Application Integration-EAI)的场景。在这种场景之下，可以在业务服务之间使用NServiceBus，在这些服务的边界之内，你可以使用BizTalk与现有的历史遗留应用进行集成。

如您所见，服务边界之后的BizTalk是一个对异构应用的整合。

2、MassTransit

MassTransit是一个.NET平台下用来创建分布式应用程序的轻量级开源消息总线。

官网：http://masstransit-project.com/

git: https://github.com/MassTransit/MassTransit

MassTransit的第一个版本开发于2007年，作者Chris Patterson 和Dru Sellers 在一个会议中偶然相识，他们觉得当时.Net平台下没有一个他们想要的服务总线框架，而那时NServiceBus也刚刚发布，很多功能都不完善，并且也没有很好的社区支持。所以他俩开发了自己的ESB产品——MassTransit，目前最新的MassTransit基于NET4.0中的异步支持重写了所有代码。MassTransit的目标并不是要在分布式领域面面俱到从而适应大型的企业级开发，而是能实现一个强壮的轻量级消息总线。

三、从Hello World开始

分布式应用开发是一个比较复杂的过程，无论从涉及的技术知识体系还是开发，调试，部署都会带来很多挑战，我希望通过这个简单的例子展示分布式开发中的基本思想。

1、准备工作

安装MSMQ服务，NServiceBus默认使用MSMQ服务，所以在开始这个例子之前确保已安装MSMQ服务。

2、 新建一个类型为Console Application的客户端：NBus.Practice.GreetingClient，客户端会命令服务端输出“Hello World”。

安装nuget包：

1	Install-Package NServiceBus

3、 初始化一个Bus，然后给服务端发出命令。

NServiceBus提供了多个Host方案，应用程序自己Host或者使用NServiceBus.Host程序来Host应用程序。当然你还可以将NServicBus程序Host在一个Windows服务中。这个客户端我们选择Host在客户端自身当中。

我们只是用几个必要的选项来配置bus。重点是此段代码配置了一个EndPoint：“Nbus.Practice.HelloWorld.Client”，这个名称代表了了此客户端的网络地址。

有了这些配置，我们就可以利用这些配置来创建一个bus出来。

1 2 3 4

using (IBus bus = Bus.Create(busConfiguration).Start())             {                 SendGreetingCommand(bus);             }

接下来的代码通过bus.Send<TCommand>(TCommand cmd)方法给服务端发送一个命令：

1 2 3 4 5 6 7 8 9 10 11 12

private static void SendGreetingCommand(IBus bus) {     Console.WriteLine("Press ‘Enter‘ to send a message.To exit, Ctrl + C");     var i = 0;     while (Console.ReadLine() != null)     {         var id = Guid.NewGuid();         bus.Send(new GreetingCommand() { Id = id, Times = i });         i++;         Console.WriteLine("Send a new GreetingCommand message with id: {0}", id.ToString("N"));     } }

GreetingCommand就是一个消息，在消息总线中，一切交流都是通过消息来实现的。

4、建立一个公共的类库：NBus.Practice.GreetingMessage来定义消息
在NServiceBus中，命令类型要继承于ICommand，事件类型要继承于IEvent，消息是一个由属性构成的简单类型，最终需要序列化并可以在网络中传播。

1 2 3 4 5

public class GreetingCommand:ICommand {     public Guid Id { get; set; }     public int Times { get; set; } }

5、这个客户端几乎要完成了，我们创建了个bus，然后发送了一条消息。现在的问题在于这条息发送给谁呢？发送给谁这件事通过配置文件来完成。

1 2 3

<MessageEndpointMappings>   <add Messages="NBus.Practice.GreetingMessage"  Endpoint="Nbus.Practice.HelloWorld.Server" /> </MessageEndpointMappings>

这个配置的含义是：将定义在程序集NBus.Practice.GreetingMessage中的消息发送到EndPoint为Nbus.Practice.HelloWorld.Server的程序中。

6、显然我们需要一个EndPoint为Nbus.Practice.HelloWorld.Server的服务端，新建一个类型为Console Application类型的服务端：NBus.Practice.GreetingServer，并且用同样的方法创建一个bus并启动。

1 2 3 4 5 6 7 8 9 10 11 12 13

static void Main(string[] args) {     var busConfiguration = new BusConfiguration();     busConfiguration.EndpointName("Nbus.Practice.HelloWorld.Server");     busConfiguration.UseSerialization<JsonSerializer>();     busConfiguration.UsePersistence<InMemoryPersistence>();       using (IBus bus = Bus.Create(busConfiguration).Start())     {         Console.WriteLine("Press any key to exit");         Console.ReadKey();     } }

为了处理GreetingCommand，新建一个GreetingHandler.cs的类，只需要继承IHandleMessages<GreetingCommand>即可表明该类会处理类型为GreetingCommand的消息。

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16

public class GreetingHandler:IHandleMessages<GreetingCommand> {     private readonly IBus _bus;       public GreetingHandler(IBus bus)     {         _bus = bus;     }       public void Handle(GreetingCommand message)     {         Console.WriteLine("Received greetingCommand:{0}, times:{1}, Hello world",message.Id,message.Times);           _bus.Publish(new GreetingEvent(){Id = message.Id,Times = message.Times});     } }

方法public void Handle(GreetingCommand message)描述了当收到GreetingCommand消息会输出了字符串“Hello World”。另外代码最后发布了一个类型为GreetingEvent的事件，所有对此事件感兴趣的订阅者都可以订阅此事件。在NServiceBus中发布一个事件采用bus.Publish<TEvent>(TEvent event)方法。你会注意到我们通过构造器注入的方式来获取bus实例。

也许你会很关心此服务端中的配置文件如何配置呢？此服务端收到了别的程序发送的消息，具体谁发送的他并不知情。另外发布了一个事件，但是具体谁来订阅该事件，作为事件的发布者并不知道，所以该项目的消息路由配置为空：

1	<MessageEndpointMappings></MessageEndpointMappings>

7、截至目前，我们已经完成了显示hello world的任务，让我们运行起来看看吧：

在项目配置中将NBus.Practice.GreetingClient和NBus.Practice.GreetingServer同时设置为启动项。CTRL+F5

运行结果：当我们在客户端中按"Enter"键，服务端会收到消息并输出“Hello World”。

此时如果我们关闭服务端，并在客户端中多敲几次回车键发送GreetingCommand消息会怎么样？让我么来模拟真实场景下服务端应用由于未知原因宕机，会出现什么情况呢？

此时如果我们打开MSMQ管理工具就会发现服务端未处理的消息存储在队列中，直到服务端再次上线重新处理这些消息，从而保证了分布式应用中数据的最终一致性。

8、刚才服务端处理完GreetingCommand并发布了GreetingEvent事件，我们接下来新建一个类型为Console Application的事件订阅者：NBus.Practice.GreetingSubscriber。

事件订阅者跟之前一样需要创建并启动一个bus。

9、新建一个Congratulation.cs来处理GreetingEvent消息：

1 2 3 4 5 6 7

public class Congratulation:IHandleMessages<GreetingEvent> {     public void Handle(GreetingEvent message)     {         Console.WriteLine("Received greeting event id:{0}, time{1}, congratulations, you have learned NServiceBus.",message.Id,message.Times);     } }

10、订阅者的消息路由如何配置?

1 2 3

<MessageEndpointMappings>    <add Messages="NBus.Practice.GreetingMessage" Endpoint="Nbus.Practice.HelloWorld.Server" /> </MessageEndpointMappings>

这句配置的含义是：订阅EndPoint为Nbus.Practice.HelloWorld.Server且消息定义在程序集为NBus.Practice.GreetingMessage中的消息。

由此可见根据程序的角色不同，配置文件的配置具有不同的含义。

11、在事件的发布/订阅模式中，订阅者可以是一个或多个，我们将新建第二个订阅者来展示此功能。新建一个类型为Class library的程序集: NBus.Practice.GreetingAnotherSubscriber。

这次的程序之所以要换成Class library是因为我们本次要使用NServiceBus.Host来Host此程序。

1	Install-Package NServiceBus.Host

NServiceBus会为项目自动添加一个EndpointConfig.cs文件，我们将在此文件中配置bus：

1 2 3 4 5 6 7 8 9

public class EndpointConfig : IConfigureThisEndpoint {     public void Customize(BusConfiguration configuration)     {         configuration.EndpointName("Nbus.Practice.HelloWorld.AnotherSubscriber");         configuration.UseSerialization<JsonSerializer>();         configuration.UsePersistence<InMemoryPersistence>();     } }

同时添加一个Handler来处理GreetingEvent消息。

1 2 3 4 5 6 7

public class GreetingLogger:IHandleMessages<GreetingEvent> {     public void Handle(GreetingEvent message)     {         Console.WriteLine("Received greeting event id:{0}, time{1}, I will log it.",message.Id,message.Times);     } }

通过下面的方式使用NServiceBus.Host.exe来Host本程序。

11、    最后把本教程中建立的四个程序全都跑起来看看效果：

四、总结

本文通过一个简单的实例展示了如何在服务总线中发送命令，如何使用发布/订阅基本思想来实现一个Hello world。这个例子很简单，但是隐隐之中展现了CQRS的基本思想：Client发送一个Command，DomainHandler收到Command后会调用Domain逻辑，此时Domain会发布领域事件，Query分支会订阅领域事件来更新Query数据库，同时还有缓存、搜索引擎、其他服务也会订阅此领域事件。各个服务之间构成了松耦合的分布式应用程序。

当然NServiceBus还有更多高级主题例如：持久化、Saga、单元测试、二级重试、依赖注入、负载均衡、更换消息队列等内容等着我们去一探究竟。

整个例子的代码：https://git.oschina.net/richieyangs/NServiceBusPractice，以后会将所有NServiceBus相关的例子放在这个项目中。