搭配使用 Lambda 與 Amazon MQ - AWS Lambda
了解 Amazon MQ 的 Lambda 消費者群組

本文為英文版的機器翻譯版本，如內容有任何歧義或不一致之處，概以英文版為準。

搭配使用 Lambda 與 Amazon MQ

注意

如果您想要將資料傳送到 Lambda 函數以外的目標，或在傳送資料之前豐富資料，請參閱 Amazon EventBridge 管道。

Amazon MQ 是一項受管訊息代理程式服務，適用於 Apache ActiveMQ 和 RabbitMQ。訊息代理程式透過主題或佇列事件目的地，允許軟體應用程式和元件使用各種程式設計語言、作業系統和正式簡訊協定來進行通訊。

Amazon MQ 也可以透過安裝 ActiveMQ 或 RabbitMQ 代理程式，並提供不同的網路拓撲和其他基礎設施需求，代表您管理 Amazon 彈性運算雲端 (AmazonEC2) 執行個體。

您可以使用 Lambda 函數來處理您的 Amazon MQ 訊息代理程式中的記錄。Lambda 透過事件來源映射叫用函數，這是從您的代理程式讀取訊息並同步叫用函數的 Lambda 資源。

警告

Lambda 事件來源對應至少處理每個事件一次，並且可能會重複處理記錄。為了避免與重複事件相關的潛在問題，我們強烈建議您將函數代碼設為冪等。若要深入了解，請參閱 AWS 知識中心如何讓 Lambda 函數具有冪等性。

Amazon MQ 事件來源映射具有下列組態限制：

  • 並行：使用 Amazon MQ 事件來源映射的 Lambda 函數具有預設並行上限設定。對於 ActiveMQ，Lambda 服務會將並行執行環境的數量限制為每個 Amazon MQ 事件來源對應的五個。對於 RabbitMQ，並行執行環境的數量限制為每個 Amazon MQ 事件來源映射 1 個。即使您變更函數的保留或佈建並行設定，Lambda 服務還是無法提供更多可用的執行環境。若要申請增加單一 Amazon MQ 事件來源映射的預設最大並行數量，請 AWS Support 聯絡事件來源映射UUID以及區域。由於增加是在特定事件來源映射層級 (而非帳戶或區域層級) 套用，因此您需要手動請求每個事件來源映射的調整比例增加。

  • 跨帳戶 - Lambda 不支援跨帳戶處理。您無法用 Lambda 處理來自不同 AWS 帳戶中 Amazon MQ 訊息代理程式的記錄。

  • 驗證 — 對於 ActiveMQ，僅支援 ActiveM SimpleAuthenticationPluginQ。對於 RabbitMQ，僅支援PLAIN驗證機制。使用者必須使用 AWS Secrets Manager 來管理其認證。如需 ActiveMQ 身份驗證的詳細資訊，請參閱 Amazon MQ 開發人員指南LDAP中的整合 ActiveMQ 代理程式。

  • 連線配額 - 代理程式每個線路層級協定允許的連線數量上限。此配額以代理程式執行個體類型為基礎。如需詳細資訊，請參閱 Amazon MQ 開發人員指南中的 Amazon MQ 中的配額的代理程式。

  • 連線能力 — 您可以在公有或私有虛擬私有雲 (VPC) 中建立代理程式。對於私有VPCs，您的 Lambda 函數需要存取VPC以接收訊息。如需詳細資訊，請參閱本節設定網路存取稍後的內容。

  • 事件目的地 - 僅支援佇列目的地。然而，您可以使用虛擬主題，當作為佇列與 Lambda 互動時，其行為在內部可作為主題。如需詳細資訊，請參閱 Apache ActiveMQ 網站上的虛擬目的地，以及 RabbitMQ 網站上的虛擬主機。

  • 網路拓撲 - 對於 ActiveMQ，每個事件來源映射僅支援單一執行個體或待命代理程式。對於 RabbitMQ，每個事件來源映射僅支援單一執行個體代理程式或叢集部署。單一執行個體代理程式需要容錯移轉端點。如需這些代理程式部署模式的詳細資訊，請參閱 Amazon MQ 開發人員指南中的 ActiveMQ 代理程式架構和 Rabbit MQ 代理程式架構。

  • 協定 - 支援的協定取決於 Amazon MQ 整合的類型。

    • 若是 ActiveMQ 整合，Lambda 會使用 OpenWire /Java 訊息服務 () JMS 通訊協定取用訊息。不支援透過其他協定來取用訊息。在JMS協議中，僅支持TextMessage和BytesMessage。Lambda 也支援JMS自訂屬性。如需有關通訊 OpenWire 協定的詳細資訊，請參閱 Apache ActiveMQ 網OpenWire站上的。

    • 對於 RabbitMQ 整合，Lambda 會使用 0-9-1 通訊協定AMQP取用訊息。不支援透過其他協定來取用訊息。有關 RabbitMQ 實現 AMQP 0-9-1 協議的更多信息，請參閱 RabbitMQ 網站上的 AMQP0-9-1 完整參考指南。

