重温.NET下Assembly的加载过程 ASP.NET Core Web API下事件驱动型架构的实现(三):基于RabbitMQ的事件总线

重温.NET下Assembly的加载过程

 

最近在工作中牵涉到了.NET下的一个古老的问题:Assembly的加载过程。虽然网上有很多文章介绍这部分内容,很多文章也是很久以前就已经出现了,但阅读之后发现,并没能解决我的问题,有些点写的不是特别详细,让人看完之后感觉还是云里雾里。最后,我决定重新复习一下这个经典而古老的问题,并将所得总结于此,然后会有一个实例对这个问题进行演示,希望能够帮助到大家。

.NET下Assembly的加载过程

.NET下Assembly的加载,最主要的一步就是确定Assembly的版本。在.NET下,托管的DLL和EXE都称之为Assembly,Assembly由AssemblyName来唯一标识,AssemblyName也就是大家所熟悉的Assembly.FullName,它是由五部分:名称、版本、语言、公钥Token、处理器架构组成的,这一点相信大家都知道。有关Assembly Name的详细描述,请参考:https://docs.microsoft.com/en-us/dotnet/framework/app-domains/assembly-names。那么版本,就是AssemblyName中的一个重要组成部分。其它四部分相同,版本如果不同的话,就不能算作是同一个Assembly。设计这样一个Assembly的版本策略,微软本身就是为了解决最开始的DLL Hell的问题,在*上着关于这段黑历史的详细描述,地址是:https://en.wikipedia.org/wiki/DLL_Hell,在此也就不多啰嗦了。

Assembly版本的重定向和最终确定

.NET下Assembly的加载过程,其实也是Assembly版本的确定和Assembly文件的定位过程,步骤如下:

  1. 在一个Assembly被编译的时候,它所引用的Assembly的全名(FullName)就会被编译器强行写入Assembly的Metadata,这个值是死的,从ILSpy可以看到,每个Reference都有它的全名信息:
    重温.NET下Assembly的加载过程     ASP.NET Core Web API下事件驱动型架构的实现(三):基于RabbitMQ的事件总线
    例如上图,System.Data依赖System.Xml,它所需要的版本是4.0.0.0,那么当CLR加载System.Data的时候,就可以暂且认为接下来需要加载的System.Xml版本是4.0.0.0。这里强调“暂且认为”,是因为这只是确定Assembly版本的第一步,那么最终System.Xml到底是不是使用4.0.0.0的版本呢?就需要看接下来这步的处理结果,也就是Assembly版本的重定向
  2. 首先,检查应用程序的配置文件,看是否存在Assembly版本重定向的设定。我们暂时先讨论应用程序配置文件就在AppDomain内的情况(如果在AppDomain之外,则需要首先下载配置文件,再继续,这里先不深入讨论)。应用程序配置文件常见的有.exe.config和web.config两种。在配置文件中,可以在runtime节点下的assemblyBinding中进行配置。例如:
    重温.NET下Assembly的加载过程     ASP.NET Core Web API下事件驱动型架构的实现(三):基于RabbitMQ的事件总线
    在这个例子中,asm6 Assembly的版本号被重定向到2.0.0.0。那么假设这就是asm6的最终版本号,那么接下来当CLR开始加载asm6的时候,如果2.0.0.0的版本没有找到,则直接抛出FileLoadException(即使3.0.0.0的版本是存在的),整个Assembly加载过程结束。FileLoadException的详细信息类似于:Could not load file or assembly 'asm6, Version=3.0.0.0, Culture=neutral, PublicKeyToken=c0305c36380ba429' or one of its dependencies. The located assembly's manifest definition does not match the assembly reference
  3. 如果在配置文件中找到了对应的版本重定向设定,那么,再接着查看Publisher Policy文件。Publisher Policy文件是一个仅包含配置文件的.NET Assembly,被安装到GAC里。它的Assembly版本重定向配置内容跟上面的应用程序配置文件的配置内容相同,不同的是,它的作用域是所有使用了该Assembly的应用程序。这种做法对于开发系统级通用框架的Assembly升级非常有用,比如.NET Framework。下面就是安装在GAC里的Publisher Policy文件的样本,需要注意:Publisher Policy会override应用程序配置信息中的版本重定向配置,而不是相反。换言之,假如asm6在上面这一步被确定为2.0.0.0,而所对应的Publisher Policy文件又将其确定为2.5.0.0,那么,暂且认为,CLR应该要加载2.5.0.0的版本。同理,“暂且认为”这个词表示,版本确定的过程还未结束
    重温.NET下Assembly的加载过程     ASP.NET Core Web API下事件驱动型架构的实现(三):基于RabbitMQ的事件总线
  4. 接下来,查找machine.config文件。同理,如果machine.config文件中存在版本重定向的设定,那么就会使用machine.config文件中的这个值,作为CLR应该去加载的Assembly的版本

