.NET中的MassTransit分布式应用框架详解_Asp.net

masstransit是一款优秀的分布式应用框架，可作为分布式应用的消息总线，也可以用作单体应用的事件总线。

引言

a free, open-source distributed application framework for .net.
一个免费、开源的.net 分布式应用框架。-- masstransit 官网

masstransit，直译公共交通，是由chris patterson开发的基于消息驱动的.net 分布式应用框架，其核心思想是借助消息来实现服务之间的松耦合异步通信，进而确保应用更高的可用性、可靠性和可扩展性。通过对消息模型的高度抽象，以及对主流的消息代理（包括rabbitmq、activemq、kafaka、azure service bus、amazon sqs等）的集成，大大简化了基于消息驱动的开发门槛，同时内置了连接管理、消息序列化和消费者生命周期管理，以及诸如重试、限流、断路器等异常处理机制，让开发者更好的专注于业务实现。
简而言之，masstransit实现了消息代理透明化。无需面向消息代理编程进行诸如连接管理、队列的申明和绑定等操作，即可轻松实现应用间消息的传递和消费。

快速体验

空口无凭，创建一个项目快速体验一下。

基于worker模板创建一个基础项目：dotnet new worker -n masstransit.demo
打开项目，添加nuget包：masstransit
定义订单创建事件消息契约：

using system;

namespace masstransit.demo
{
    public record ordercreatedevent
    {
        public guid orderid { get; set; }
    }
}

4.修改worker类，发送订单创建事件：

namespace masstransit.demo;

public class worker : backgroundservice
{
    readonly ibus _bus;//注册总线
    public worker(ibus bus)
    {
        _bus = bus;
    }
    protected override async task executeasync(cancellationtoken stoppingtoken)
    {
        while (!stoppingtoken.iscancellationrequested)
        {
            //模拟并发送订单创建事件
            await _bus.publish(new ordercreatedevent(guid.newguid()), stoppingtoken);
            await task.delay(1000, stoppingtoken);
        }
    }
}

5.仅需实现iconsumer<ordercreatedevent>泛型接口，即可实现消息的订阅：

public class ordercreatedeventconsumer: iconsumer<ordercreatedevent>
{
    private readonly ilogger<ordercreatedeventconsumer> _logger;
    public ordercreatedeventconsumer(ilogger<ordercreatedeventconsumer> logger)
    {
        _logger = logger;
    }
    public task consume(consumecontext<ordercreatedevent> context)
    {
        _logger.loginformation($"received order:{context.message.orderid}");
        return task.completedtask;
    }
}

6.注册服务：

using masstransit;
using masstransit.demo;

ihost host = host.createdefaultbuilder(args)
    .configureservices(services =>
    {
        services.addhostedservice<worker>();
        services.addmasstransit(configurator =>
        {
            //注册消费者
            configurator.addconsumer<ordercreatedeventconsumer>();
            //使用基于内存的消息路由传输
            configurator.usinginmemory((context, cfg) =>
            {
                cfg.configureendpoints(context);
            });
        });
    })
    .build();

await host.runasync();

7.运行项目，一个简单的进程内事件发布订阅的应用就完成了。

如果需要使用rabbitmq 消息代理进行消息传输，则仅需安装masstransit.rabbitmqnuget包，然后指定使用rabbitmq 传输消息即可。

using masstransit;
using masstransit.demo;

ihost host = host.createdefaultbuilder(args)
    .configureservices(services =>
    {
        services.addhostedservice<worker>();
        services.addmasstransit(configurator =>
        {
            configurator.addconsumer<ordercreatedeventconsumer>();
            
            // configurator.usinginmemory((context, cfg) =>
            // {
            //     cfg.configureendpoints(context);
            // });
            
            configurator.usingrabbitmq((context, cfg) =>
            {
                cfg.host(
                    host: "localhost",
                    port: 5672,
                    virtualhost: "/",
                    configure: hostconfig =>
                    {
                        hostconfig.username("guest");
                        hostconfig.password("guest");
                    });
                cfg.configureendpoints(context);
            });
        });
    })
    .build();

