消费成功后如何从仓库删除MessageData？

问题描述 #

我通过RabbitMQ传输大量数据，为减轻队列负载采用Claim Check模式将字节blob存储在队列外，使用FileSystemMessageDataRepository并设置7天TTL以应对消费者故障。但希望消息消费成功后删除对应的MessageData，却发现IMessageDataRepository无删除方法，也无自动删除选项，求解决方案。

我的RabbitMQ配置代码 #

var messageDataRepository = new FileSystemMessageDataRepository(new DirectoryInfo(massTransitSettings.RabbitMqDataMessageRepositoryPath));
serviceCollection.AddSingleton<IMessageDataRepository>(messageDataRepository);
serviceCollection.AddMassTransit(x =>
{
    x.UsingRabbitMq((context,cfg) =>
    {
        cfg.Host(massTransitSettings.RabbitMqHost, massTransitSettings.RabbitMqVirtualHost, h => {
            h.Username(massTransitSettings.RabbitMqUsername);
            h.Password(massTransitSettings.RabbitMqPassword);
        });

        cfg.UseMessageData(messageDataRepository);

        cfg.ConfigureEndpoints(context);
    });
});

消息发送与TTL设置代码 #

var messageData = await messageDataRepository.PutBytes(bufferToSend, TimeSpan.FromDays(7), cancellationToken);
var payload = new MyPayload(messageData, bytesRead);
await bus.Publish(payload, cancellationToken);

解决方案 #

1. 直接转换仓库实例调用删除方法 #

IMessageDataRepository确实没定义删除接口，但具体实现FileSystemMessageDataRepository自带了DeleteAsync方法。你可以在消费者处理完消息后，把仓库实例转换成具体类型，调用删除：

public class MyPayloadConsumer : IConsumer<MyPayload>
{
    private readonly IMessageDataRepository _repository;

    public MyPayloadConsumer(IMessageDataRepository repository)
    {
        _repository = repository;
    }

    public async Task Consume(ConsumeContext<MyPayload> context)
    {
        // 先处理业务逻辑
        var data = await context.Message.MessageData.Value;
        // ... 你的业务代码

        // 消费成功后删除对应文件
        if (_repository is FileSystemMessageDataRepository fileRepo)
        {
            await fileRepo.DeleteAsync(new Uri(context.Message.MessageData.Address), context.CancellationToken);
        }
    }
}

2. 封装自定义接口解耦依赖 #

如果不想直接依赖具体实现，可封装一个带删除功能的接口和适配器，避免强耦合：

// 自定义带删除功能的接口
public interface IMessageDataRepositoryWithDelete : IMessageDataRepository
{
    Task DeleteAsync(Uri address, CancellationToken cancellationToken);
}

// 适配器类，包装FileSystemMessageDataRepository
public class FileSystemMessageDataRepositoryAdapter : IMessageDataRepositoryWithDelete
{
    private readonly FileSystemMessageDataRepository _innerRepo;

    public FileSystemMessageDataRepositoryAdapter(FileSystemMessageDataRepository innerRepo)
    {
        _innerRepo = innerRepo;
    }

    public Task<MessageData<T>> GetAsync<T>(Uri address, CancellationToken cancellationToken)
        => _innerRepo.GetAsync<T>(address, cancellationToken);

    public Task<MessageData> GetAsync(Uri address, CancellationToken cancellationToken)
        => _innerRepo.GetAsync(address, cancellationToken);

    public Task<MessageData> PutBytes(byte[] value, TimeSpan timeToLive, CancellationToken cancellationToken)
        => _innerRepo.PutBytes(value, timeToLive, cancellationToken);

    public Task<MessageData<T>> PutObject<T>(T value, TimeSpan timeToLive, CancellationToken cancellationToken)
        => _innerRepo.PutObject<T>(value, timeToLive, cancellationToken);

    public Task DeleteAsync(Uri address, CancellationToken cancellationToken)
        => _innerRepo.DeleteAsync(address, cancellationToken);
}

修改DI注册代码：

var fileRepo = new FileSystemMessageDataRepository(new DirectoryInfo(massTransitSettings.RabbitMqDataMessageRepositoryPath));
// 注册自定义接口
serviceCollection.AddSingleton<IMessageDataRepositoryWithDelete>(new FileSystemMessageDataRepositoryAdapter(fileRepo));
// 同时注册原接口给MassTransit用
serviceCollection.AddSingleton<IMessageDataRepository>(fileRepo);

之后在消费者中注入IMessageDataRepositoryWithDelete，直接调用DeleteAsync即可。

3. 后台定时清理兜底 #

就算实现了主动删除，也建议加个定时任务清理过期文件，防止消费者异常导致文件残留。用FileSystemMessageDataRepository的CleanupAsync方法就行：

// 实现后台服务定时清理
public class MessageDataCleanupService : BackgroundService
{
    private readonly FileSystemMessageDataRepository _fileRepo;
    private readonly ILogger<MessageDataCleanupService> _logger;

    public MessageDataCleanupService(IMessageDataRepository repository, ILogger<MessageDataCleanupService> logger)
    {
        _fileRepo = repository as FileSystemMessageDataRepository ?? throw new InvalidOperationException("Repository must be FileSystemMessageDataRepository");
        _logger = logger;
    }

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        while (!stoppingToken.IsCancellationRequested)
        {
            try
            {
                // 每天执行一次清理
                await _fileRepo.CleanupAsync(stoppingToken);
                _logger.LogInformation("Message data cleanup finished");
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, "Failed to clean up message data");
            }

            await Task.Delay(TimeSpan.FromDays(1), stoppingToken);
        }
    }
}

注册这个后台服务：

serviceCollection.AddHostedService<MessageDataCleanupService>();

这样既能主动删除已消费的文件，又能通过定时清理处理异常场景。

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