至此,Assembly的最终版本已被确定,接下来就是搜索Assembly文件并进行加载的过程了。

Assembly文件的搜索和加载过程

现在,CLR已经开始加载确定版本的Assembly了,接下来就是搜索Assembly文件的过程。这个过程也叫作Assembly Probing。CLR会做以下事情:

  1. 首先,查看所需的Assembly是否已经加载过,如果已经加载了,那就直接使用那个已经加载的Assembly的版本与当前所需的版本进行比对,如果匹配,则使用那个已经加载的Assembly,如果不匹配,则抛出FileLoadException,执行结束
  2. 然后,看Assembly是否已被强签名(Strongly Named),如果是,则去GAC里查找Assembly。如果找到,则直接加载,整个Assembly加载过程结束。如果没有找到,那么就进行下一步,继续搜索Assembly文件。当然,如果Assembly没有进行强签名,那么就跳过这一步,直接继续
  3. 接着,CLR开始搜索(Probing)可能的Assembly位置,这又要分多种情况:
    1. 首先,查看文件中是否有指定<codeBase>,codeBase配置允许应用程序针对Assembly的不同版本指定装载地址,遵循如下规律:
      1. 如果所指定的Assembly文件位于当前应用程序域的启动目录(或其子目录)下,则使用相对路径指定href的值
      2. 如果所指定的Assembly文件位于其它目录,或任何其它地方,则href必须给出全路径,并且Assembly必须强签名的
    2. 然后,CLR对应用程序域的根目录以及相关的子目录进行探索:
      1. 假设Assembly的名字是abc.dll,那么CLR会探索以下目录:
        1. [appdomain_base]\abc.dll
        2. [appdomain_base]\abc\abc.dll
      2. 假设abc.dll还有语言设置(culture不是neutral),那么CLR会探索以下目录:
        1. [appdomain_base]\[culture]\abc.dll
        2. [appdomain_base]\[culture]\abc\abc.dll
    3. 如果找到符合版本的Assembly,则加载,否则进入下一步
  4. 最后,CLR会查看应用程序配置文件中是否有<probling>节点,如果有,则按probling节点所指定的privatePath值进行逐一探索。这个过程也会考虑culture的因素,类似于上面这步这样,对相应的子目录进行搜索。如果找到对应的Assembly,则加载,否则抛出FileLoadException,整个加载过程结束。注意,这里“逐一探索”的过程,不是遍历并找最佳匹配的过程。CLR仅根据Assembly的名字(不带版本号的名字)在privatePath下查找Assembly的文件,找到第一个名字匹配但是版本不匹配的话,就抛异常并终止加载了,它不会继续搜索privatePath中余下的其它路径

在加载Assembly文件失败的时候,AppDomain会触发AssemblyResolve的事件,在这个事件的订阅函数中,允许客户程序自定义对加载失败的Assembly的处理方式,比如,可以通过Assembly.LoadFrom或者Assembly.LoadFile调用“手动地”将Assembly加载到AppDomain。

