ASP.NET Web API中结合RabbitMQ合并多队列消息实现关联邮件发送的问题

看起来你现在的核心痛点是如何把分别来自两个RabbitMQ队列的「文件URL消息」和「模板创建消息」关联起来，再一起处理发送带附件的邮件对吧？我先帮你捋下当前代码里的问题，再给你一套可行的解决方案。

当前代码的核心问题 #

你现在的ProcessMessageAsync方法尝试把每个消息同时反序列化成AlertMessage和FileUploadMessage，这显然行不通——每个队列的消息只会是其中一种类型，所以那个alertMessage != null && uploadMessage != null的判断逻辑永远不会成立，自然无法处理消息配对。另外，你也没有做消息的临时存储，两类消息是独立到达的，根本没法直接配对。

解决方案思路 #

要实现消息配对，关键是给两类消息加一个共同的关联标识，再配合缓存机制临时存储先到达的消息，等配对消息到来后再合并处理。具体步骤如下：

1. 给消息添加关联标识

不管是文件上传接口还是模板创建接口，都要让前端传递一个唯一的关联ID（比如TemplateId，因为最终是模板创建时要绑定对应的文件），后端生成消息时把这个ID包含进去。这样两类消息就能通过这个ID关联起来。

2. 引入缓存存储未配对消息

用IMemoryCache（单实例部署）或者分布式缓存（比如Redis，多实例部署时必须用）来临时存储先到达的消息。给缓存设置过期时间（比如30分钟），避免无效消息一直占用内存。

3. 修改消费与处理逻辑

根据队列名区分消息类型，收到消息后先尝试从缓存查找配对的另一半：

• 如果找到，就合并两类消息，调用邮件服务发送带附件的邮件，处理完成后删除缓存；
• 如果没找到，就把当前消息存入缓存，等待配对消息到来。

修改后的代码示例 #

首先更新消费逻辑，给处理方法传入队列名：

protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
    StartConsuming("alertQueue", stoppingToken); // 模板消息队列
    StartConsuming("fileUploadQueue", stoppingToken); // 文件URL队列

    await Task.CompletedTask;
}

private void StartConsuming(string queueName, CancellationToken cancellationToken)
{
    _channel.QueueDeclare(queue: queueName, durable: false, exclusive: false, autoDelete: false, arguments: null);

    var consumer = new EventingBasicConsumer(_channel);
    consumer.Received += async (model, ea) =>
    {
        var body = ea.Body.ToArray();
        var message = Encoding.UTF8.GetString(body);

        bool processedSuccessfully = false;
        try
        {
            processedSuccessfully = await ProcessMessageAsync(message, queueName); // 传入队列名区分消息类型
        }
        catch (Exception ex)
        {
            _logger.LogError($"Exception occurred while processing message from queue {queueName}: {ex}");
        }

        if (processedSuccessfully)
        {
            _channel.BasicAck(deliveryTag: ea.DeliveryTag, multiple: false);
        }
        else
        {
            _channel.BasicReject(deliveryTag: ea.DeliveryTag, requeue: true);
        }
    };

    _channel.BasicConsume(queue: queueName, autoAck: false, consumer: consumer);
}

然后更新处理逻辑，加入缓存配对逻辑：

// 先在构造函数注入IMemoryCache（多实例的话换IDistributedCache）
private readonly IMemoryCache _memoryCache;

public YourConsumerService(IOptions<RabbitMqSetting> rabbitMqSetting, 
                          ILogger<YourConsumerService> logger, 
                          IServiceProvider serviceProvider,
                          IMemoryCache memoryCache)
{
    _rabbitMqSetting = rabbitMqSetting.Value;
    _logger = logger;
    _serviceProvider = serviceProvider;
    _memoryCache = memoryCache;

    var factory = new ConnectionFactory
    {
        HostName = _rabbitMqSetting.HostName,
        UserName = _rabbitMqSetting.UserName,
        Password = _rabbitMqSetting.Password
    };
    _connection = factory.CreateConnection();
    _channel = _connection.CreateModel();
}

