Orleans中RabbitMQ流：避免重启未订阅时消息丢弃及处理残留消息

一、阻止Silo在订阅者就绪前拉取消息 #

通过两个核心配置可以避免启动时无订阅者导致的消息丢弃：

1. 配置RabbitMQ流仅在有订阅者时启动消费 #

在RabbitMQ流的配置中启用StartConsumingWhenSubscriptionExists选项，流提供者会等待至少一个订阅者就绪后，才从RabbitMQ队列拉取消息。

修改你的流配置代码：

siloHostBuilder = siloHostBuilder.AddRabbitMqStream(providerName, configurator =>
{
    configurator.ConfigureRabbitMq(ob =>
    {
        ob.Configure(options =>
        {
            // 保留原有连接配置
            options.Connection.HostName = "localhost";
            options.Connection.Port = 5672;
            options.Connection.VirtualHost = "/";
            options.Connection.UserName = "guest";
            options.Connection.Password = "guest";
            options.QueueNamePrefix = queueNamePrefix;
            options.UseQueuePartitioning = true;
            options.NumberOfQueues = 1;
            
            // 添加此配置项
            options.StartConsumingWhenSubscriptionExists = true;
        });
    });
});

2. 替换内存PubSub存储为持久化存储 #

默认的AddMemoryGrainStorage("PubSubStore")是内存级存储，Silo重启后会丢失所有订阅信息。换成持久化存储（如Redis、SQL Server），确保订阅信息在重启后保留，流提供者能识别已存在的订阅者。

示例：使用Redis作为PubSub存储（需安装Orleans.Persistence.Redis包）：

// 替换原有的AddMemoryGrainStorage("PubSubStore")
siloHostBuilder.AddRedisGrainStorage("PubSubStore", options =>
{
    options.Configuration = "localhost:6379";
});

二、处理Silo重启后队列中的残留消息 #

通过配置检查点存储和持久化订阅，确保重启后能正确续接未完成的消息消费：

1. 添加流检查点存储 #

检查点存储用于跟踪消费者的消费进度，重启后可以从上次处理的位置继续消费，避免重复或遗漏消息。

示例：添加Redis检查点存储（需安装Orleans.Streaming.Redis包）：

siloHostBuilder.AddRedisCheckpointStorage("CheckpointStore", options =>
{
    options.Configuration = "localhost:6379";
});

然后在RabbitMQ流配置中关联该检查点存储：

siloHostBuilder = siloHostBuilder.AddRabbitMqStream(providerName, configurator =>
{
    configurator.ConfigureRabbitMq(ob =>
    {
        // 原有配置...
    });
    // 关联检查点存储
    configurator.UseCheckpointStore("CheckpointStore");
});

2. 使用持久化流订阅 #

在Grain订阅流时，指定Persistent = true，确保订阅信息被持久化到PubSubStore，Silo重启后自动恢复订阅：

public class MyConsumerGrain : Grain, IMyConsumerGrain
{
    private IStreamProvider _streamProvider;

    public override async Task OnActivateAsync()
    {
        _streamProvider = this.GetStreamProvider(providerName);
        var streamId = StreamId.Create("MyStreamNamespace", "MyStreamId");
        var stream = _streamProvider.GetStream<MyMessage>(streamId);
        
        // 使用持久化订阅
        await stream.SubscribeAsync(
            OnMessageReceived,
            new StreamSubscriptionOptions { Persistent = true }
        );
        await base.OnActivateAsync();
    }

    private Task OnMessageReceived(MyMessage message, StreamSequenceToken token)
    {
        // 自定义消息处理逻辑
        return Task.CompletedTask;
    }
}

额外注意事项 #

• 确保RabbitMQ队列和消息都配置为持久化，避免Silo停机时消息丢失。
• 生产环境中，内存存储（如AddMemoryGrainStorage）仅用于开发测试，必须替换为持久化存储。

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