fuslogvw Assembly绑定日志查看器

在.NET SDK中带了一个fuslogvw.exe的应用程序,通过它可以查看详细的Assembly加载过程。使用方法非常简单,使用管理员身份启动Visual Studio 2017 Developer Command Prompt,然后在命令行输入fuslogvw.exe,即可启动日志查看器。启动之后,点击Settings按钮,以启用日志记录功能:

重温.NET下Assembly的加载过程     ASP.NET Core Web API下事件驱动型架构的实现(三):基于RabbitMQ的事件总线

日志启动之后,点击Refresh按钮,然后启动你的.NET应用程序,就可以看到当前应用程序所依赖的Assembly的加载过程日志了:

重温.NET下Assembly的加载过程     ASP.NET Core Web API下事件驱动型架构的实现(三):基于RabbitMQ的事件总线

接下来,我会做一个例子程序,然后使用这个工具来分析Assembly的加载过程。

插件系统的实现与Assembly加载过程的分析

理论结合实际,看看如何通过实际代码来诠释以上所述Assembly的加载过程。一个比较好的例子就是设计一个简单的插件系统,并通过观察系统加载插件的过程,来了解Assembly加载的来龙去脉。为了简单直观,我把这个插件系统称为PluginDemo。这个插件很简单,主体程序是一个控制台应用程序,然后我们实现两个插件:Earth和Mars,在不同的插件的Initialize方法中,会输出不同的字符串。

整个应用程序的项目结构如下:

重温.NET下Assembly的加载过程     ASP.NET Core Web API下事件驱动型架构的实现(三):基于RabbitMQ的事件总线

该插件系统包含4个C#的项目:

  • PluginDemo.Common:它定义了AddIn抽象类,所有的插件实现都需要继承于这个抽象类。此外,AddInDefinition类是一个用来保存插件Metadata的类。为了演示,插件的Metadata仅仅包含插件类型的Assembly Qualified Name
  • PluginDemo.App:插件系统的应用程序。这个程序执行的时候,会扫描程序目录下Modules目录中的DLL,并根据module.xml的Metadata信息,加载相应的插件对象,并执行Initialize方法
  • PluginDemo.Plugins.Earth:其中的一个插件实现
  • PluginDemo.Plugins.Mars:另一个插件实现

注意:除了PluginDemo.Common之外的其它三个项目,都对PluginDemo.Common有引用关系。而PluginDemo.App项目仅仅在项目本身依赖于PluginDemo.Plugins.Earth和PluginDemo.Plugins.Mars,它不会去引用这两个项目。目的就是为了当PluginDemo.App被编译时,其余两个插件项目也会同时被编译并输出到指定位置。

在Earth插件的CustomAddIn类中,我们实现了Initialize方法,并在此输出一个字符串:

1
2
3
4
5
6
7
8
9
public class CustomAddIn : AddIn
{
    public override string Name => "Earth AddIn";
 
    public override void Initialize()
    {
        Console.WriteLine("Earth Plugin initialized.");
    }
}

在Mars插件的CustomAddIn类中,我们也实现了Initialize方法,并在此输出一个字符串:

1
2
3
4
5
6
7
8
9
public class CustomAddIn : AddIn
{
    public override string Name => "Mars AddIn";
 
    public override void Initialize()
    {
        Console.WriteLine("Mars AddIn initialized.");
    }
}

那么,在插件系统主程序中,就会扫描Modules子目录下的module.xml文件,然后解析每个module.xml文件获得每个插件类的Assembly Qualified Name,然后通过Type.GetType方法获得插件类,进而创建实例、调用Initialize方法。代码如下:

