Lambda マネージドインスタンスの Java ランタイム

Java ランタイムの場合、Lambda マネージドインスタンスは同時実行に OS スレッドを使用します。Lambda は初期化中、実行環境ごとにハンドラーオブジェクトを 1 回ロードし、複数のスレッドを作成します。これらのスレッドは並行して実行されるため、状態と共有リソースをスレッドセーフに処理する必要があります。各スレッドは、同じハンドラーオブジェクトと静的フィールドを共有します。

同時実行数の設定

Lambda が各実行環境に送信する同時リクエストの最大数は、関数設定の PerExecutionEnvironmentMaxConcurrency 設定で制御できます。これはオプションの設定となっており、デフォルト値はランタイムによって異なります。Java ランタイムの場合の同時リクエスト数は、vCPU あたり 32 件がデフォルト値になっていますが、独自の値を設定することもできます。この値により、Java ランタイムで使用されるスレッドの数も決まります。Lambda は、各実行環境の容量に応じて、設定された最大数までの同時リクエストの数を自動的に調整し、それらのリクエストを取得します。

同時実行のための関数の構築

Lambda マネージドインスタンスを使用する場合は、他のマルチスレッド環境と同じスレッドセーフな環境を適用する必要があります。ハンドラーオブジェクトはすべてのランタイムワーカースレッドで共有されるため、ミュータブルな状態はスレッドセーフである必要があります。これには、コレクション、データベース接続、およびリクエスト処理中に変更される静的オブジェクトが含まれます。

AWS SDK クライアントはスレッドセーフなため、特別な処理は必要ありません。

例: データベース接続プール

次のコードでは、スレッド間で共有される静的データベース接続オブジェクトを使用します。使用する接続ライブラリによっては、スレッドセーフではない場合があります。

public class DBQueryHandler implements RequestHandler<Object, String> {
    // Single connection shared across all threads - NOT SAFE
    private static Connection connection;

    public DBQueryHandler() {
        this.connection = DriverManager.getConnection(jdbcUrl, username, password);
    }

    @Override
    public String handleRequest(Object input, Context context) {
        PreparedStatement stmt = connection.prepareStatement(query);
        ResultSet rs = stmt.executeQuery();
        // Multiple threads using same connection causes issues
        return result.toString();
    }
}

接続プールを使用することで、スレッドセーフにアプローチできます。次の例では、関数ハンドラーはプールから接続を取得します。接続は 1 つのリクエストのコンテキストでのみ使用されます。

public class DBQueryHandler implements RequestHandler<Object, String> {

    private static HikariDataSource dataSource;

    public DBQueryHandler() {
        HikariConfig config = new HikariConfig();
        config.setJdbcUrl("jdbc:mysql://localhost:3306/your_database");
        dataSource = new HikariDataSource(config); // Create pool once per Lambda container
    }

    @Override
    public String handleRequest(Object input, Context context) {
        String query = "SELECT column_name FROM your_table LIMIT 10";
        StringBuilder result = new StringBuilder("Data:\n");

        // try-with-resources automatically calls close() on the connection,
        // which returns it to the HikariCP pool (does NOT close the physical DB connection)
        try (Connection connection = dataSource.getConnection();
             PreparedStatement stmt = connection.prepareStatement(query);
             ResultSet rs = stmt.executeQuery()) {

            while (rs.next()) {
                result.append(rs.getString("column_name")).append("\n");
            }

        } catch (Exception e) {
            context.getLogger().log("Error: " + e.getMessage());
            return "Error";
        }

        return result.toString();
    }
}

例: コレクション

以下の標準 Java コレクションはスレッドセーフではありません。

public class Handler implements RequestHandler<Object, String> {
    private static List<String> items = new ArrayList<>();
    private static Map<String, Object> cache = new HashMap<>();

    @Override
    public String handleRequest(Object input, Context context) {
        items.add("list item");  // Not thread-safe
        cache.put("key", input); // Not thread-safe
        return "Success";
    }
}

代わりに、以下のスレッドセーフコレクションを使用します。

public class Handler implements RequestHandler<Object, String> {
    private static final List<String> items = 
        Collections.synchronizedList(new ArrayList<>());
    private static final ConcurrentHashMap<String, Object> cache = 
        new ConcurrentHashMap<>();

    @Override
    public String handleRequest(Object input, Context context) {
        items.add("list item");  // Thread-safe
        cache.put("key", input); // Thread-safe
        return "Success";
    }
}

共有の /tmp ディレクトリ

/tmp ディレクトリは、実行環境内のすべての同時リクエスト間で共有されます。同じファイルに同時に書き込むと、別のプロセスによってファイルが上書きされるなど、データの破損が発生する可能性があります。これに対処するには、共有ファイルのファイルロックを実装するか、スレッドごとまたはリクエストごとに一意のファイル名を使用して競合を回避します。空き容量を占有しないよう、不要なファイルはクリーンアップしてください。

ログ記録

ログインターリービング (ログでインターリーブされるさまざまなリクエストからのログエントリ) は、複数同時実行システムでは正常です。

Lambda マネージドインスタンスを使用する関数は常に、高度なログ記録コントロールで導入した構造化された JSON ログ形式を使用します。この形式には requestId が含まれ、ログエントリを 1 つのリクエストに関連付けることができます。context.getLogger() から LambdaLogger オブジェクトを使用すると、requestId は各ログエントリに自動的に含まれます。詳細については、「Java での Lambda の高度なログ記録コントロールの使用」を参照してください。

リクエストコンテキスト

context オブジェクトはリクエストスレッドにバインドされます。context.getAwsRequestId() を使用すると、現在のリクエストのリクエスト ID にスレッドセーフにアクセスできます。

context.getXrayTraceId() を使用して X-Ray トレース ID にアクセスします。これにより、現在のリクエストのトレース ID にスレッドセーフにアクセスできます。Lambda は、Lambda マネージドインスタンスの _X_AMZN_TRACE_ID 環境変数をサポートしていません。AWS SDK を使用すると、X-Ray トレース ID が自動的に伝播されます。

プログラムで仮想スレッドを使用する場合、または初期化中にスレッドを作成する場合は、必要なリクエストコンテキストをこれらのスレッドに渡す必要があります。

初期化とシャットダウン

関数の初期化は、実行環境ごとに 1 回行われます。初期化中に作成されたオブジェクトは、スレッド間で共有されます。

拡張機能を持つ Lambda 関数の場合、実行環境はシャットダウン中に SIGTERM シグナルを出力します。このシグナルは拡張機能によって使用され、バッファのフラッシュなどのクリーンアップタスクをトリガーします。SIGTERM イベントにサブスクライブすることで、データベース接続を閉じるなどの関数のクリーンアップタスクを実行できます。実行環境のライフサイクルの詳細については、「Lambda 実行環境のライフサイクルの概要」を参照してください。

依存関係バージョン

Lambda マネージドインスタンスには、以下のパッケージバージョンの最小要件が必要です。

Powertools for AWS Lambda (Java)

Powertools for AWS Lambda (Java) は Lambda マネージドインスタンスと互換性があり、ログ記録、トレース、メトリクスなどのユーティリティを利用できます。詳細については、「Powertools for AWS Lambda (Java)」を参照してください。

次のステップ