MassTransit自定义请求/响应机制下查询/命令拓扑配置问询——适配模块化单体架构的RabbitMQ需求

Let's work through your requirements step by step and adjust your MassTransit configuration to meet all your topology needs. First, let's recap the core goals we need to hit:

• Shared queue for same API instances: Any instance of API1 should process commands/queries sent by any API1 instance, using round-robin distribution.
• Isolation between different APIs: Even if API1 and API2 have the same consumer type, API2's consumers should never process commands/queries initiated by API1 (and vice versa).
• Fixed queues for commands/queries: Queues shouldn't multiply as you scale instances—just add more consumers to the existing queue.

Key Issues in Your Current Code #

Your existing code has a couple of gaps:

• You're hardcoding TestCommand for send/publish configurations instead of making it dynamic for any consumer's message type.
• The fanout exchange issue comes from not fully disabling MassTransit's default consume topology and not properly configuring direct exchanges for all message types.
• Queue names aren't fixed, which would lead to new queues per instance if not addressed.

Revised Configuration Method #

Here's the updated AddNonEventConsumer method with explanations for each change:

private static IRabbitMqBusFactoryConfigurator AddNonEventConsumer<TConsumer>(
    IRabbitMqBusFactoryConfigurator config, IRegistration context)
    where TConsumer : class, IConsumer
{
    // Get the unique identifier for the current API (uses the executable assembly name)
    var apiIdentifier = Assembly.GetEntryAssembly()?.GetName().Name 
        ?? throw new InvalidOperationException("Could not determine API identifier");

    // Extract the message type from the consumer's IConsumer<T> interface
    var consumerInterface = typeof(TConsumer).GetInterfaces()
        .FirstOrDefault(i => i.IsGenericType && i.GetGenericTypeDefinition() == typeof(IConsumer<>));
    
    if (consumerInterface == null)
    {
        throw new InvalidOperationException(
            $"Consumer {typeof(TConsumer).Name} does not implement IConsumer<T>");
    }
    var messageType = consumerInterface.GetGenericArguments()[0];
    var messageTypeName = messageType.FullName 
        ?? throw new InvalidOperationException("Could not get message type full name");

    // Fixed queue name: [API Identifier].[Message Type Name]
    // Ensures all instances of the same API share this queue
    var queueName = $"{apiIdentifier}.{messageType.Name}";

    // Configure receive endpoint with fixed queue
    config.ReceiveEndpoint(queueName, e =>
    {
        // Disable MassTransit's default consume topology to avoid fanout exchanges
        e.ConfigureConsumeTopology = false;

        // Register the consumer for this endpoint
        e.Consumer<TConsumer>(context);

        // Manually bind to a direct exchange for the message type
        e.Bind(messageTypeName, b =>
        {
            b.ExchangeType = ExchangeType.Direct;
            // Use the API's unique identifier as the routing key
            // Ensures only this API's queue receives messages sent with this routing key
            b.RoutingKey = apiIdentifier;
        });
    });

    // Dynamically configure send rules for the current message type
    // Avoids hardcoding specific command types like TestCommand
    var sendConfigMethod = typeof(RabbitMqBusFactoryConfiguratorExtensions)
        .GetMethod(nameof(RabbitMqBusFactoryConfiguratorExtensions.Send), 
            new[] { typeof(IRabbitMqBusFactoryConfigurator), typeof(Action<IRabbitMqSendTopologyConfigurator<>>) })
        .MakeGenericMethod(messageType);

    sendConfigMethod.Invoke(null, new object[]
    {
        config,
        (Action<IRabbitMqSendTopologyConfigurator<object>>)(sendTopology =>
        {
            sendTopology.UseRoutingKeyFormatter(_ => apiIdentifier);
            sendTopology.ExchangeType = ExchangeType.Direct;
        })
    });

    return config;
}

How This Meets Your Requirements #

Let's map each requirement to the code:

1. Shared queue for same API instances:

   • We use a fixed queue name ({apiIdentifier}.{messageType.Name}) so all instances of the same API connect to the same queue. RabbitMQ automatically uses round-robin distribution when multiple consumers are connected to a single queue.

2. Isolation between different APIs:

   • Each API uses its own unique apiIdentifier (from the assembly name) as the routing key. When sending a message, we set the routing key to the current API's identifier. The queue is only bound to the message exchange with this specific routing key, so messages from API1 will never reach API2's queue (and vice versa).

3. Fixed queues for commands/queries:

   • The queue name is static per API and message type. Scaling out just adds more consumers to the existing queue—no new queues are created automatically.

Additional Setup Notes #

• Register Consumers: In your ConfigureServices method, register all your non-event consumers and call this method for each:

  services.AddMassTransit(x =>
  {
      x.AddConsumer<ProCommandConsumer>();
      x.AddConsumer<CyclistQueryConsumer>();
      // Add other non-event consumers here
  
      x.UsingRabbitMq((context, cfg) =>
      {
          cfg.Host("your-rabbitmq-host");
  
          // Register each consumer with our custom topology
          cfg.AddNonEventConsumer<ProCommandConsumer>(context);
          cfg.AddNonEventConsumer<CyclistQueryConsumer>(context);
          // Register other consumers here
      });
  });
  

• Request/Response Usage: Use MassTransit's IRequestClient<T> to send commands/queries—this handles the response flow automatically, using temporary queues for responses (which is fine and doesn't conflict with your requirements).
• Event Handling: Since you mentioned events are out of scope, just ensure you use separate configuration for event consumers (with fanout exchanges and instance-specific queues) as needed.

内容的提问来源于stack exchange，提问作者Danioss