在佇列模式中輪詢 Kafka 主題

佇列模式是一種使用共享群組的耗用模式，可在 Apache Kafka 4.0 中使用 (Kafka 佇列輪詢器需要 4.2 版或更新版本）。透過佇列模式，您可以使用共用群組，以無順序、高輸送量的方式處理 Kafka 主題訊息。與僅將分割區指派給取用者的傳統 Kafka 取用者群組不同，共用群組可讓多個取用者同時從相同的分割區讀取。這類似於 Amazon SQS 如何處理來自佇列的訊息。

當您為自我管理Apache Kafka叢集建立事件來源映射時，您可以選擇兩種取用模式：

• 串流 （預設）。使用具有每個分割區排序的 Kafka 取用者群組。Lambda 會將每個分割區指派給一個取用者。

• 佇列。使用Apache Kafka在 4.0 中可用的共用群組，進行未排序的高輸送量處理 (Kafka 佇列輪詢器需要 4.2 版或更新版本）。透過共用群組，多個取用者可以同時從相同的分割區讀取，類似於 Amazon SQS 如何處理佇列中的訊息。

如果您未指定耗用模式，事件來源映射會使用 串流。此頁面涵蓋佇列耗用模式。如需搭配自我管理 使用 Lambda 的一般資訊Apache Kafka，請參閱 搭配使用 Lambda 與自我管理 Apache Kafka。

在佇列模式中，Lambda 使用 Kafka 共用取用者從主題輪詢訊息。每個事件輪詢器都會處理所有分割區的訊息，而不需要分割區層級排序。這會增加每個輪詢器的並行。

最低需求

若要使用佇列耗用模式，您的 Kafka 叢集必須符合下列要求：

• Apache Kafka 啟用共用群組的 4.2 版或更新版本。Kafka share.version 預設會啟用 ，而且您可以透過 管理它kafka-features.sh。

• 僅限自我管理 Apache Kafka(SMK) 叢集。Amazon Managed Streaming for Apache Kafka(Amazon MSK) 目前不支援佇列模式。一旦 MSK 在啟用共享群組的情況下提供 Kafka 4.2 或更新版本， 就會 AWS 新增 MSK 支援。

串流和佇列比較

功能	串流 （預設）	佇列
排序	每個分割區的排序保證	無訂購保證
消費者模型	取用者群組 （獨佔分割區指派）	共用群組 （並行分割區存取）
每個事件輪詢器的輸送量	高達 5 MB/s	高達 5 MB/s
每個事件輪詢器的並行調用	最高 5	最高 10
擴展模式	隨需和佈建	隨需和佈建
EPU 共用	可以與其他 Kafka Streams 事件來源映射共用 EPUs	隔離。佇列事件來源映射只會與其他佇列事件來源映射共用
使用案例	排序的事件處理	高輸送量無順序工作負載

建立佇列映射

若要建立佇列模式事件來源映射，請將 ConsumptionMode 欄位設定為 SelfManagedKafkaEventSourceConfig Queue。在下列範例中，將 my-function、中介裝置端點和 取代my-queue-topic為您自己的值。

AWS CLI

下列 AWS CLI 命令會建立佇列模式事件來源映射。將 my-function 取代為您的函數名稱、將中介裝置端點取代為您的中介裝置，並將 my-queue-topic 取代為您的主題。

PROMPT> aws lambda create-event-source-mapping \
  --function-name my-function \
  --self-managed-event-source '{"Endpoints":{"KAFKA_BOOTSTRAP_SERVERS":["broker1.us-west-2.example.com:9092","broker2.us-west-2.example.com:9092"]}}' \
  --topics my-queue-topic \
  --batch-size 100 \
  --self-managed-kafka-event-source-config '{"ConsumptionMode": "Queue"}' \
  --provisioned-poller-config '{"MinimumPollers": 1, "MaximumPollers": 10}'

AWS CloudFormation

下列 AWS CloudFormation 範本會建立佇列模式事件來源映射。將中介裝置端點和 my-queue-topic 取代為您自己的值。

MyKafkaQueueESM:
  Type: AWS::Lambda::EventSourceMapping
  Properties:
    FunctionName: !Ref MyFunction
    SelfManagedEventSource:
      Endpoints:
        KafkaBootstrapServers:
          - broker1.us-west-2.example.com:9092
          - broker2.us-west-2.example.com:9092
    Topics:
      - my-queue-topic
    BatchSize: 100
    SelfManagedKafkaEventSourceConfig:
      ConsumptionMode: Queue
    ProvisionedPollerConfig:
      MinimumPollers: 1
      MaximumPollers: 10

組態參數

下表列出適用於佇列模式事件來源映射的參數。

參數	描述	範圍/預設
BatchSize	每個批次傳送至函數的記錄數目上限	1–10，000。預設：100
Enabled	事件來源映射是否作用中	true 或 false 預設：true
FilterCriteria	僅處理相符記錄的事件篩選模式	最多 5 個篩選模式
MaximumRetryAttempts	失敗記錄的重試次數上限	-1–10，000。預設：0 （在佇列模式中，Kafka 代理程式會在未設定故障時目的地時執行重試）
BisectBatchOnFunctionError	函數錯誤時將批次分割成一半	true 或 false 預設：false
ProvisionedPollerConfig.MinimumPollers	最低事件輪詢器 （佈建模式）	1–200。預設：1
ProvisionedPollerConfig.MaximumPollers	事件輪詢器上限 （佈建模式）	1–2，000。預設：200
ProvisionedPollerConfig.PollerGroupName	EPU 共用的群組名稱 （佈建模式，僅限佇列事件來源映射）	字串. 每個群組最多 100 個事件來源映射
SelfManagedKafkaEventSourceConfig.ConsumptionMode	事件來源的使用模式	必須Queue用於佇列模式
SchemaRegistryConfig	事件驗證的結構描述登錄檔	請參閱 在 Lambda 中將結構描述登錄檔與 Kafka 事件來源搭配使用

重要

佇列模式事件來源映射不支援 MaximumBatchingWindowInSeconds 參數。Kafka Share Consumer 使用明確認可。必須先明確確認批次中的所有記錄，消費者才能擷取新的記錄。這與以時間為基礎的批次處理時段不相容。如果您在建立佇列模式事件來源映射MaximumBatchingWindowInSeconds時指定 ，API 會傳回驗證錯誤。

支援的事件來源映射功能

下表列出佇列模式可用的 Lambda 事件來源映射功能。

功能	支援	備註
佈建模式	是	設定 MinimumPollers和 MaximumPollers
隨需模式	是	預設擴展模式。Lambda 會自動擴展輪詢器，最高可達 5 MB/s 或 50 個並行調用，以先達到者為準。不需要額外的組態。
事件篩選	是	每個事件來源映射最多 5 個篩選條件模式
結構描述登錄檔	是	Avro 和 Protobuf 搭配 Glue 結構描述登錄檔或 Confluent 結構描述登錄檔。結構描述登錄檔篩選僅適用於佈建模式。
低延遲處理	是	在佈建模式中可用
重試組態	是	MaximumRetryAttempts, BisectBatchOnFunctionError
失敗時目的地 (Kafka 主題）	是	將失敗的記錄傳送至 Kafka 主題
失敗時目的地 (SNS/SQS)	是	將失敗的記錄傳送至 SNS 或 SQS
事件來源映射記錄	是	事件來源映射診斷的 CloudWatch Logs
事件來源映射分組 (PollerGroupName)	是	佇列模式事件來源映射只能與其他佇列模式事件來源映射共用 EPUs （佈建模式）
批次處理時段	否	由於共用取用者封鎖模式行為而不支援
開始位置	否	不支援 StartingPosition 和 StartingPositionTimestamp。佇列模式一律會從共用群組的遞交位置讀取。

擴展模式

佇列模式事件來源映射支援隨需 （預設） 和佈建擴展模式。

隨需模式 (預設)

在隨需模式中，Lambda 會自動管理佇列模式事件來源映射的事件輪詢器。

隨需模式每個事件來源映射最多可擴展 5 個事件輪詢器。它會以每秒 5 MB 的彙總輸送量或 50 個並行調用 (5 個輪詢器乘以 10) 達到其上限。這會先達到任何限制。若要擴展超過這些限制，請使用佈建模式。隨需模式不需要超出標準事件來源映射參數的額外組態。

下表列出隨需模式限制。

限制	Value
每個事件輪詢器的輸送量	高達 1 MB/s
每個事件輪詢器的並行調用	最高 10
事件輪詢器上限	5
最大彙總輸送量	最高 5 MB/s (5 個輪詢器乘以 1 MB/s)
彙總並行上限	最多 50 個並行調用 (5 個輪詢器乘以 10)

範例 – 隨需模式 (AWS CLI) 中的佇列模式事件來源映射

下列 AWS CLI 命令會在隨需模式中建立佇列模式事件來源映射。將 my-function、代理程式端點、my-queue-topic、子網路 IDs 和安全群組 ID 取代為您自己的值。

PROMPT> aws lambda create-event-source-mapping \
  --function-name my-function \
  --self-managed-event-source '{"Endpoints":{"KAFKA_BOOTSTRAP_SERVERS":["broker1.us-west-2.example.com:9092","broker2.us-west-2.example.com:9092"]}}' \
  --topics my-queue-topic \
  --batch-size 100 \
  --self-managed-kafka-event-source-config '{"ConsumptionMode": "Queue"}' \
  --source-access-configurations '[{"Type":"VPC_SUBNET","URI":"subnet-0123456789abcdef0"},{"Type":"VPC_SUBNET","URI":"subnet-0123456789abcdef1"},{"Type":"VPC_SECURITY_GROUP","URI":"sg-0123456789abcdef0"}]'

佈建模式

對於具有嚴格效能要求或尖峰流量模式的工作負載，請使用佈建模式。佈建模式可讓您直接控制佇列模式事件來源映射專用的事件輪詢器數量下限和上限。

佈建模式中的每個佇列模式事件輪詢器最多支援 5 MB/s 的輸送量和最多 10 個並行 Lambda 調用。

下表列出佈建模式限制。

限制	Value
每個事件輪詢器的輸送量	高達 5 MB/s
每個事件輪詢器的並行調用	最高 10
最低事件輪詢器 (MinimumPollers)	1–200。預設：1
事件輪詢器上限 (MaximumPollers)	1–2，000。預設：200
每個輪詢器群組的事件來源映射	最高 100
每個輪詢器群組的彙總輪詢器上限	2,000
每個 EPU 的輸送量 （彙總）	每個輪詢器最多 5 MB/s；彙總取決於輪詢器封裝密度

佈建模式會根據事件輪詢單元 (EPU) 用量收費。如需定價詳細資訊，請參閱 AWS Lambda 定價。

範例 – 使用佈建模式的佇列模式事件來源映射 (AWS CLI)

下列 AWS CLI 命令會在佈建模式中建立佇列模式事件來源映射。

PROMPT> aws lambda create-event-source-mapping \
  --function-name my-function \
  --self-managed-event-source '{"Endpoints":{"KAFKA_BOOTSTRAP_SERVERS":["broker1.us-west-2.example.com:9092","broker2.us-west-2.example.com:9092"]}}' \
  --topics my-queue-topic \
  --batch-size 100 \
  --self-managed-kafka-event-source-config '{"ConsumptionMode": "Queue"}' \
  --provisioned-poller-config '{"MinimumPollers": 5, "MaximumPollers": 100}' \
  --source-access-configurations '[{"Type":"VPC_SUBNET","URI":"subnet-0123456789abcdef0"},{"Type":"VPC_SUBNET","URI":"subnet-0123456789abcdef1"},{"Type":"VPC_SECURITY_GROUP","URI":"sg-0123456789abcdef0"}]'

將下列預留位置取代為您自己的值：

• subnet-0123456789abcdef0 和 subnet-0123456789abcdef1：您的子網路 IDs。

• sg-0123456789abcdef0：您的安全群組 ID。

• MinimumPollers 和 MaximumPollers：根據您的輸送量需求進行設定。

注意

若要從佈建模式切換回隨需模式，請傳遞空的 ProvisionedPollerConfig：

PROMPT> aws lambda update-event-source-mapping \
  --uuid a1b2c3d4-5678-90ab-cdef-EXAMPLE11111 \
  --provisioned-poller-config '{}'

擴展模式選擇

考量事項	隨需	佈建
設定	沒有其他組態	設定 MinimumPollers和 MaximumPollers
成本	無需額外的 EPU 費用	以每小時用量為基礎的 EPU 費用
最適合	低並行工作負載	嚴格延遲要求、尖峰流量
事件來源映射分組 (PollerGroupName)	不支援	支援。在佇列事件來源映射之間共用 EPUs
低延遲處理	不適用	Available
結構描述登錄檔篩選 (Avro/Protobuf)	不適用	Available

EPU 共用隔離

在佈建模式中，您可以將相同 VPC 中的多個事件來源映射分組，透過指派相同的 來共用事件輪詢單元 (EPU) 容量PollerGroupName。事件來源映射分組可讓低輸送量事件來源映射封裝至共用 EPUs，而不是每個都需要專用容量，從而降低佈建模式成本。如需詳細資訊，請參閱佈建模式的成本最佳化。

佇列模式事件來源映射只會與其他佇列模式事件來源映射共用 EPUs。佇列模式事件來源映射PollerGroupName所使用的 無法與 Kafka Streams 事件來源映射共用，反之亦然。如果您設定PollerGroupName已由其他取用模式使用的 ， 會CreateEventSourceMapping傳回驗證錯誤。若要將事件來源映射分組與佇列模式搭配使用，請建立PollerGroupName專用於佇列事件來源映射的 。

範例事件

當 Lambda 使用佇列模式事件來源調用函數時，事件承載與標準 Kafka 事件相同。每個記錄都包含主題、分割區、位移、時間戳記和 base64 編碼的索引鍵和值：

{
  "eventSource": "SelfManagedKafka",
  "bootstrapServers": "broker1.example.com:9092,broker2.example.com:9092",
  "records": {
    "my-queue-topic-0": [
      {
        "topic": "my-queue-topic",
        "partition": 0,
        "offset": 15,
        "timestamp": 1545084650987,
        "timestampType": "CREATE_TIME",
        "key": "abcDEFghiJKLmnoPQRstuVWXyz1234==",
        "value": "SGVsbG8sIHRoaXMgaXMgYSB0ZXN0Lg==",
        "headers": [
          {
            "headerKey": [104, 101, 97, 100, 101, 114, 86, 97, 108, 117, 101]
          }
        ]
      }
    ],
    "my-queue-topic-3": [
      {
        "topic": "my-queue-topic",
        "partition": 3,
        "offset": 42,
        "timestamp": 1545084651000,
        "timestampType": "CREATE_TIME",
        "key": "YW5vdGhlcktleQ==",
        "value": "VGhpcyBpcyBhbm90aGVyIG1lc3NhZ2U=",
        "headers": []
      }
    ]
  }
}

需要冪等性

Lambda 事件來源映射至少會處理每個事件一次，因此可能會重複處理記錄。由於佇列模式不保證排序，如果您的函數程式碼不等冪，重複或out-of-order處理可能會產生不正確的結果，例如重複計數或不一致狀態。讓您的函數程式碼成為等冪。

限制

• 僅限自我管理的 Kafka 叢集。啟動時不支援 Amazon MSK。

• Apache Kafka 需要 4.2 或更新版本。叢集必須支援已啟用共用群組的共用取用者通訊協定。

• 不支援批次處理時段。MaximumBatchingWindowInSeconds 參數無法與 Kafka 佇列事件來源映射搭配使用。

• 不支援起始位置。 StartingPosition和 StartingPositionTimestamp無法與佇列模式事件來源映射搭配使用。共用群組會從其遞交的位移位置耗用 。

• EPU 隔離。佇列模式事件來源映射無法與 Kafka Streams 事件來源映射共用 EPUs。使用個別PollerGroupName的值。