如何通过MassTransit向现有SQS FIFO队列发送原始JSON消息？

问题背景 #

需要向现有SQS FIFO队列（URL：https://sqs.region.amazonaws.com/accountid/mt-send-test.fifo）发送原始JSON消息，已在Startup中配置ReceiveEndpoint，但发送时遇到两个问题：

1. 无法通过queue:mt-send-test.fifo解析正确的端点URI，使用完整URL调用GetSendEndpoint会抛出异常
2. 发送消息时出现Amazon.SQS.AmazonSQSException: 'The request must contain the parameter MessageGroupId.'异常

解决方案 #

1. 正确获取SQS FIFO队列的SendEndpoint #

MassTransit针对SQS队列有专属的URI格式，不能直接使用queue:前缀或完整的SQS HTTP URL，需要使用amazonsqs://协议的URI：

var endpoint = await _bus.GetSendEndpoint(new Uri("amazonsqs://region/accountid/mt-send-test.fifo"));

替换其中的region和accountid为实际的AWS区域和账户ID即可。

2. 补充FIFO队列必需的MessageGroupId #

SQS FIFO队列强制要求每个消息指定MessageGroupId，用于将消息分组到不同的消息组中，同一组内的消息会按顺序处理。可以通过以下方式设置：

方式一：发送时通过SendContext动态设置

适合需要根据业务逻辑动态指定分组ID的场景：

MyMessage mtMessage = // 初始化你的消息对象
await endpoint.Send(mtMessage, context =>
{
    // 设置消息分组ID，替换为你的业务分组标识
    context.SetMessageGroupId("order-processing-group");
    
    // 可选：如果需要消息去重，设置MessageDeduplicationId（FIFO队列可选，若开启Content-Based Deduplication则可省略）
    // context.SetMessageDeduplicationId(Guid.NewGuid().ToString());
});

方式二：在消息类上标记固定分组ID

如果该消息类型始终属于同一个分组，可以直接在消息类上添加特性：

[MessageGroupId("default-group-id")]
public class MyMessage
{
    // 消息属性
}

发送时无需额外设置，MassTransit会自动使用该分组ID。

3. 确认序列化配置生效 #

已在ReceiveEndpoint中配置了UseRawJsonSerializer，为了确保发送时也使用原始JSON序列化，建议在总线配置时全局设置默认序列化器：

cfg.SetSerializer(new RawJsonSerializer());

这样无论是消费还是发送消息，都会使用原始JSON格式，避免序列化不一致的问题。

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