使用MassTransit、Postgres与EF Core实现消息去重的方法咨询

针对你的问题：MassTransit本身没有内置和Amazon SQS类似的、基于Correlation ID的消息自动去重功能，但可以结合PostgreSQL的数据库特性，在消息发布阶段实现重复消息拦截，确保相同Correlation ID的消息不会被重复加入队列。

方案一：发布前通过唯一约束校验Correlation ID #

这是最直接的方式，通过在PostgreSQL中维护一张已发布消息的Correlation ID记录表，利用唯一约束来拦截重复值：

1. 创建去重记录表
   先在PostgreSQL中创建一张用于记录已发布消息Correlation ID的表，并给correlation_id字段添加唯一约束：

   CREATE TABLE published_message_correlations (
       id SERIAL PRIMARY KEY,
       correlation_id UUID NOT NULL UNIQUE,
       created_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP
   );
   

2. 在发布消息前执行校验
   在发布消息的业务逻辑中，用PostgreSQL事务包裹“插入Correlation ID”和“发布消息”的操作：

   using var transaction = await _dbContext.Database.BeginTransactionAsync();
   try
   {
       // 尝试插入Correlation ID，重复值会触发唯一约束异常
       await _dbContext.PublishedMessageCorrelations.AddAsync(new PublishedMessageCorrelation
       {
           CorrelationId = message.CorrelationId
       });
       await _dbContext.SaveChangesAsync();
   
       // 插入成功，发布消息
       await _bus.Publish(message);
   
       await transaction.CommitAsync();
   }
   catch (PostgresException ex) when (ex.SqlState == "23505") // 唯一约束冲突的错误码
   {
       // 忽略重复消息，回滚事务
       await transaction.RollbackAsync();
   }
   

   当存在相同Correlation ID的消息时，插入操作会触发PostgreSQL的唯一约束异常，此时直接回滚事务、不发布消息即可。

方案二：结合MassTransit的Transactional Outbox实现 #

如果你已经在使用MassTransit的Transactional Outbox（事务发件箱）功能，可以直接给Outbox表的correlation_id字段添加唯一约束，从根源上阻止重复消息进入发件箱：

1. 给Outbox表添加唯一约束
   针对MassTransit自动创建的mt_outbox表，执行以下SQL添加唯一约束：

   ALTER TABLE mt_outbox ADD CONSTRAINT uq_mt_outbox_correlation_id UNIQUE (correlation_id);
   

2. 配置Outbox并发布消息
   当你通过Transactional Outbox发布消息时，MassTransit会先将消息存入mt_outbox表。如果存在相同Correlation ID的消息，插入操作会触发唯一约束异常，消息不会被存入Outbox，自然不会被后续发布到队列中：

   // 配置Transactional Outbox（示例）
   services.AddMassTransit(x =>
   {
       x.AddEntityFrameworkOutbox<AppDbContext>(o =>
       {
           o.QueryDelay = TimeSpan.FromSeconds(10);
           o.UsePostgres();
           o.UseBusOutbox();
       });
   
       // 其他配置...
   });
   
   // 通过Outbox发布消息
   await _dbContext.AddOutboxMessage(message);
   await _dbContext.SaveChangesAsync();
   

注意事项 #

• 确保Correlation ID的全局唯一性：如果Correlation ID不是全局唯一的，会误拦截合法的不同消息，建议使用UUID作为Correlation ID。
• 定期清理过期记录：对于方案一中的published_message_correlations表，可添加定时任务清理过期（比如超过7天）的记录，避免表数据过大。
• 消费者端去重：如果需要在消费者端确保消息不被重复处理，还可以结合MassTransit的Saga或自定义存储来实现，但这和你需求的“发布阶段忽略重复消息”是两个不同的环节。

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