1
2
3
4
5
6
7
8
9
10
11
12
static void Main()
{
    var directory = new DirectoryInfo("Modules");
    foreach(var file in directory.EnumerateFiles("module.xml", SearchOption.AllDirectories))
    {
        var addinDefinition = AddInDefinition.ReadFromFile(file.FullName);
        var addInType = Type.GetType(addinDefinition.FullName);
        var addIn = (AddIn)Activator.CreateInstance(addInType);
        Console.WriteLine($"{addIn.Id} - {addIn.Name}");
        addIn.Initialize();
    }
}

接下来,修改App.config文件,修改为:

1
2
3
4
5
6
7
8
<?xml version="1.0" encoding="utf-8" ?>
<configuration>
  <runtime>
    <assemblyBinding xmlns="urn:schemas-microsoft-com:asm.v1">
      <probing privatePath="Modules\Earth;Modules\Mars;" />
    </assemblyBinding>
  </runtime>
</configuration>

此时,运行程序,可以得到:

重温.NET下Assembly的加载过程     ASP.NET Core Web API下事件驱动型架构的实现(三):基于RabbitMQ的事件总线

目前没有什么问题。接下来,对两个AddIn分别做一些修改。让这两个AddIn依赖于不同版本的Newtonsoft.Json,比如,Earth依赖于7.0.0.0的版本,Mars依赖于6.0.0.0的版本,然后分别修改两个CustomAddIn的Initialize方法,在方法中各自调用一次JsonConvert.SerializeObject方法,以触发Newtonsoft.Json这个Assembly的加载。此时再次运行程序,你将看到下面的异常:

重温.NET下Assembly的加载过程     ASP.NET Core Web API下事件驱动型架构的实现(三):基于RabbitMQ的事件总线

现在,刷新fuslogvw.exe,找到Newtonsoft.Json的日志:

重温.NET下Assembly的加载过程     ASP.NET Core Web API下事件驱动型架构的实现(三):基于RabbitMQ的事件总线

双击打开日志,可以看到如下信息:

重温.NET下Assembly的加载过程     ASP.NET Core Web API下事件驱动型架构的实现(三):基于RabbitMQ的事件总线

从整个过程可以看出:

  1. PluginDemo.App.exe正在试图加载PluginDemo.Plugins.Mars Assembly
  2. PluginDemo.Plugins.Mars开始调用Newtonsoft.Json
  3. 扫描应用程序配置文件、Host配置文件以及machine.config文件,均无找到Newtonsoft.Json的重定向信息,此时,Newtonsoft.Json版本确定为6.0.0.0
  4. GAC扫描失败,继续查找文件
  5. 首先查找应用程序当前目录下有没有Newtonsoft.Json,以及Newtonsoft.Json子目录下有没有Newtonsoft.Json.dll,发现都没有,继续
  6. 然后,通过App.config中的probing的privatePath设定,首先查找Modules\Earth目录(因为这个目录放在privatePath的第一个),找到了一个叫做Newtonsoft.Json.dll的Assembly,于是,判断版本是否相同。结果,找到的是7.0.0.0,而它需要的却是6.0.0.0,版本不匹配,于是就抛出异常,退出程序

那么接下来,改一改App.config文件,将privatePath下的两个值换个位置呢?

重温.NET下Assembly的加载过程     ASP.NET Core Web API下事件驱动型架构的实现(三):基于RabbitMQ的事件总线

再试试:

重温.NET下Assembly的加载过程     ASP.NET Core Web API下事件驱动型架构的实现(三):基于RabbitMQ的事件总线

此时,Earth AddIn又出错了。那么,我们加上版本重定向的配置,指定当程序需要加载7.0.0.0版本的Newtonsoft.Json时,让它重定向到6.0.0.0的版本:

重温.NET下Assembly的加载过程     ASP.NET Core Web API下事件驱动型架构的实现(三):基于RabbitMQ的事件总线

再次执行,成功了:

重温.NET下Assembly的加载过程     ASP.NET Core Web API下事件驱动型架构的实现(三):基于RabbitMQ的事件总线

看看日志:

