Nginx集群中多.NET Core实例通过RabbitMQ同步InMemoryCache方案咨询

1. 核心问题修正：切换RabbitMQ为发布/订阅模式 #

你当前的问题根源是用了点对点队列消费——多个实例争抢同一个队列的消息，自然只有一个实例能收到并更新缓存。改成发布/订阅模式后，消息会广播给所有订阅的实例，确保每个实例都能触发缓存更新：

• 创建Fanout类型交换机：这种交换机会把收到的消息转发给所有绑定的队列，实现广播效果
• 每个.NET Core实例创建专属队列：避免实例间抢消息，每个实例监听自己的队列
• 将所有实例的专属队列绑定到同一个Fanout交换机：消息发送到交换机后，所有绑定队列都会收到副本

2. .NET Core代码实现示例 #

初始化RabbitMQ连接与交换机 #

在Program.cs或配置类中完成基础配置：

using RabbitMQ.Client;
using System.Text;

var factory = new ConnectionFactory() { HostName = "你的RabbitMQ地址" };
using var connection = factory.CreateConnection();
using var channel = connection.CreateModel();

// 声明持久化的Fanout交换机
channel.ExchangeDeclare(
    exchange: "cache-update-exchange",
    type: ExchangeType.Fanout,
    durable: true
);

// 为当前实例创建唯一队列（自动生成随机名称，保证实例间队列不重复）
var queueName = channel.QueueDeclare(durable: true).QueueName;
// 将队列绑定到交换机
channel.QueueBind(
    queue: queueName,
    exchange: "cache-update-exchange",
    routingKey: "" // Fanout交换机不需要路由键
);

实例监听消息并更新缓存 #

每个实例启动时启动消费者，监听专属队列的消息：

using var consumer = new EventingBasicConsumer(channel);
consumer.Received += (model, ea) =>
{
    var body = ea.Body.ToArray();
    var cacheKey = Encoding.UTF8.GetString(body);
    
    // 执行缓存更新逻辑：清除指定缓存或重新加载数据
    _cache.Remove(cacheKey);
    // 若需要重新加载：_cache.Set(cacheKey, FetchLatestDataFromDb(cacheKey));
    
    // 手动确认消息（开启持久化时建议使用，避免消息丢失）
    channel.BasicAck(ea.DeliveryTag, multiple: false);
};

// 开启手动确认，关闭自动ACK
channel.BasicConsume(
    queue: queueName,
    autoAck: false,
    consumer: consumer
);

触发缓存更新：生产者发送消息 #

当数据变更需要更新缓存时，向交换机发送消息：

using var connection = factory.CreateConnection();
using var channel = connection.CreateModel();

var cacheKey = "需要更新的缓存键"; // 也可以用"all"表示全量更新缓存
var body = Encoding.UTF8.GetBytes(cacheKey);

// 发送持久化消息到交换机
var properties = channel.CreateBasicProperties();
properties.Persistent = true;

channel.BasicPublish(
    exchange: "cache-update-exchange",
    routingKey: "",
    basicProperties: properties,
    body: body
);

3. 关键注意事项 #

• 幂等性保障：确保缓存更新操作是幂等的——重复收到同一个更新消息时，不会导致缓存数据异常（比如清除缓存的操作天然幂等，重新加载数据则要保证查询逻辑返回一致结果）
• 新实例初始化：新启动的实例需要主动拉取最新数据初始化缓存，避免刚启动时缓存为空
• 消息可靠性：开启交换机、队列、消息的持久化配置，配合手动ACK，防止RabbitMQ重启或实例宕机导致消息丢失

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