private async Task<bool> ProcessMessageAsync(string message, string queueName)
{
    try
    {
        using (var scope = _serviceProvider.CreateScope())
        {
            var emailService = scope.ServiceProvider.GetRequiredService<IEmailSender>();
            const string cachePrefix = "MessagePair_";

            if (queueName == "alertQueue")
            {
                // 处理模板消息
                var alertMessage = JsonConvert.DeserializeObject<AlertMessage>(message);
                if (alertMessage == null)
                {
                    _logger.LogWarning("无效的AlertMessage格式");
                    return false;
                }

                var cacheKey = $"{cachePrefix}{alertMessage.TemplateId}";
                // 尝试找对应的文件消息
                if (_memoryCache.TryGetValue(cacheKey, out FileUploadMessage uploadMessage))
                {
                    // 配对成功，发送邮件
                    await SendEmailWithAttachment(alertMessage, uploadMessage, emailService);
                    _memoryCache.Remove(cacheKey);
                    _logger.LogInformation($"已处理TemplateId为{alertMessage.TemplateId}的配对消息");
                    return true;
                }
                else
                {
                    // 没找到，存入缓存等配对
                    var cacheOpts = new MemoryCacheEntryOptions
                    {
                        AbsoluteExpirationRelativeToNow = TimeSpan.FromMinutes(30)
                    };
                    _memoryCache.Set(cacheKey, alertMessage, cacheOpts);
                    _logger.LogInformation($"已将AlertMessage存入缓存，等待配对的FileUploadMessage。TemplateId: {alertMessage.TemplateId}");
                    return true;
                }
            }
            else if (queueName == "fileUploadQueue")
            {
                // 处理文件URL消息
                var uploadMessage = JsonConvert.DeserializeObject<FileUploadMessage>(message);
                if (uploadMessage == null)
                {
                    _logger.LogWarning("无效的FileUploadMessage格式");
                    return false;
                }

                var cacheKey = $"{cachePrefix}{uploadMessage.TemplateId}";
                // 尝试找对应的模板消息
                if (_memoryCache.TryGetValue(cacheKey, out AlertMessage alertMessage))
                {
                    // 配对成功，发送邮件
                    await SendEmailWithAttachment(alertMessage, uploadMessage, emailService);
                    _memoryCache.Remove(cacheKey);
                    _logger.LogInformation($"已处理TemplateId为{uploadMessage.TemplateId}的配对消息");
                    return true;
                }
                else
                {
                    // 没找到，存入缓存等配对
                    var cacheOpts = new MemoryCacheEntryOptions
                    {
                        AbsoluteExpirationRelativeToNow = TimeSpan.FromMinutes(30)
                    };
                    _memoryCache.Set(cacheKey, uploadMessage, cacheOpts);
                    _logger.LogInformation($"已将FileUploadMessage存入缓存，等待配对的AlertMessage。TemplateId: {uploadMessage.TemplateId}");
                    return true;
                }
            }
            else
            {
                _logger.LogWarning($"未知队列：{queueName}");
                return false;
            }
        }
    }
    catch (JsonException jsonEx)
    {
        _logger.LogError($"JSON反序列化错误：{jsonEx.Message}");
        return false;
    }
    catch (Exception ex)
    {
        _logger.LogError($"消息处理错误：{ex.Message}");
        return false;
    }
}

// 邮件发送逻辑
private async Task SendEmailWithAttachment(AlertMessage alertMessage, FileUploadMessage uploadMessage, IEmailSender emailService)
{
    // 这里根据你的业务实现邮件发送：比如用文件URL下载附件，结合模板内容发送
    await emailService.SendEmailWithAttachmentAsync(
        recipient: alertMessage.RecipientEmail,
        subject: "新模板通知",
        content: alertMessage.TemplateContent,
        attachmentUrl: uploadMessage.FileUrl
    );
}

额外注意事项 #

• 多实例部署：如果你的消费者是多实例运行的，一定要用分布式缓存（比如Redis）代替IMemoryCache，否则不同实例的缓存不共享，会导致消息配对失败。
• 过期消息处理：对于缓存过期没配对的消息，可以设置死信队列，把这些消息转进去，后续人工排查或者自动重试。
• 可靠性保障：如果邮件发送失败，要考虑重试机制（比如把消息重新放回队列）或者存入数据库做补偿处理，避免消息丢失。

备注：内容来源于stack exchange，提问作者PASHA