重温.NET下Assembly的加载过程     ASP.NET Core Web API下事件驱动型架构的实现(三):基于RabbitMQ的事件总线

版本已经被重定向到6.0.0.0,并且在Mars目录下找到了6.0.0.0的Newtonsoft.Json,加载成功了。

这个案例的源代码可以点击此处下载

总结

本文详细介绍了.NET下Assembly的版本确定和加载过程,最后给出了一个实例,对这个过程进行了演示。

 
 
 
--

ASP.NET Core Web API下事件驱动型架构的实现(三):基于RabbitMQ的事件总线

 

在上文中,我们讨论了事件处理器中对象生命周期的问题,在进入新的讨论之前,首先让我们总结一下,我们已经实现了哪些内容。下面的类图描述了我们已经实现的组件及其之间的关系,貌似系统已经变得越来越复杂了。

重温.NET下Assembly的加载过程     ASP.NET Core Web API下事件驱动型架构的实现(三):基于RabbitMQ的事件总线

其中绿色的部分就是上文中新实现的部分,包括一个简单的Event Store,一个事件处理器执行上下文的接口,以及一个基于ASP.NET Core依赖注入框架的执行上下文的实现。接下来,我们打算淘汰PassThroughEventBus,然后基于RabbitMQ实现一套新的事件总线。

事件总线的重构

根据前面的结论,事件总线的执行需要依赖于事件处理器执行上下文,也就是上面类图中PassThroughEventBus对于IEventHandlerExecutionContext的引用。更具体些,是在事件总线订阅某种类型的事件时,需要将事件处理器注册到IEventHandlerExecutionContext中。那么在实现RabbitMQ时,也会有着类似的设计需求,即RabbitMQEventBus也需要依赖IEventHandlerExecutionContext接口,以保证事件处理器生命周期的合理性。

为此,我们新建一个基类:BaseEventBus,并将这部分公共的代码提取出来,需要注意以下几点:

  1. 通过BaseEventBus的构造函数传入IEventHandlerExecutionContext实例,也就限定了所有子类的实现中,必须在构造函数中传入IEventHandlerExecutionContext实例,这对于框架的设计非常有利:在实现新的事件总线时,框架的使用者无需查看API文档,即可知道事件总线与IEventHandlerExecutionContext之间的关系,这符合SOLID原则中的Open/Closed Principle
  2. BaseEventBus的实现应该放在EdaSample.Common程序集中,更确切地说,它应该放在EdaSample.Common.Events命名空间下,因为它是属于框架级别的组件,并且不会依赖任何基础结构层的组件

BaseEventBus的代码如下:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
public abstract class BaseEventBus : IEventBus
{
    protected readonly IEventHandlerExecutionContext eventHandlerExecutionContext;
 
    protected BaseEventBus(IEventHandlerExecutionContext eventHandlerExecutionContext)
    {
        this.eventHandlerExecutionContext = eventHandlerExecutionContext;
    }
 
    public abstract Task PublishAsync<TEvent>(TEvent @event, CancellationToken cancellationToken = default) where TEvent : IEvent;
 
    public abstract void Subscribe<TEvent, TEventHandler>()
        where TEvent : IEvent
        where TEventHandler : IEventHandler<TEvent>;
     
    // Disposable接口实现代码省略
}

在上面的代码中,PublishAsync和Subscribe方法是抽象方法,以便子类根据不同的需要来实现。

接下来就是调整PassThroughEventBus,使其继承于BaseEventBus:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
public sealed class PassThroughEventBus : BaseEventBus
{
    private readonly EventQueue eventQueue = new EventQueue();
    private readonly ILogger logger;
 
    public PassThroughEventBus(IEventHandlerExecutionContext context,
        ILogger<PassThroughEventBus> logger)
        : base(context)
    {
        this.logger = logger;
        logger.LogInformation($"PassThroughEventBus构造函数调用完成。Hash Code:{this.GetHashCode()}.");
 
        eventQueue.EventPushed += EventQueue_EventPushed;
    }
 
