以Amazon MQ 消息代理为来源 - 亚马逊 EventBridge
使用者组执行角色权限网络配置以 Amazon MQ 为源配置管道
Amazon Web Services 文档中描述的 Amazon Web Services 服务或功能可能因区域而异。要查看适用于中国区域的差异，请参阅 中国的 Amazon Web Services 服务入门 (PDF)。

本文属于机器翻译版本。若本译文内容与英语原文存在差异，则一律以英文原文为准。

以Amazon MQ 消息代理为来源

您可以使用 EventBridge Pips 接收来自 Amazon MQ 消息代理的记录。然后，您可以选择筛选或增强这些记录，然后再将它们发送到一个可用的目的地进行处理。在设置管道时，您可以选择特定于 Amazon MQ 的设置。 EventBridge 在将数据发送到目标时，管道会维护消息代理的记录顺序。

Amazon MQ 是一项托管消息代理服务，用于 Apache ActiveMQ 和 RabbitMQ。消息代理允许软件应用程序和组件使用不同的编程语言、操作系统和正式消息收发协议进行通信，使用主题或队列进行通信。

Amazon MQ 还可以通过安装 ActiveMQ 代理或 RabbitMQ 代理以及代表您管理 Amazon Elastic Comute Cloud (Amazon EC2) 实例。安装代理后，它会为您的实例提供不同的网络拓扑和其他基础设施需求。

Amazon MQ 源有以下配置限制：

  • 跨账户 — EventBridge 不支持跨账户处理。您不能使用 EventBridge 处理来自不同Amazon账户中的 Amazon MQ 消息代理的记录。

  • 身份验证 — 对于 ActiveMQ，仅支持 ActiveMQ SimpleAuthenticationPlugin。对于 RabbitMQ，仅支持 PLAIN 身份验证机制。要管理证书，请使用Amazon Secrets Manager。有关 ActiveMQ 身份验证的更多信息，请参阅 Amazon MQ 开发人员指南中的使用 LDAP 集成 ActiveMQ 代理。

  • 连接配额 — 代理具有每个有线级协议允许的最大连接数。此配额基于代理实例类型。有关更多信息，请参阅 Amazon MQ 开发人员指南中的 *Amazon MQ 中的配额的代理部分。

  • 连接 — 您可以在公共或私有虚拟私有云（VPC）中创建代理。对于私有 VPC，您的管道需要具备对 VPC 的访问权限才能接收消息。

  • 事件目标 — 仅支持队列目标。但是，您可以使用虚拟主题，虚拟主题在内部与主题行为一致，在与管道交互时在外部与主题行为一致。有关更多信息，请参阅 Apache ActiveMQ 网站上的虚拟目标和 RabbitMQ 网站上的虚拟主机。

  • 网络拓扑 — 对于 ActiveMQ，管道仅支持一个单实例或备用代理。对于 RabbitMQ，每个管道仅支持一个单实例代理或集群部署。单实例代理需要一个失效转移端点。有关这些代理部署模式的更多信息，请参阅 Amazon MQ 开发人员指南中的 Active MQ 代理架构和 Rabbit MQ 代理架构。

  • 协议 — 支持的协议取决于您使用的 Amazon MQ 集成。

    • 对于 ActiveMQ 集成， EventBridge 使用 OpenWire /Java Message Service (JMS) 协议来使用消息。任何其他协议都不支持消息消费。 EventBridge 仅支持 JMS 协议中的TextMessage和BytesMessage操作。有关 OpenWire 协议的更多信息，请参阅 OpenWireApache ActiveMQ 网站上的。

    • 对于 RabbitMQ 集成， EventBridge 使用 AMQP 0-1 协议来使用消息。消息的使用不支持任何其他协议。有关 RabbitMQ 的 AMQP 0-9-1 协议实施的详细信息，请参阅 RabbitMQ 网站上的 AMQP 0-9-1 完整参考指南。

EventBridge 自动支持 Amazon MQ 支持的最新版本的 ActiveMQ 和 RabbitMQ。有关受支持的最新版本，请参阅 Amazon MQ 开发人员指南中的 Amazon MQ 发布说明。

注意

默认情况下，Amazon MQ 代理有一个每周维护时段。代理在该时段内无法使用。对于没有待机状态的经纪商，在窗口结束之前 EventBridge 不会处理消息。

事件示例

以下示例事件显示了管道接收到的信息。您可以使用此事件来创建和筛选事件模式，或者定义输入转换。并非所有字段都可以筛选。有关可以筛选哪些字段的更多信息，请参阅亚马逊 EventBridge 管道筛选。

ActiveMQ

