MassTransit入门(一)

本文地址源码

• MassTransit是一个面向.net的免费开源分布式应用程序框架。
• MassTransit使得创建应用程序和服务变得很容易，这些应用程序和服务利用基于消息的、松散耦合的异步通信来获得更高的可用性、可靠性和可伸缩性。
• MassTransit 8.x版本。

实现简单发布订阅

• 添加Nuget包引用:
  • MassTransit
  • MassTransit.RabbitMQ(演示也可基于内存)

生产端

• 配置MassTransit

builder.Services.AddMassTransit(x =>{       // 使用内存    //x.UsingInMemory();    // 使用RabbitMq    x.UsingRabbitMq((context, config) =>    {        config.Host("rabbitmq://localhost:5672", host =>        {            host.Username("admin");            host.Password("admin");        });    });});

• 定义消息体

public class OrderEto{    public Guid Id { get; init; }    public string Name { get; set; }        public DateTime CreationTime { get; set; }}

• 发布消息

[ApiController][Route("[controller]")]public class PublishController : ControllerBase{    private readonly ILogger<PublishController> _logger;    private readonly IPublishEndpoint _publishEndpoint;    public PublishController(ILogger<PublishController> logger, IPublishEndpoint publishEndpoint)    {        _logger = logger;        _publishEndpoint = publishEndpoint;    }    [HttpGet]    public async Task Get()    {        await _publishEndpoint.Publish<OrderEto>(new OrderEto()        {            Id = Guid.NewGuid(),            Name = "Phone",            CreationTime = DateTime.Now        });    }}

消费者端

builder.Services.AddMassTransit(x =>{        // 通过扫描程序集注册消费者    x.AddConsumers(typeof(Program).Assembly);       // 通过类型单个注册消费者    // x.AddConsumer<OrderEtoConsumer>(typeof(OrderEtoConsumerDefinition));        // x.SetKebabCaseEndpointNameFormatter();        // 通过泛型单个注册消费者    //x.AddConsumer<OrderEtoConsumer, OrderEtoConsumerDefinition>();        // 通过指定命名空间注册消费者    // x.AddConsumersFromNamespaceContaining<OrderEtoConsumer>();        // 使用内存队列    // x.UsingInMemory();    x.UsingRabbitMq((context, config) =>    {              config.Host("rabbitmq://localhost:5672", hostconfig =>        {            hostconfig.Username("admin");            hostconfig.Password("admin");        });                config.ConfigureEndpoints(context);           });});

• 消费者定义

public class OrderEtoConsumer : IConsumer<OrderEto>{    private readonly ILogger<OrderEtoConsumer> _logger;    public OrderEtoConsumer(ILogger<OrderEtoConsumer> logger)    {        _logger = logger;    }    public Task Consume(ConsumeContext<OrderEto> context)    {        _logger.LogInformation($"MassTransit.Consumer.One 收到消息:{JsonSerializer.Serialize(context.Message)}");        return Task.CompletedTask;    }}public class OrderEtoConsumerDefinition : ConsumerDefinition<OrderEtoConsumer>{    protected override void ConfigureConsumer(IReceiveEndpointConfigurator endpointConfigurator, IConsumerConfigurator<OrderEtoConsumer> consumerConfigurator)    {        endpointConfigurator.UseMessageRetry(r => r.Intervals(500, 1000));    }}

Abp Vnext Vue实现

• Github地址
• 文档地址
• 演示地址