如何在AWS CDK中定义SNS消息体过滤策略？

核心实现思路 #

在AWS CDK中，为SNS订阅配置负载过滤策略，核心是通过Subscription的filterPolicy属性（结合SubscriptionFilter工具类）定义规则。若需基于消息体而非消息属性过滤，需额外开启filterPolicyWithMessageBody开关。

完整代码示例 #

基于消息属性的过滤

import * as cdk from 'aws-cdk-lib';
import { Construct } from 'constructs';
import * as sns from 'aws-cdk-lib/aws-sns';
import * as subs from 'aws-cdk-lib/aws-sns-subscriptions';
import * as lambda from 'aws-cdk-lib/aws-lambda';

export class SnsFilterPolicyStack extends cdk.Stack {
  constructor(scope: Construct, id: string, props?: cdk.StackProps) {
    super(scope, id, props);

    // 创建目标SNS主题
    const targetTopic = new sns.Topic(this, 'FilteredOrderTopic', {
      displayName: '订单事件过滤主题',
    });

    // 创建作为订阅者的Lambda函数
    const orderProcessor = new lambda.Function(this, 'OrderProcessorFunc', {
      runtime: lambda.Runtime.NODEJS_18_X,
      code: lambda.Code.fromInline(`
        exports.handler = async (event) => {
          console.log('处理符合过滤规则的订单事件:', event.Records[0].Sns.Message);
          return { statusCode: 200 };
        };
      `),
      handler: 'index.handler',
    });

    // 配置过滤策略并绑定订阅
    targetTopic.addSubscription(new subs.LambdaSubscription(orderProcessor, {
      filterPolicy: {
        // 过滤消息属性中event_type为order_created/order_updated的消息
        event_type: sns.SubscriptionFilter.stringFilter({
          allowlist: ['order_created', 'order_updated'],
        }),
        // 过滤消息属性中amount大于等于100的消息
        amount: sns.SubscriptionFilter.numericFilter({
          greaterThanOrEqualTo: 100,
        }),
        // 过滤消息属性中is_priority为true的消息
        is_priority: sns.SubscriptionFilter.booleanFilter({
          allowlist: [true],
        }),
      },
    }));
  }
}

基于消息体的过滤

若需直接过滤消息体中的字段，需开启filterPolicyWithMessageBody，并通过JSONPath指定字段路径：

targetTopic.addSubscription(new subs.LambdaSubscription(orderProcessor, {
  // 开启消息体过滤模式
  filterPolicyWithMessageBody: true,
  filterPolicy: {
    // 过滤消息体中order.status为SHIPPED的消息
    '$.order.status': sns.SubscriptionFilter.stringFilter({
      allowlist: ['SHIPPED'],
    }),
    // 过滤消息体中order.items_count大于等于3的消息
    '$.order.items_count': sns.SubscriptionFilter.numericFilter({
      greaterThanOrEqualTo: 3,
    }),
  },
}));

常见编译失败排查 #

• 类型不匹配：直接传递普通JSON对象给filterPolicy会触发类型错误，必须使用SubscriptionFilter提供的静态方法（stringFilter/numericFilter等）构建规则
• 规则冲突：部分过滤类型不允许同时设置allowlist和denylist，需遵循SNS过滤策略的语法限制
• 字段路径错误：基于消息体过滤时，JSONPath格式错误（如缺少$前缀）会导致编译或运行时错误

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