MassTransit中Request&Response基本使用

Posted 2021-08-31 dotNET跨平台

MassTransit 是一个自由、开源、轻量级的消息总线基于.Net框架, 用于创建分布式应用程序。方便搭建基于消息的松耦合异步通信的应用程序和服务。MassTransit 在现有消息传输上提供了一组广泛的功能, 从而使开发人员能够友好地使用基于消息的会话模式异步连接服务。基于消息的通信是实现面向服务的体系结构的可靠和可扩展的方式。

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

发布订阅模式

这种场景十分常见，发送一个消息(或事件)到消息队列中，有一个或是多个订阅方对预期的消息接收处理。

基于需要搭建了两个WebApi程序，用于模拟发送方和订阅方，其中的RabbitMQ已预先搭建好了，只在程序中引用包配置下即可。

<PackageReference Include="MassTransit" Version="7.2.0" />
<PackageReference Include="MassTransit.AspNetCore" Version="7.2.0" />
<PackageReference Include="MassTransit.RabbitMQ" Version="7.2.0" />

发布端配置

在Startup中增加MassTransit需要的服务及初始化配置。

• 对RabbitMQ的连接地址端口、虚拟主机、访问账号密码等系列配置。

• 对发送方需要发送的消息初始化一个请求客户端，配置请求信息及推送到MQ的地址。

services.AddMassTransit(x =>
{
    x.UsingRabbitMq((context, cfg) =>
    {
        cfg.Host(Configuration["RabbitmqConfig:HostIP"], ushort.Parse(Configuration["RabbitmqConfig:HostPort"]), Configuration["RabbitmqConfig:VirtualHost"], h =>
        {
            h.Username(Configuration["RabbitmqConfig:Username"]);
            h.Password(Configuration["RabbitmqConfig:Password"]);
        });
    });
    x.AddRequestClient<ValueEntered>(new Uri(GetServiceAddress("events-valueentered")));
});
services.AddMassTransitHostedService();

为了快速了解，使用Controller在Action中发起对MQ的消息推送

[ApiController]
[Route("[controller]")]
public class ValueController : ControllerBase
{
    readonly IPublishEndpoint _publishEndpoint;
    public ValueController(IPublishEndpoint publishEndpoint)
    {
        _publishEndpoint = publishEndpoint;
    }
    [HttpPost]
    public async Task<ActionResult> Post(string value)
    {
        await _publishEndpoint.Publish<ValueEntered>(new
        {
            Value = value
        });
        return Ok();
    }
}

订阅端配置

订阅端也创建一个WebApi应用，在Startup中增加MassTransit的服务，使用到的Nuget包和发布端一样。

• 对RabbitMQ的连接地址端口、虚拟主机、访问账号密码等系列配置。

• 为订阅端增加一个订阅处理的Handler，即如下的ValueEnteredEventConsumer

• 增加一个接受点，指定队列名称，即发送端发送的队列名称，设置该队列消费处理的Consumer，即ValueEnteredEventConsumer

services.AddMassTransit(x =>
{
    x.AddConsumer<ValueEnteredEventConsumer>();
    x.UsingRabbitMq((context, cfg) =>
    {
        cfg.Host(Configuration["RabbitmqConfig:HostIP"], ushort.Parse(Configuration["RabbitmqConfig:HostPort"]), Configuration["RabbitmqConfig:VirtualHost"], h =>
        {
            h.Username(Configuration["RabbitmqConfig:Username"]);
            h.Password(Configuration["RabbitmqConfig:Password"]);
        });
        cfg.ReceiveEndpoint("events-valueentered", e =>
        {
            e.ConfigureConsumer<ValueEnteredEventConsumer>(context);
        });
    });
});
services.AddMassTransitHostedService();

如此一来，通过Postman发送一个请求，经发布端发布一个消息到RabbitMQ，订阅端侦听消息，处理消息，一切都很熟悉。

请求响应模式

在发布订阅的基础上，改变以往的习惯，当发布一个消息后，等待订阅方的处理，并将消息推送回RabbitMQ，发送方接受到处理后的消息继续执行。

请求端

在Startup中新加上一个用于发送消息(CheckOrderStatus)的请求客户端及指定消息队列名称(为每一个消息创建一个单独的队列)。

x.AddRequestClient<CheckOrderStatus>(new Uri(GetServiceAddress("events-checkorderstatus")));

增加一个Controller及Action，来请求及获取处理结果(OrderStatusResult)。

[ApiController]
[Route("[controller]")]
public class OrderController : ControllerBase
{
    private readonly IRequestClient<CheckOrderStatus> _client;
    public OrderController(IRequestClient<CheckOrderStatus> client)
    {
        _client = client;
    }
    public async Task<OrderStatusResult> Get(string id)
    {
        var response = await _client.GetResponse<OrderStatusResult>(new { OrderId = id });
        return response.Message;
    }
}

响应端

同样在响应端Startup中对新的消息设置下消息侦听队列以及相应的Handler如下的ValueEnteredEventConsumer去消费消息并返回处理结果。

x.AddConsumer<CheckOrderStatusConsumer >();
x.UsingRabbitMq((context, cfg) =>
{
    // ...
    cfg.ReceiveEndpoint("events-checkorderstatus", e =>
    {
    e.ConfigureConsumer<CheckOrderStatusConsumer >(context);
    });
});

Consumer中获取请求参数，执行请求，返回执行结果。

public class CheckOrderStatusConsumer : IConsumer<CheckOrderStatus>
{
    public async Task Consume(ConsumeContext<CheckOrderStatus> context)
    {
        if (context.Message.OrderId == "9527")
        {
            throw new InvalidOperationException("Order not found");
        }
        Console.WriteLine($"OrderId:{context.Message.OrderId}");
        await context.RespondAsync<OrderStatusResult>(new
        {
            OrderId = context.Message.OrderId,
            Timestamp = Guid.NewGuid().ToString(),
            StatusCode = "1",
            StatusText = "Close"
        });
    }
}

这样一来，当请求端发起一个消息(事件)到RabbitMQ，响应端侦听并处理完毕返回处理结果到RabbitMQ，请求端依照响应结果继续执行后续请求。

HTTP方式差异

与以往的Http请求方式有所不同，通过httpClient.PostAsync发送请求，接收端处理并返回结果，而走requestClient发送请求到RabbitMQ，再由RabbitMQ推送到侦听节点消费并返回结果，如下第一二部分结构。