SQS与AWS事件桥的集成

Question

我正在尝试建立演示环境来尝试使用SQS作为AWS事件桥接源代码。 我尝试将几个文档上传到SQS以查看Event Bridge是否检测到任何更改，但我没有看到任何触发的事件。如何使用AWS事件桥接测试SQS作为源代码？

Resources:
  Queue:
    Type: AWS::SQS::Queue
    Properties:
      QueueName: !Sub ${AWS::StackName}

  LambdaHandlerExecutionRole:
    Type: AWS::IAM::Role
    Properties:
      AssumeRolePolicyDocument:
        Version: '2012-10-17'
        Statement:
          - Effect: Allow
            Principal:
              Service:
                - lambda.amazonaws.com
            Action: sts:AssumeRole
      ManagedPolicyArns:
        - arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole

  EventConsumerFunction:
    Type: AWS::Lambda::Function
    Properties:
      Handler: index.lambda_handler
      Role: !GetAtt LambdaHandlerExecutionRole.Arn
      Code:
        ZipFile: |
          import json

          def lambda_handler(event, context):
              print("Received event: " + json.dumps(event, indent=2))

      Runtime: python3.7
      Timeout: 50

  EventRule:
    Type: AWS::Events::Rule
    Properties:
      Description: eventEventRule
      State: ENABLED
      EventPattern:
        source:
          - aws.sqs
        resources:
          - !GetAtt Queue.Arn
      Targets:
        - Arn: !GetAtt EventConsumerFunction.Arn
          Id: EventConsumerFunctionTarget

  PermissionForEventsToInvokeLambda:
    Type: AWS::Lambda::Permission
    Properties:
      FunctionName: !Ref EventConsumerFunction
      Action: lambda:InvokeFunction
      Principal: events.amazonaws.com
      SourceArn: !GetAtt EventRule.Arn

- thotam

Answer 1

SQS数据事件（发布新消息）不是Event Bridge（EB）的源事件。只有管理事件可以被EB选中，例如：

• 清空队列
• 创建新队列
• 删除队列

此外，您的事件规则应该更通用：

  EventRule:
    Type: AWS::Events::Rule
    Properties:
      Description: eventEventRule
      State: ENABLED
      EventPattern:
        source:
          - aws.sqs
        # resources:
        #   - !GetAtt Queue.Arn
      Targets:
        - Arn: !GetAtt EventConsumerFunction.Arn
          Id: EventConsumerFunctionTarget

您还可以启用CloudWatch试用版，以便检测SQS的API事件。这样可以获取更多的事件。

- Marcin

---

谢谢您的澄清。看起来它不符合我的要求。 - thotam

@thotam 你想要达到什么目的？如果你想要发布消息到 EB，你需要这样做：SQS -> Lambda -> EB。 - Marcin

我正在尝试使用EventBridge来基于SQS消息数量扩展Fargate容器。目前，我的流程是：SQS --> Cloudwatch Alarm（ApproximateNumberOfMessagesVisible）--> SNS --> Lambda --> Fargate调整大小。而现在，我想通过EventBridge来实现这个过程：SQS --> EventBridge --> Fargate。 - thotam

很遗憾，你不能使用SQS->EB->Fargate这个方案。为什么不直接使用SQS->Fargate呢？ - Marcin

我们需要短暂的Fargate容器，仅在SQS中有消息时运行。Lambda不适用于我们的用例，因为我们的作业运行时间超过15分钟。 - thotam

Answer 2

我可能有些迟到，但这份内容可以帮助其他人。请查看以下链接：https://docs.aws.amazon.com/cdk/api/latest/docs/@aws-cdk_aws-ecs-patterns.QueueProcessingFargateService.html 该链接介绍了如何使用 AWS CDK 创建一个能够根据 SQS 队列中消息数量来自动扩展 Fargate 容器的服务。

下面是一个最简单的使用 AWS CDK 定义 stack 的示例：

queue = sqs.Queue(stack, "Queue")

cluster = aws_ecs.Cluster(
            stack, 'FargateCluster'
        )

queue_processing_fargate_service = QueueProcessingFargateService(stack, "Service",
        cluster=cluster,
        memory_limit_mi_b=512,
        image=ecs.ContainerImage.from_registry("test"),
        command=["-c", "4", "amazon.com"],
        enable_logging=False,
        desired_task_count=2,
        environment={
            "TEST_ENVIRONMENT_VARIABLE1": "test environment variable 1 value",
            "TEST_ENVIRONMENT_VARIABLE2": "test environment variable 2 value"
        },
        queue=queue,
        max_scaling_capacity=5,
        container_name="test"
    )

- Potato

---

请添加更多细节以扩展您的答案，例如工作代码或文档引用。 - Community

嘿，我正在尝试使用这个QueueProcessingFargateService，但是我怎么知道队列中的哪些数据会进入我的Python函数，或者说如何访问消息？关于这个东西的文档相当缺乏。 - Will.Evo