    private async void EventQueue_EventPushed(object sender, EventProcessedEventArgs e)
        => await this.eventHandlerExecutionContext.HandleEventAsync(e.Event);
 
    public override Task PublishAsync<TEvent>(TEvent @event, CancellationToken cancellationToken = default)
    {
        return Task.Factory.StartNew(() => eventQueue.Push(@event));
    }
 
    public override void Subscribe<TEvent, TEventHandler>()
    {
        if (!this.eventHandlerExecutionContext.HandlerRegistered<TEvent, TEventHandler>())
        {
            this.eventHandlerExecutionContext.RegisterHandler<TEvent, TEventHandler>();
        }
    }
     
    // Disposable接口实现代码省略
}

代码都很简单,也就不多做说明了,接下来,我们开始实现RabbitMQEventBus。

RabbitMQEventBus的实现

首先需要新建一个.NET Standard 2.0的项目,使用.NET Standard 2.0的项目模板所创建的项目,可以同时被.NET Framework 4.6.1或者.NET Core 2.0的应用程序所引用。创建新的类库项目的目的,是因为RabbitMQEventBus的实现需要依赖RabbitMQ C#开发库这个外部引用。因此,为了保证框架核心的纯净和稳定,需要在新的类库项目中实现RabbitMQEventBus。

Note:对于RabbitMQ及其C#库的介绍,本文就不再涉及了,网上有很多资料和文档,博客园有很多朋友在这方面都有使用经验分享,RabbitMQ官方文档也写得非常详细,当然是英文版的,如果英语比较好的话,建议参考官方文档。

以下就是在EdaSample案例中,RabbitMQEventBus的实现,我们先读一读代码,再对这部分代码做些分析。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
public class RabbitMQEventBus : BaseEventBus
{
    private readonly IConnectionFactory connectionFactory;
    private readonly IConnection connection;
    private readonly IModel channel;
    private readonly string exchangeName;
    private readonly string exchangeType;
    private readonly string queueName;
    private readonly bool autoAck;
    private readonly ILogger logger;
    private bool disposed;
 
    public RabbitMQEventBus(IConnectionFactory connectionFactory,
        ILogger<RabbitMQEventBus> logger,
        IEventHandlerExecutionContext context,
        string exchangeName,
        string exchangeType = ExchangeType.Fanout,
        string queueName = null,
        bool autoAck = false)
        : base(context)
    {
        this.connectionFactory = connectionFactory;
        this.logger = logger;
        this.connection = this.connectionFactory.CreateConnection();
        this.channel = this.connection.CreateModel();
        this.exchangeType = exchangeType;
        this.exchangeName = exchangeName;
        this.autoAck = autoAck;
 
        this.channel.ExchangeDeclare(this.exchangeName, this.exchangeType);
 
        this.queueName = this.InitializeEventConsumer(queueName);
 
        logger.LogInformation($"RabbitMQEventBus构造函数调用完成。Hash Code:{this.GetHashCode()}.");
    }
 
    public override Task PublishAsync<TEvent>(TEvent @event, CancellationToken cancellationToken = default(CancellationToken))
    {
        var json = JsonConvert.SerializeObject(@event, new JsonSerializerSettings { TypeNameHandling = TypeNameHandling.All });
        var eventBody = Encoding.UTF8.GetBytes(json);
        channel.BasicPublish(this.exchangeName,
            @event.GetType().FullName,
            null,
            eventBody);
        return Task.CompletedTask;
    }
 
    public override void Subscribe<TEvent, TEventHandler>()
    {
        if (!this.eventHandlerExecutionContext.HandlerRegistered<TEvent, TEventHandler>())
        {
            this.eventHandlerExecutionContext.RegisterHandler<TEvent, TEventHandler>();
            this.channel.QueueBind(this.queueName, this.exchangeName, typeof(TEvent).FullName);
        }
    }
 