await host.runasync();

运行项目，masstransit会自动在指定的rabbitmq上创建一个类型为fanout的masstransit.demo.ordercreatedeventexchange和一个与ordercreatedevent同名的队列进行消息传输，如下图所示。

核心概念

masstranist 为了实现消息代理的透明化和应用间消息的高效传输，抽象了以下概念，其中消息流转流程如下图所示：

message：消息契约，定义了消息生产者和消息消费者之间的契约。
producer：生产者，发送消息的一方都可以称为生产者。
sendendpoint：发送端点，用于将消息内容序列化，并发送到传输模块。
transport：传输模块，消息代理透明化的核心，用于和消息代理通信，负责发送和接收消息。
receiveendpoint：接收端点，用于从传输模块接收消息，反序列化消息内容，并将消息路由到消费者。
consumer：消费者，用于消息消费。

从上图可知，本质上还是发布订阅模式的实现，接下来就核心概念进行详解。

message

message：消息，可以使用class、interface、struct和record来创建，消息作为一个契约，需确保创建后不能篡改，因此应只保留只读属性且不应包含方法和行为。masstransit使用的是包含命名空间的完全限定名即typeof(t).fullname来表示特定的消息类型。因此若在另外的项目中消费同名的消息类型，需确保消息的命名空间相同。另外需注意消息不应继承，以避免发送基类消息类型造成的不可预期的结果。为避免此类情况，官方建议使用接口来定义消息。在masstransit中，消息主要分为两种类型：

command：命令，用于告诉服务做什么，命令被发送到指定端点，仅被一个服务接收并执行。一般以动名词结构命名，如：updateaddress、cancelorder。
event：事件，用于告诉服务什么发生了，事件被发布到多个端点，可以被多个服务消费。一般以过去式结构命名，如：addressupdated，ordercanceled。

经过masstransit发送的消息，会使用信封包装，包含一些附加信息，数据结构举例如下：

{
    "messageid": "6c600000-873b-00ff-9a8f-08da8da85542",
    "requestid": null,
    "correlationid": null,
    "conversationid": "6c600000-873b-00ff-9526-08da8da85544",
    "initiatorid": null,
    "sourceaddress": "rabbitmq://localhost/thinkpad_masstransitdemo_bus_ptoyyyr88cyx9s1gbdpe5kniy1?temporary=true",
    "destinationaddress": "rabbitmq://localhost/masstransit.demo:ordercreatedevent",
    "responseaddress": null,
    "faultaddress": null,
    "messagetype": [
        "urn:message:masstransit.demo:ordercreatedevent"
    ],
    "message": {
        "orderid": "fd8a3598-4c3a-4ec9-bbf9-d5f508e1a0d8"
    },
    "expirationtime": null,
    "senttime": "2022-09-03t12:32:15.0796943z",
    "headers": {},
    "host": {
        "machinename": "thinkpad",
        "processname": "masstransit.demo",
        "processid": 24684,
        "assembly": "masstransit.demo",
        "assemblyversion": "1.0.0.0",
        "frameworkversion": "6.0.5",
        "masstransitversion": "8.0.6.0",
        "operatingsystemversion": "microsoft windows nt 10.0.19044.0"
    }
}

从以上消息实例中可以看出一个包装后的消息包含以下核心属性：

messageid：全局唯一的消息id
messagetype：消息类型
message：消息体，也就是具体的消息实例
sourceaddress：消息来源地址
destinationaddress：消息目标地址
responseaddress：响应地址，在请求响应模式中使用
faultaddress：消息异常发送地址，用于存储异常消费消息
headers：消息头，允许应用自定义扩展信息
correlationid：关联id，在saga状态机中会用到，用来关联系列事件
host：宿主，消息来源应用的宿主信息

producer

producer，生产者，即用于生产消息。在masstransit主要借助以下对象进行命令的发送和事件的发布。

从以上类图可以看出，消息的发送主要核心依赖于两个接口：

isendendpoint：提供了send方法，用于发送命令。
ipublishendpoint：提供了publish方法，用于发布事件。

