MassTransit配置Amazon SQS FIFO队列批量消费者失败求助

我尝试用MassTransit给Amazon SQS FIFO队列配置批量消费者，但不管怎么弄，消费者都是按单条消息触发的。怀疑这是FIFO队列的特性限制，想问问其他开发者有没有遇到类似情况，下面是我用的代码：

// Message:
public record BatchMessage();

// Consumer:
public class BatchMessageConsumer : IConsumer<Batch<BatchMessage>>
{
    public Task Consume(ConsumeContext<Batch<BatchMessage>> context)
    {
        Debug.WriteLine($"Messages={context.Message.Length}");

        return Task.CompletedTask;
    }
}

// ConsumerDefinition:
public class BatchMessageConsumerDefinition : ConsumerDefinition<BatchMessageConsumer>
{
    public BatchMessageConsumerDefinition()
    {
        // SQS Queue name.
        Endpoint(x => x.Name = $"Demo-BatchMessage.fifo");
    }
}

// Configuration:
services.AddMassTransit(x =>
{
    x.AddConsumers(typeof(Program).Assembly);

    x.UsingAmazonSqs((context, sqs) =>
    {
        sqs.Host("region", (_) => { });

        // SNS Topic name.
        sqs.Message<BatchMessage>(x => x.SetEntityName("BatchMessage.fifo"));

        sqs.ConfigureEndpoints(context);
    });
});

// Trigger:
var groupId = Guid.NewGuid().ToString();

for (var i = 1; i <= 10; i++)
{
    await _publishEndpoint.Publish(new BatchMessage(), (context) =>
    {
        // Required for FIFO messages.
        context.TrySetGroupId(groupId);
        context.TrySetDeduplicationId(context.MessageId.ToString());
    });
}

问题原因 #

MassTransit默认的批量消费者逻辑适配普通队列，但SQS FIFO队列有**消息组(Message Group)**的强制顺序要求——同一消息组内的消息必须按顺序处理，默认批量拉取逻辑无法直接生效。另外，当前代码没有显式配置批量参数（比如批量大小、等待超时），这也是消费者单条触发的关键原因。

解决方案 #

1. 显式配置消费者的批量参数

修改BatchMessageConsumerDefinition，指定批量大小、等待时间，并开启同一消息组内批量处理的限制：

public class BatchMessageConsumerDefinition : ConsumerDefinition<BatchMessageConsumer>
{
    public BatchMessageConsumerDefinition()
    {
        Endpoint(x => 
        {
            x.Name = $"Demo-BatchMessage.fifo";
            // 配置批量参数
            x.BatchOptions = new BatchOptions
            {
                MessageLimit = 10, // 单次批量处理的最大消息数
                TimeLimit = TimeSpan.FromSeconds(5), // 等待凑齐批量的超时时间
                // 仅对同一消息组的消息进行批量处理，符合FIFO队列的顺序要求
                GroupMessageTypes = true
            };
        });
    }
}

2. 调整SQS端点的批量接收配置

在MassTransit的SQS配置中，开启批量接收，并匹配消费者的批量参数：

services.AddMassTransit(x =>
{
    x.AddConsumers(typeof(Program).Assembly);

    x.UsingAmazonSqs((context, sqs) =>
    {
        sqs.Host("region", (_) => { });

        sqs.Message<BatchMessage>(x => x.SetEntityName("BatchMessage.fifo"));

        // 手动配置接收端点，开启SQS批量接收
        sqs.ReceiveEndpoint("Demo-BatchMessage.fifo", e =>
        {
            e.ConfigureConsumer<BatchMessageConsumer>(context);
            // SQS拉取消息的批量大小，要和消费者的MessageLimit一致
            e.BatchSize = 10;
            // 长轮询等待时间，建议不超过批量超时时间
            e.WaitTimeSeconds = 5;
        });
    });
});

关键注意点 #

• FIFO消息组限制：同一消息组的消息必须顺序处理，所以批量只能针对同组消息。如果消息分属不同组，消费者会按组分别批量处理，不会跨组合并。
• 参数匹配：SQS的BatchSize必须和消费者的MessageLimit一致，否则会出现拉取数量和处理数量不匹配的情况。
• 发布逻辑：你的发布代码已经给批量消息设置了相同的GroupId，这部分是正确的，确保了消息进入同一组，才能被批量处理。

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