從 Amazon MSK 觸發器調用 Lambda 函數

下列程式碼範例說明如何實作 Lambda 函數，該函數會接收透過從 Amazon MSK 叢集接收記錄而觸發的事件。該函數檢索MSK有效負載並記錄記錄內容。

.NET

AWS SDK for .NET

注意

還有更多關於 GitHub。尋找完整範例，並了解如何在無伺服器範例儲存庫中設定和執行。

消費 Amazon MSK 事件 Lambda 使用. NET。

using System.Text;
using Amazon.Lambda.Core;
using Amazon.Lambda.KafkaEvents;


// Assembly attribute to enable the Lambda function's JSON input to be converted into a .NET class.
[assembly: LambdaSerializer(typeof(Amazon.Lambda.Serialization.SystemTextJson.DefaultLambdaJsonSerializer))]

namespace MSKLambda;

public class Function
{
    
    
    /// <param name="input">The event for the Lambda function handler to process.</param>
    /// <param name="context">The ILambdaContext that provides methods for logging and describing the Lambda environment.</param>
    /// <returns></returns>
    public void FunctionHandler(KafkaEvent evnt, ILambdaContext context)
    {

        foreach (var record in evnt.Records)
        {
            Console.WriteLine("Key:" + record.Key); 
            foreach (var eventRecord in record.Value)
            {
                var valueBytes = eventRecord.Value.ToArray();    
                var valueText = Encoding.UTF8.GetString(valueBytes);
                
                Console.WriteLine("Message:" + valueText);
            }
        }
    }
    

}

Java

SDK對於爪哇 2.x

注意

還有更多關於 GitHub。尋找完整範例，並了解如何在無伺服器範例儲存庫中設定和執行。

使用 Java 使用 Lambda 消費 Amazon MSK 事件。

import com.amazonaws.services.lambda.runtime.Context;
import com.amazonaws.services.lambda.runtime.RequestHandler;
import com.amazonaws.services.lambda.runtime.events.KafkaEvent;
import com.amazonaws.services.lambda.runtime.events.KafkaEvent.KafkaEventRecord;

import java.util.Base64;
import java.util.Map;

public class Example implements RequestHandler<KafkaEvent, Void> {

    @Override
    public Void handleRequest(KafkaEvent event, Context context) {
        for (Map.Entry<String, java.util.List<KafkaEventRecord>> entry : event.getRecords().entrySet()) {
            String key = entry.getKey();
            System.out.println("Key: " + key);

            for (KafkaEventRecord record : entry.getValue()) {
                System.out.println("Record: " + record);

                byte[] value = Base64.getDecoder().decode(record.getValue());
                String message = new String(value);
                System.out.println("Message: " + message);
            }
        }

        return null;
    }
}

JavaScript

SDK對於 JavaScript （3）

注意

還有更多關於 GitHub。尋找完整範例，並了解如何在無伺服器範例儲存庫中設定和執行。

消費 Amazon MSK 事件 Lambda 使用 JavaScript.

exports.handler = async (event) => {
    // Iterate through keys
    for (let key in event.records) {
      console.log('Key: ', key)
      // Iterate through records
      event.records[key].map((record) => {
        console.log('Record: ', record)
        // Decode base64
        const msg = Buffer.from(record.value, 'base64').toString()
        console.log('Message:', msg)
      }) 
    }
}

Python

SDK對於 Python（肉毒桿菌 3）

注意

還有更多關於 GitHub。尋找完整範例，並了解如何在無伺服器範例儲存庫中設定和執行。

使用 Python 使用 Lambda 消費 Amazon MSK 事件。

import base64

def lambda_handler(event, context):
    # Iterate through keys
    for key in event['records']:
        print('Key:', key)
        # Iterate through records
        for record in event['records'][key]:
            print('Record:', record)
            # Decode base64
            msg = base64.b64decode(record['value']).decode('utf-8')
            print('Message:', msg)

Ruby

SDK對於紅寶石

注意

還有更多關於 GitHub。尋找完整範例，並了解如何在無伺服器範例儲存庫中設定和執行。

使用紅寶石使用 Lambda 消費 Amazon MSK 事件。

require 'base64'

def lambda_handler(event:, context:)
  # Iterate through keys
  event['records'].each do |key, records|
    puts "Key: #{key}"

    # Iterate through records
    records.each do |record|
      puts "Record: #{record}"

      # Decode base64
      msg = Base64.decode64(record['value'])
      puts "Message: #{msg}"
    end
  end
end

如需 AWS SDK開發人員指南和程式碼範例的完整清單，請參閱使用 Lambda AWS SDK。本主題也包含有關入門的資訊以及舊SDK版的詳細資訊。