使用Masstransit的微服务架构：消费者能否作为发送者及双向通信实现

当然可以！MassTransit完全支持消费者在处理消息的同时，作为发送者发布或发送新的消息。你只需要在消费者类中注入 ISendEndpointProvider 或者 IPublishEndpoint，就能在消费逻辑里触发新的消息流转。

举个简单的代码示例：

public class OrderCompletedConsumer : IConsumer<OrderCompleted>
{
    private readonly IPublishEndpoint _publishEndpoint;

    // 构造函数注入发布端点
    public OrderCompletedConsumer(IPublishEndpoint publishEndpoint)
    {
        _publishEndpoint = publishEndpoint;
    }

    public async Task Consume(ConsumeContext<OrderCompleted> context)
    {
        // 处理OrderCompleted消息的核心逻辑
        Console.WriteLine($"完成订单处理：{context.Message.OrderId}");

        // 同时发布新消息，触发下游服务的业务流程
        await _publishEndpoint.Publish(new InventoryDeductionRequested
        {
            OrderId = context.Message.OrderId,
            ProductIds = context.Message.ProductIds
        });
    }
}

这里的消费者在处理完订单完成事件后，立刻发布了库存扣减请求，完美扮演了消费者+发送者的双重角色。

这刚好对应MassTransit原生支持的请求响应模式，完全适配你需要的“认证服务发请求，产品服务处理后回传数据”的场景。下面一步步给你讲具体实现：

1. 定义请求与响应的消息契约 #

建议在两个服务都能引用的共享类库中定义消息结构（避免重复编写不一致的代码）：

// 认证服务发给产品服务的请求消息，携带用户信息
public record GetProductsForUserRequest
{
    public Guid UserId { get; init; }
    public string UserRole { get; init; }
}

// 产品服务回传给认证服务的响应消息，携带产品数据
public record GetProductsForUserResponse
{
    public List<ProductDto> Products { get; init; } = new();
}

// 产品数据DTO
public record ProductDto
{
    public Guid Id { get; init; }
    public string Name { get; init; } = string.Empty;
    public decimal Price { get; init; }
}

2. 产品微服务配置：作为请求的处理方（返回响应） #

先编写处理请求的消费者，再配置MassTransit：

消费者代码 #

public class GetProductsForUserConsumer : IConsumer<GetProductsForUserRequest>
{
    private readonly IProductRepository _productRepository;

    public GetProductsForUserConsumer(IProductRepository productRepository)
    {
        _productRepository = productRepository;
    }

    public async Task Consume(ConsumeContext<GetProductsForUserRequest> context)
    {
        // 根据用户信息查询对应产品（替换成你的业务逻辑）
        var products = await _productRepository.GetProductsByUser(context.Message.UserId, context.Message.UserRole);
        
        // 将查询结果包装成响应，返回给请求方
        await context.RespondAsync(new GetProductsForUserResponse
        {
            Products = products.Select(p => new ProductDto
            {
                Id = p.Id,
                Name = p.Name,
                Price = p.Price
            }).ToList()
        });
    }
}

Startup.cs中的MassTransit配置 #

public void ConfigureServices(IServiceCollection services)
{
    services.AddMassTransit(x =>
    {
        // 注册请求消费者
        x.AddConsumer<GetProductsForUserConsumer>();

        x.UsingRabbitMq((context, cfg) =>
        {
            cfg.Host("rabbitmq://localhost", h =>
            {
                h.Username("guest");
                h.Password("guest");
            });

            // 配置消费者的接收端点
            cfg.ReceiveEndpoint("product-service-get-products", e =>
            {
                e.ConfigureConsumer<GetProductsForUserConsumer>(context);
            });
        });
    });

    services.AddMassTransitHostedService();
    // 其他服务配置（如仓储、MVC等）...
}

3. 认证微服务配置：作为请求的发起方（等待响应） #

配置MassTransit并注册请求客户端，然后在业务逻辑中发送请求：

Startup.cs中的MassTransit配置 #

public void ConfigureServices(IServiceCollection services)
{
    services.AddMassTransit(x =>
    {
        x.UsingRabbitMq((context, cfg) =>
        {
            cfg.Host("rabbitmq://localhost", h =>
            {
                h.Username("guest");
                h.Password("guest");
            });
        });
    });

    // 注册请求客户端，用于发送请求并接收响应
    services.AddRequestClient<GetProductsForUserRequest>();

    services.AddMassTransitHostedService();
    // 其他服务配置（JWT认证、MVC等）...
}

发送请求并接收响应的代码示例 #

比如在认证服务的API接口中：

[ApiController]
[Route("api/auth")]
public class AuthController : ControllerBase
{
    private readonly IRequestClient<GetProductsForUserRequest> _requestClient;

    public AuthController(IRequestClient<GetProductsForUserRequest> requestClient)
    {
        _requestClient = requestClient;
    }

    [HttpGet("user-products")]
    public async Task<IActionResult> GetUserProducts(Guid userId)
    {
        // 从JWT Claim中获取用户角色（根据你的实际认证逻辑调整）
        var userRole = User.Claims.First(c => c.Type == "role").Value;

        // 发送请求并等待响应
        var response = await _requestClient.GetResponse<GetProductsForUserResponse>(new GetProductsForUserRequest
        {
            UserId = userId,
            UserRole = userRole
        });

        // 返回产品数据给前端
        return Ok(response.Message.Products);
    }
}

这样就完整实现了双向通信：认证服务发起请求，产品服务处理后回传数据，整个流程基于MassTransit的请求响应机制，可靠性和易用性都有保障。

内容的提问来源于stack exchange，提问作者TAHA ALAMI IDRISSI