Lambda 會自動支援 Amazon MQ 支援的最新版 ActiveMQ 和 RabbitMQ。如需支援的最新版，請參閱 Amazon MQ 開發人員指南中的 Amazon MQ 版本備註。

注意

根據預設，Amazon MQ 具有每週代理程式維護時段。代理程式在該時段不可用。若是無待命狀態的代理程式，則 Lambda 無法在該時段處理任何訊息。

主題

了解 Amazon MQ 的 Lambda 消費者群組

若要與 Amazon MQ 互動，Lambda 會建立可以從您的 Amazon MQ 代理程式讀取的取用者群體。建立用戶群組時，會使用與事件來源對應相同的識別碼UUID。

對於 Amazon MQ 事件來源，Lambda 會批次處理記錄，並在單個承載中將它們傳送到您的函數。要控制行為，您可以設定批次間隔和批次大小。Lambda 會提取訊息，直到它處理最大為 6 MB 的承載大小、批次間隔過期或記錄數達到完整批次大小。如需詳細資訊，請參閱批次處理行為。

用戶群組以位元組BLOB的形式擷取訊息，base64 將它們編碼為單一JSON裝載，然後叫用您的函數。如果函數針對批次中的任何訊息傳回錯誤，Lambda 會重試整個批次的訊息，直至處理成功或訊息過期。

注意

雖然 Lambda 函數的逾時限制通常為 15 分鐘，但適用於 Amazon 的事件來源對應MSK、自我管理的 Apache 卡夫卡、Amazon DocumentDB 以及 ActiveMQ 和 RabbitMQ 的 Amazon MQ 只支援 14 分鐘逾時限制上限的函數。此限制條件可確保事件來源映射能夠正確處理函數錯誤和重試。

您可以使用 Amazon CloudWatch 中的指ConcurrentExecutions標監視給定函數的並行使用情況。如需並行的詳細資訊，請參閱為函數配置保留並發。

範例 Amazon MQ 記錄事件
ActiveMQ
{
   "eventSource": "aws:mq",
   "eventSourceArn": "arn:aws:mq:us-east-2:111122223333:broker:test:b-9bcfa592-423a-4942-879d-eb284b418fc8",
   "messages": [
      { 
        "messageID": "ID:b-9bcfa592-423a-4942-879d-eb284b418fc8-1---mq---us-east-2.amazonaws.com.rproxy.goskope.com-37557-1234520418293-4:1:1:1:1", 
        "messageType": "jms/text-message",
        "deliveryMode": 1,
        "replyTo": null,
        "type": null,
        "expiration": "60000",
        "priority": 1,
        "correlationId": "myJMSCoID",
        "redelivered": false,
        "destination": { 
          "physicalName": "testQueue" 
        },
        "data":"QUJDOkFBQUE=",
        "timestamp": 1598827811958,
        "brokerInTime": 1598827811958, 
        "brokerOutTime": 1598827811959, 
        "properties": {
          "index": "1",
          "doAlarm": "false",
          "myCustomProperty": "value"
        }
      },
      { 
        "messageID": "ID:b-9bcfa592-423a-4942-879d-eb284b418fc8-1---mq---us-east-2.amazonaws.com.rproxy.goskope.com-37557-1234520418293-4:1:1:1:1",
        "messageType": "jms/bytes-message",
        "deliveryMode": 1,
        "replyTo": null,
        "type": null,
        "expiration": "60000",
        "priority": 2,
        "correlationId": "myJMSCoID1",
        "redelivered": false,
        "destination": { 
          "physicalName": "testQueue" 
        },
        "data":"LQaGQ82S48k=",
        "timestamp": 1598827811958,
        "brokerInTime": 1598827811958, 
        "brokerOutTime": 1598827811959, 
        "properties": {
          "index": "1",
          "doAlarm": "false",
          "myCustomProperty": "value"
        }
      }
   ]
}
RabbitMQ

{
  "eventSource": "aws:rmq",
  "eventSourceArn": "arn:aws:mq:us-east-2:111122223333:broker:pizzaBroker:b-9bcfa592-423a-4942-879d-eb284b418fc8",
  "rmqMessagesByQueue": {
    "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=="
      }
    ]
  }
}
注意

在 RabbitMQ 範例中，pizzaQueue 是 RabbitMQ 佇列的名稱，/ 是虛擬主機的名稱。接收訊息時，事件來源會在 pizzaQueue::/ 列出訊息。