    protected override void Dispose(bool disposing)
    {
        if (!disposed)
        {
            if (disposing)
            {
                this.channel.Dispose();
                this.connection.Dispose();
 
                logger.LogInformation($"RabbitMQEventBus已经被Dispose。Hash Code:{this.GetHashCode()}.");
            }
 
            disposed = true;
            base.Dispose(disposing);
        }
    }
 
    private string InitializeEventConsumer(string queue)
    {
        var localQueueName = queue;
        if (string.IsNullOrEmpty(localQueueName))
        {
            localQueueName = this.channel.QueueDeclare().QueueName;
        }
        else
        {
            this.channel.QueueDeclare(localQueueName, true, false, false, null);
        }
 
        var consumer = new EventingBasicConsumer(this.channel);
        consumer.Received += async (model, eventArgument) =>
        {
            var eventBody = eventArgument.Body;
            var json = Encoding.UTF8.GetString(eventBody);
            var @event = (IEvent)JsonConvert.DeserializeObject(json, new JsonSerializerSettings { TypeNameHandling = TypeNameHandling.All });
            await this.eventHandlerExecutionContext.HandleEventAsync(@event);
            if (!autoAck)
            {
                channel.BasicAck(eventArgument.DeliveryTag, false);
            }
        };
 
        this.channel.BasicConsume(localQueueName, autoAck: this.autoAck, consumer: consumer);
 
        return localQueueName;
    }
}

阅读上面的代码,需要注意以下几点:

  1. 正如上面所述,构造函数需要接受IEventHandlerExecutionContext对象,并通过构造函数的base调用,将该对象传递给基类
  2. 构造函数中,queueName参数是可选参数,也就是说:
    1. 如果通过RabbitMQEventBus发送事件消息,则无需指定queueName参数,仅需指定exchangeName即可,因为在RabbitMQ中,消息的发布方无需知道消息是发送到哪个队列中
    2. 如果通过RabbitMQEventBus接收事件消息,那么也分两种情况:
      1. 如果两个进程在使用RabbitMQEventBus时,同时指定了queueName参数,并且queueName的值相同,那么这两个进程将会轮流处理路由至queueName队列的消息
      2. 如果两个进程在使用RabbitMQEventBus时,同时指定了queueName参数,但queueName的值不相同,或者都没有指定queueName参数,那么这两个进程将会同时处理路由至queueName队列的消息
    3. 有关Exchange和Queue的概念,请参考RabbitMQ的官方文档
  3. 在Subscribe方法中,除了将事件处理器注册到事件处理器执行上下文之外,还通过QueueBind方法,将指定的队列绑定到Exchange上
  4. 事件数据都通过Newtonsoft.Json进行序列化和反序列化,使用TypeNameHandling.All这一设定,使得序列化的JSON字符串中带有类型名称信息。在此处这样做既是合理的,又是必须的,因为如果没有带上类型名称的信息,JsonConvert.DeserializeObject反序列化时,将无法判定得到的对象是否可以转换为IEvent对象,这样就会出现异常。但如果是实现一个更为通用的消息系统,应用程序派发出去的事件消息可能还会被由Python或者Java所实现的应用程序所使用,那么对于这些应用,它们并不知道Newtonsoft.Json是什么,也无法通过Newtonsoft.Json加入的类型名称来获知事件消息的初衷(Intent),Newtonsoft.Json所带的类型信息又会显得冗余。因此,简单地使用Newtonsoft.Json作为事件消息的序列化、反序列化工具,其实是欠妥的。更好的做法是,实现自定义的消息序列化、反序列化器,在进行序列化的时候,将.NET相关的诸如类型信息等,作为Metadata(元数据)附着在序列化的内容上。理论上说,在序列化的数据中加上一些元数据信息是合理的,只不过我们对这些元数据做一些标注,表明它是由.NET框架产生的,第三方系统如果不关心这些信息,可以对元数据不做任何处理
  5. 在Dispose方法中,注意将RabbitMQ所使用的资源dispose掉