[
  {
    "eventSource": "aws:amq",
    "eventSourceArn": "arn:aws:mq:us-west-2:112556298976:broker:test:b-9bcfa592-423a-4942-879d-eb284b418fc8",    
    "messageID": "ID:b-9bcfa592-423a-4942-879d-eb284b418fc8-1.mq.us-west-2.amazonaws.com-37557-1234520418293-4:1:1:1:1",
    "messageType": "jms/text-message",
    "data": "QUJDOkFBQUE=",
    "connectionId": "myJMSCoID",
    "redelivered": false,
    "destination": {
      "physicalname": "testQueue"
    },
    "timestamp": 1598827811958,
    "brokerInTime": 1598827811958,
    "brokerOutTime": 1598827811959
  },
  {
    "eventSource": "aws:amq",
    "eventSourceArn": "arn:aws:mq:us-west-2:112556298976:broker:test:b-9bcfa592-423a-4942-879d-eb284b418fc8",        
    "messageID": "ID:b-9bcfa592-423a-4942-879d-eb284b418fc8-1.mq.us-west-2.amazonaws.com-37557-1234520418293-4:1:1:1:1",
    "messageType": "jms/bytes-message",
    "data": "3DTOOW7crj51prgVLQaGQ82S48k=",
    "connectionId": "myJMSCoID1",
    "persistent": false,
    "destination": {
      "physicalname": "testQueue"
    },
    "timestamp": 1598827811958,
    "brokerInTime": 1598827811958,
    "brokerOutTime": 1598827811959
  }
]

RabbitMQ

[
  {
    "eventSource": "aws:rmq",
    "eventSourceArn": "arn:aws:mq:us-west-2:111122223333:broker:pizzaBroker:b-9bcfa592-423a-4942-879d-eb284b418fc8",
    "eventSourceKey": "pizzaQueue::/",
    "basicProperties": {
      "contentType": "text/plain",
      "contentEncoding": null,
      "headers": {
        "header1": {
          "bytes": [
            118,
            97,
            108,
            117,
            101,
            49
          ]
        },
        "header2": {
          "bytes": [
            118,
            97,
            108,
            117,
            101,
            50
          ]
        },
        "numberInHeader": 10
      },
      "deliveryMode": 1,
      "priority": 34,
      "correlationId": null,
      "replyTo": null,
      "expiration": "60000",
      "messageId": null,
      "timestamp": "Jan 1, 1970, 12:33:41 AM",
      "type": null,
      "userId": "AIDACKCEVSQ6C2EXAMPLE",
      "appId": null,
      "clusterId": null,
      "bodySize": 80
    },
    "redelivered": false,
    "data": "eyJ0aW1lb3V0IjowLCJkYXRhIjoiQ1pybWYwR3c4T3Y0YnFMUXhENEUifQ=="
  }
]

使用者组

若要与 Amazon MQ 进行交互，请 EventBridge 创建可以从 Amazon MQ 代理中读取的使用者组。使用与管道 UUUID 相同的 ID 创建使用者组。

对于 Amazon MQ 源，会将记录合并为 EventBridge 批处理，然后通过单个有效负载中将其发送到您的函数。要控制行为，您可以配置批处理时段和批处理大小。 EventBridge 提取消息，直到出现下列情况之一为止：

  • 处理过的记录达到最大有效载荷大小 6 MB。

  • 批处理窗口到期。

  • 记录数量达到完整的批次大小。

EventBridge 会将您的批处理转换为单个有效负载，然后调用您的函数。消息不会被持久化或反序列化。相反，使用者组会将其作为字节 BLOB 进行检索。然后以 base64 格式编码为 JSON 有效负载。如果管道为批处理中的任何消息返回错误，将 EventBridge 重试整批消息，直到处理成功或消息过期为止。

执行角色权限

设置管道时，您可以使用现有的执行角色，也可以使用所需权限为您 EventBridge 创建一个执行角色。对于 Amazon MQ，Pip EventBridge es 需要以下权限才能管理与您的Amazon MQ 消息代理相关的资源。如果您要设置自己的执行角色，则必须自己添加这些权限。

注意

如果您不确定访问源代码所需的确切权限范围，请使用 Pip EventBridge es 控制台创建新角色，然后检查策略中列出的操作。

网络配置

默认情况下，在 PubliclyAccessible 标志设置为 false 时创建 Amazon MQ 代理。只有在设置PubliclyAccessible为 true 时，代理才会接收公有 IP 地址。要获得对管道的完全访问权限，您的代理必须使用公有终端节点或提供对 VPC 的访问权限。

如果Amazon MQ 代理不可公开访问，则 EventBridge 必须具有对与代理关联的 Amazon Virtual Private Cloud (Amazon VPC) 资源的访问权限。要访问您的 Amazon MQ 代理的 VPC， EventBridge 需要对来源子网进行出站互联网访问。对于公有子网，它可以是互联网网关，对于私有子网，它需要是网络地址转换器 (NAT)，例如 NAT 网关或您自己的 NAT。确保 NAT 具有公有 IP 地址并且可以连接到互联网。

