如何基于消息属性过滤Amazon SQS队列消息？Python实现遇阻

首先明确：Amazon SQS标准队列本身不支持直接在receive_message调用中基于消息属性过滤消息。如果要实现服务端级别的过滤（无需拉取后再筛选），推荐以下两种方案：

方案1：SNS主题 + 带过滤策略的SQS订阅 #

通过SNS主题路由消息，给每个需要过滤条件的SQS队列设置订阅过滤策略，SNS会自动将符合属性的消息投递到对应队列，消费时直接读取目标队列即可。

完整代码示例 #

import boto3
import json

# 初始化客户端
sns_client = boto3.client('sns', region_name="us-west-1")
sqs_client = boto3.client('sqs', region_name="us-west-1")

# 1. 创建SNS主题
topic_response = sns_client.create_topic(Name='test_topic_development')
topic_arn = topic_response['TopicArn']

# 2. 创建三个SQS队列，分别对应不同的过滤条件
# 队列1：接收method=DATA的消息
data_queue_response = sqs_client.create_queue(QueueName='data_queue_development')
data_queue_url = data_queue_response['QueueUrl']
data_queue_arn = sqs_client.get_queue_attributes(QueueUrl=data_queue_url, AttributeNames=['QueueArn'])['Attributes']['QueueArn']

# 队列2：接收method=HELP的消息
help_queue_response = sqs_client.create_queue(QueueName='help_queue_development')
help_queue_url = help_queue_response['QueueUrl']
help_queue_arn = sqs_client.get_queue_attributes(QueueUrl=help_queue_url, AttributeNames=['QueueArn'])['Attributes']['QueueArn']

# 队列3：接收method=OUT_OF_SCOPE的消息
oos_queue_response = sqs_client.create_queue(QueueName='oos_queue_development')
oos_queue_url = oos_queue_response['QueueUrl']
oos_queue_arn = sqs_client.get_queue_attributes(QueueUrl=oos_queue_url, AttributeNames=['QueueArn'])['Attributes']['QueueArn']

# 3. 给每个队列设置订阅和过滤策略
# 订阅DATA队列，过滤method=DATA
sns_client.subscribe(
    TopicArn=topic_arn,
    Protocol='sqs',
    Endpoint=data_queue_arn,
    Attributes={
        'FilterPolicy': json.dumps({'method': ['DATA']})
    }
)

# 订阅HELP队列，过滤method=HELP
sns_client.subscribe(
    TopicArn=topic_arn,
    Protocol='sqs',
    Endpoint=help_queue_arn,
    Attributes={
        'FilterPolicy': json.dumps({'method': ['HELP']})
    }
)

# 订阅OUT_OF_SCOPE队列，过滤method=OUT_OF_SCOPE
sns_client.subscribe(
    TopicArn=topic_arn,
    Protocol='sqs',
    Endpoint=oos_queue_arn,
    Attributes={
        'FilterPolicy': json.dumps({'method': ['OUT_OF_SCOPE']})
    }
)

# 4. 给SNS主题发送带属性的消息
# 发送DATA类型消息
sns_client.publish(
    TopicArn=topic_arn,
    Message=json.dumps({'test_data_key': 'test_data_value'}),
    MessageAttributes={
        'method': {
            'DataType': 'String',
            'StringValue': 'DATA'
        }
    }
)

# 发送HELP类型消息
sns_client.publish(
    TopicArn=topic_arn,
    Message=json.dumps({'test_help_key': 'test_help_value'}),
    MessageAttributes={
        'method': {
            'DataType': 'String',
            'StringValue': 'HELP'
        }
    }
)

# 发送OUT_OF_SCOPE类型消息
sns_client.publish(
    TopicArn=topic_arn,
    Message=json.dumps({'test_out_of_scope_key': 'test_out_of_scope_value'}),
    MessageAttributes={
        'method': {
            'DataType': 'String',
            'StringValue': 'OUT_OF_SCOPE'
        }
    }
)

# 5. 消费DATA队列的消息（只会收到method=DATA的消息）
def consume_queue(queue_url):
    response = sqs_client.receive_message(
        QueueUrl=queue_url,
        AttributeNames=['All'],
        MessageAttributeNames=['All'],
        MaxNumberOfMessages=10,
        WaitTimeSeconds=5  # 长轮询，减少空响应
    )
    
    if 'Messages' in response:
        for message in response['Messages']:
            print(f"收到消息：Body={message['Body']}")
            print(f"消息属性method={message['MessageAttributes']['method']['StringValue']}")
            
            # 处理完成后删除消息
            sqs_client.delete_message(
                QueueUrl=queue_url,
                ReceiptHandle=message['ReceiptHandle']
            )
    else:
        print(f"队列{queue_url}中无符合条件的消息")

# 测试消费DATA队列
consume_queue(data_queue_url)

方案2：Lambda触发器 + SQS过滤规则 #

如果用Lambda处理SQS消息，可以在Lambda触发器配置中添加过滤规则，只有符合属性条件的消息才会触发Lambda执行，Lambda只会收到符合条件的消息：

1. 创建Lambda函数，配置目标SQS队列作为触发器
2. 在触发器设置的过滤规则中添加条件，比如：{"method": ["DATA"]}
3. 当队列中有符合条件的消息时，Lambda会自动触发并处理，无需手动拉取和过滤

原代码的问题说明 #

你提供的代码最后部分检查的是send_message的返回结果，而非调用receive_message获取的消息，因此根本没有执行拉取消息的操作。即使补充receive_message调用，标准SQS也不支持过滤逻辑，仍会拉取队列内所有消息。

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