使用RabbitMQEventBus

在Customer服务中,使用RabbitMQEventBus就非常简单了,只需要引用RabbitMQEventBus的程序集,然后在Startup.cs文件的ConfigureServices方法中,替换PassThroughEventBus的使用即可:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
public void ConfigureServices(IServiceCollection services)
{
    this.logger.LogInformation("正在对服务进行配置...");
 
    services.AddMvc();
 
    services.AddTransient<IEventStore>(serviceProvider =>
        new DapperEventStore(Configuration["mssql:connectionString"],
            serviceProvider.GetRequiredService<ILogger<DapperEventStore>>()));
 
    var eventHandlerExecutionContext = new EventHandlerExecutionContext(services,
        sc => sc.BuildServiceProvider());
    services.AddSingleton<IEventHandlerExecutionContext>(eventHandlerExecutionContext);
    // services.AddSingleton<IEventBus, PassThroughEventBus>();
 
    var connectionFactory = new ConnectionFactory { HostName = "localhost" };
    services.AddSingleton<IEventBus>(sp => new RabbitMQEventBus(connectionFactory,
        sp.GetRequiredService<ILogger<RabbitMQEventBus>>(),
        sp.GetRequiredService<IEventHandlerExecutionContext>(),
        RMQ_EXCHANGE,
        queueName: RMQ_QUEUE));
 
    this.logger.LogInformation("服务配置完成,已注册到IoC容器!");
}

Note:一种更好的做法是通过配置文件来配置IoC容器,在曾经的Microsoft Patterns and Practices Enterprise Library Unity Container中,使用配置文件是很方便的。这样只需要Customer服务能够通过配置文件来配置IoC容器,同时只需要让Customer服务依赖(注意,不是程序集引用)于不同的事件总线的实现即可,无需对Customer服务重新编译。

下面来验证一下效果。首先确保RabbitMQ已经配置并启动妥当,我是安装在本地机器上,使用默认安装。首先启动ASP.NET Core Web API,然后通过Powershell发起两次创建Customer的请求:

重温.NET下Assembly的加载过程     ASP.NET Core Web API下事件驱动型架构的实现(三):基于RabbitMQ的事件总线

查看一下数据库是否更新正常:

重温.NET下Assembly的加载过程     ASP.NET Core Web API下事件驱动型架构的实现(三):基于RabbitMQ的事件总线

并检查一下日志信息:

重温.NET下Assembly的加载过程     ASP.NET Core Web API下事件驱动型架构的实现(三):基于RabbitMQ的事件总线

RabbitMQ中Exchange的信息:

重温.NET下Assembly的加载过程     ASP.NET Core Web API下事件驱动型架构的实现(三):基于RabbitMQ的事件总线

总结

本文提供了一种RabbitMQEventBus的实现,目前来说是够用的,而且这种实现是可以使用在实际项目当中的。在实际使用中,或许也会碰到一些与RabbitMQ本身有关的问题,这就需要具体问题具体分析了。此外,本文没有涉及事件消息丢失、重发然后保证最终一致性的问题,这些内容会在后面讨论。从下文开始,我们着手逐步实现CQRS架构的领域事件和事件存储部分。

源代码的使用

本系列文章的源代码在https://github.com/daxnet/edasample这个Github Repo里,通过不同的release tag来区分针对不同章节的源代码。本文的源代码请参考chapter_3这个tag,如下:

重温.NET下Assembly的加载过程     ASP.NET Core Web API下事件驱动型架构的实现(三):基于RabbitMQ的事件总线

欢迎访问我的博客新站:http://sunnycoding.net

 

上一篇:net.sf.json.JSONObject maven下载到了但是java后台一直用不了问题


下一篇:关于maven包的引入net.sf.json的问题