使用以下规则配置您的 Amazon VPC 安全组（至少）：

  • 入站规则 – 对于没有公共可访问性的代理，允许指定为源的安全组的所有端口上的所有流量。对于具有公共可访问性的代理，允许所有目标的所有端口上的所有流量。

  • 出站规则 – 允许所有目标的所有端口上的所有流量传输。

注意

可通过 Amazon MQ API 发现您的 Amazon VPC 配置。您不需要在设置期间对其进行配置。

以 Amazon MQ 为源配置管道

添加Amazon MQ 来源

使用控制台添加 Amazon MQ 来源

  1. 通过 https://console.aws.amazon.com/events/ 打开亚马逊 EventBridge 主机。

  2. 在导航窗格中，选择 Pips。

  3. 选择 “创建管道”。

  4. 输入管道的名称。

  5. （可选）添加管道的描述。

  6. 在 “生成管道” 选项卡上，为 “来源” 选择 Amazon MQ。

  7. 对于 Amazon MQ 代理，请选择要用作来源的直播。

  8. 在队列名称中，输入管道将从中读取的队列的名称。

  9. 对于 “身份验证方法”，选择 BASIC_AUTH。

  10. 对于密钥，选择密钥。

  11. （可选）对于 Additional sets (其他设置)-可选，执行以下操作：

    1. 在 “Batch 大小-可选” 中，输入每个批次的最大消息数。默认值为 100。

    2. 对于 “Batch 窗口-可选”，输入收集记录的最大秒数，然后再继续。

现在，源已配置完毕，您可以为管道添加可选筛选、可选扩充或目标。

（可选）配置过滤

您可以向管道添加筛选功能，这样您就可以只将 Amazon MQ 队列中的一部分记录发送到目标。

使用控制台配置筛选

  1. 选择 “筛选”。

  2. 在 “示例事件-可选” 下，您将看到一个示例 Amazon MQ 活动，您可以使用它来建立自己的赛事模式，也可以通过选择 Enter your own 来参加自己的活动。

  3. 在 “事件模式” 下，输入要筛选记录的事件模式。有关构建事件模式的更多信息，请参阅Amazon EventBridge 事件模式。

    以下是仅在 “城市” 字段中发送值为 Seatt le 的事件的示例事件模式。

    {
  "data": {
    "City": ["Seattle"]
  }
}

现在事件已被筛选，您可以添加可选的增强功能和管道目标。

（可选）定义浓缩

您可以将用于丰富的事件数据发送到 Lambda 函数、Amazon Step Functions状态机、Amazon API Gateway 或 API 目标。

选择充实

  1. 选择 “充实”。

  2. 在 “详细信息” 下的 “服务” 中，选择要用于充实内容的服务和相关设置。

您也可以在发送数据进行增强之前对其进行转换。

（可选）定义输入变压器

  1. 选择增强输入变压器-可选。

  2. 对于示例事件/事件负载，选择示例事件类型。

  3. 对于 Transformer，输入转换器语法，例如"Event happened at <$.detail.field>."其中<$.detail.field>对示例事件中字段的引用。您也可以双击示例事件中的某个字段，将其添加到转换器中。

  4. 对于 Output，验证输出是否符合您的期望。

既然数据已经过筛选和增强，您必须定义要将事件数据发送到的目标。

配置目标

配置目标

  1. 选择目标。

  2. 在 “详细信息” 下，为 “目标服务” 选择目标。显示的字段因您所选的目标而异。根据需要输入特定于该目标类型的信息。

您也可以在将数据发送到目标之前对其进行转换。

（可选）定义输入变压器

  1. 选择目标输入变压器-可选。

  2. 对于示例事件/事件负载，选择示例事件类型。

  3. 对于 Transformer，输入转换器语法，例如"Event happened at <$.detail.field>."其中<$.detail.field>对示例事件中字段的引用。您也可以双击示例事件中的某个字段，将其添加到转换器中。

  4. 对于 Output，验证输出是否符合您的期望。

现在管道已配置完毕，请确保其设置配置正确。

配置管道设置

默认情况下，管道处于活动状态，但您可以将其禁用。您也可以指定管道的权限并添加标签。

配置管道设置

  1. 选择 “管道设置” 选项卡。

  2. 默认情况下，新创建的管道在创建后立即处于活动状态。如果要为激活管道创建非活动管道，请在 “激活” 下为 “激活管道” 关闭 “活动”。

  3. 在 “权限” 下，对于 “执行角色”，执行以下操作之一：

    1. 若要为此管道 EventBridge 创建新的执行角色，请选择为此特定资源创建新角色。 在 “角色名称” 下，您可以选择编辑角色名称。

    2. 要使用现有执行角色，请选择 “使用现有角色”。在 “角色名称” 下，选择角色。

  4. （可选）在 Tags-可选下，选择添加新标签，然后为规则输入一个或多个标签。有关更多信息，请参阅 亚马逊 EventBridge 标签。

  5. 选择 “创建管道”。