Ubuntu下ASP.NET Core API与监控服务MediatR通信失败问题排查

MediatR 默认是进程内通信组件，你的API和监控服务是两个独立运行的进程，各自的MediatR实例只负责管理自身进程内的通知处理器（INotificationHandler）。API调用_mediator.Publish时，只会在API进程内部寻找对应的Handler，完全触及不到监控服务的进程，所以监控服务自然收不到通知。

要实现跨服务通知，必须引入专门的跨进程通信机制，下面以常用的RabbitMQ为例给出具体改造步骤：

1. 共享类库调整 #

保留Ping类（无需再实现INotification，因为不再依赖MediatR的进程内机制）：

public class Ping 
{
    // 可添加需要传递的用户登录信息，比如UserId、LoginTime等
    public string UserId { get; set; }
    public DateTime LoginTime { get; set; }
}

2. API服务改造 #

• 安装RabbitMQ客户端包，比如RabbitMQ.Client（基础包）或EasyNetQ（简化封装）
• 在Program.cs中注册RabbitMQ连接服务：

  builder.Services.AddSingleton<RabbitMQ.Client.IConnection>(sp =>
  {
      var factory = new RabbitMQ.Client.ConnectionFactory()
      {
          HostName = "localhost", // 你的RabbitMQ部署地址
          UserName = "guest",
          Password = "guest"
      };
      return factory.CreateConnection();
  });
  
  builder.Services.AddSingleton<RabbitMQ.Client.IModel>(sp =>
  {
      var connection = sp.GetRequiredService<RabbitMQ.Client.IConnection>();
      var channel = connection.CreateModel();
      // 声明Fanout类型交换机，适合广播通知场景
      channel.ExchangeDeclare(exchange: "login_notifications", type: ExchangeType.Fanout);
      return channel;
  });
  

• 在AuthController中替换MediatR发布逻辑，改为发送消息到RabbitMQ：

  private readonly RabbitMQ.Client.IModel _channel;
  
  public AuthController(RabbitMQ.Client.IModel channel)
  {
      _channel = channel;
  }
  
  [HttpPost("login")]
  public async Task<IActionResult> Login(LoginRequest request)
  {
      // 原有登录逻辑...
  
      // 发送登录通知
      var ping = new Ping { UserId = user.Id, LoginTime = DateTime.UtcNow };
      var messageBytes = System.Text.Json.JsonSerializer.SerializeToUtf8Bytes(ping);
      _channel.BasicPublish(
          exchange: "login_notifications",
          routingKey: "",
          basicProperties: null,
          body: messageBytes
      );
  
      return Ok();
  }
  

3. 监控服务改造 #

• 同样安装RabbitMQ客户端包
• 在Worker类中实现RabbitMQ消费者，监听通知：

  private readonly RabbitMQ.Client.IConnection _connection;
  private readonly RabbitMQ.Client.IModel _channel;
  private readonly ILogger<Worker> _logger;
  
  public Worker(ILogger<Worker> logger)
  {
      _logger = logger;
      var factory = new RabbitMQ.Client.ConnectionFactory()
      {
          HostName = "localhost",
          UserName = "guest",
          Password = "guest"
      };
      _connection = factory.CreateConnection();
      _channel = _connection.CreateModel();
      // 声明和API端一致的交换机
      _channel.ExchangeDeclare(exchange: "login_notifications", type: ExchangeType.Fanout);
      // 声明临时队列（服务重启后自动删除）
      var queueName = _channel.QueueDeclare().QueueName;
      // 绑定队列到交换机，接收广播消息
      _channel.QueueBind(queue: queueName, exchange: "login_notifications", routingKey: "");
  }
  
  protected override async Task ExecuteAsync(CancellationToken stoppingToken)
  {
      var consumer = new RabbitMQ.Client.EventingBasicConsumer(_channel);
      consumer.Received += (model, ea) =>
      {
          var body = ea.Body.ToArray();
          var ping = System.Text.Json.JsonSerializer.Deserialize<Ping>(body);
          // 执行原来的Handle逻辑
          _logger.LogInformation("收到用户登录通知：UserId={UserId}, LoginTime={LoginTime}", ping.UserId, ping.LoginTime);
          // 手动确认消息已处理，避免重复消费
          _channel.BasicAck(deliveryTag: ea.DeliveryTag, multiple: false);
      };
      _channel.BasicConsume(queue: queueName, autoAck: false, consumer: consumer);
  
      while (!stoppingToken.IsCancellationRequested)
      {
          await Task.Delay(1000, stoppingToken);
      }
  }
  
  public override void Dispose()
  {
      _channel.Close();
      _connection.Close();
      base.Dispose();
  }
  

• Redis Pub/Sub：轻量级，适合简单通知场景，使用StackExchange.Redis包实现发布订阅
• HTTP回调：API直接调用监控服务暴露的接口发送通知，实现简单但可靠性较低（依赖监控服务在线）
• gRPC：高性能的跨进程通信方案，适合频繁、低延迟的通知场景

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