但基于上图的继承体系，可以看出通过ibus、isendendpointprovider和consumecontext进行命令的发送；通过ibus和ipublishendpointprovider进行事件的发布。具体举例如下：

发送命令

1.通过ibus发送：

private readonly ibus _bus;
public async task post(createorderrequest request)
{
    //通过以下方式配置对应消息类型的目标地址
    endpointconvention.map<createorderrequest>(new uri("queue:create-order"));
    await _bus.send(request);
}

2.通过isendendpointprovider发送：

private readonly isendendpointprovider  _sendendpointprovider;
public async task post(createorderrequest request)
{
    var serviceaddress = new uri("queue:create-order");
    var endpoint = await _sendendpointprovider.getsendendpoint(serviceaddress);
    await endpoint.send(request);
}

3.通过consumecontext发送：

public class createorderrequestconsumer:iconsumer<createorderrequest>
{    
    public async task consume(consumecontext<createorderrequest> context)
    {
    	//do something else
        var destinationaddress = new uri("queue:lock-stock");
        var command = new lockstockrequest(context.message.orderid);
       
        await context.send<lockstockrequest>(destinationaddress, command);
 		// 也可以通过获取`sendendpoint`发送命令
        // var endpoint = await context.getsendendpoint(destinationaddress);
        // await endpoint.send<lockstockrequest>(command);
    	
    }
}

发布事件

1.通过ibus发布：

private readonly ibus _bus;
public async task post(createorderrequest request)
{
    //do something
    await _bus.publish(request);
}

2.通过ipublishendpoint发布：

private readonly ipublishendpoint _publishendpoint;
public async task post(createorderrequest request)
{
    //do something
    var order = createorder(request);
    await _publishendpoint.publish<ordercreatedevent>(new ordercreateevent(order.id));
}

3.通过consumecontext发布：

public class createorderrequestconsumer: iconsumer<createorderrequest>
{    
    public async task consume(consumecontext<createorderrequest> context)
    {
        var order = createorder(conext.message);
    	await context.publish<ordercreatedevent>(new ordercreateevent(order.id));
    }
}

consumer

consumer，消费者，即用于消费消息。masstransit 包括多种消费者类型，主要分为无状态和有状态两种消费者类型。

无状态消费者

无状态消费者，即消费者无状态，消息消费完毕，消费者就释放。主要的消费者类型有：iconsumer<tmessage>、jobconsumer、iactivity和routingslip等。其中iconsumer<tmessage>已经在上面的快速体验部分举例说明。而jobconsumer<tmessage>主要是对iconsumer<tmessage>的补充，其主要应用场景在于执行耗时任务。
而对于iactivity和routingslip则是masstransit courier的核心对象，主要用于实现saga模式的分布式事务。masstransit courier 实现了routing slip模式，通过按需有序组合一系列的activity，得到一个用来限定消息处理顺序的routing slip。而每个activity的具体抽象就是iactivity和iexecuteactivity。二者的差别在于iactivity定义了execute和compensate两个方法，而iexecuteactivitiy仅定义了execute方法。其中execute代表正向操作，compensate代表反向补偿操作。用一个简单的下单流程：创建订单->扣减库存->支付订单举例而言，其示意图如下所示。而对于具体实现，可参阅文章：aspnetcore&masstransit courier实现分布式事务

有状态消费者

有状态消费者，即消费者有状态，其状态会持久化，代表的消费者类型为masstransitstatemachine。masstransitstatemachine是masstransit automatonymous 库定义的，automatonymous 是一个.net 状态机库，用于定义状态机，包括状态、事件和行为。masstransitstatemachine就是状态机的具体抽象，可以用其编排一系列事件来实现状态的流转，也可以用来实现saga模式的分布式事务。并支持与ef core和dapper集成将状态持久化到关系型数据库，也支持将状态持久化到mongodb、redis等数据库。masstransitstatemachine对于saga模式分布式事务的实现方式与routingslip不同，还是以简单的下单流程：创建订单->扣减库存->支付订单举例而言，其示意图如下所示。基于masstransitstatemachine 实现分布式事务详参后续文章。