如何使用messagegroupid参数仅接收标记有所需ID的队列消息?
我一直在尝试使用下面的代码行进行检索,但它总是会接收来自其他组ID的所有队列消息。
List<Message> messages = sqs.receiveMessage(receiveMessageRequest.withAttributeNames("MessageGroupId")).getMessages();
应该如何正确操作?
如何使用messagegroupid参数仅接收标记有所需ID的队列消息?
我一直在尝试使用下面的代码行进行检索,但它总是会接收来自其他组ID的所有队列消息。
List<Message> messages = sqs.receiveMessage(receiveMessageRequest.withAttributeNames("MessageGroupId")).getMessages();
应该如何正确操作?
ReceiveMessageRequest 不用于基于消息属性的过滤。如果查看 ReceiveMessageRequest.html.withAttributeNames() 的文档,它会说:
需要与每个消息一起返回的属性列表。
总体来说,您无法过滤从 SQS 返回的消息。您可以限制数量,但不能说:“给我所有符合此模式的消息。”
_awsSqsService.SendBackToQueueAsync(groupMessages)
方法最终通过 AWS SQS ChangeMessageVisibilityBatchRequest
将消息发送回队列。)public async Task Run()
{
var batchContainsMessages = true;
while (batchContainsMessages)
{
var messageBatch = await _awsSqsService.GetMessageBatchAsync();
if(messageBatch.Messages.Count > 0 && messageBatch.HttpStatusCode == HttpStatusCode.OK)
{
await ProcessMessageBatchAsync(messageBatch.Messages);
}
else
{
batchContainsMessages = false;
}
}
}
private async Task ProcessMessageBatchAsync(List<Message> messages)
{
// SQS fifo queues will often return a batch of messages with different MessageGroupIds.
// Because of this, we need to group them ourselves, process one group (the first group),
// and send the rest back to the queue to be processed in the next iteration.
// This ensures that we process as many messages as possible per group in a single batch (max is 10)
var messageGroups = GetMessageGroups(messages);
var isFirstGroup = true;
foreach (var group in messageGroups)
{
var groupId = Int32.Parse(group.Key);
var groupMessages = group.Value;
if (isFirstGroup)
{
isFirstGroup = false;
await ProcessMessagesAsync(groupId, groupMessages);
await _awsSqsService.DeleteMessagesAsync(groupMessages);
}
else
{
await _awsSqsService.SendBackToQueueAsync(groupMessages);
}
}
}