Programación asincrónica mediante AWS SDK for Java 2.x

El AWS SDK for Java 2.x incorpora clientes asincrónicos con compatibilidad de E/S no bloqueante que implementan un alto nivel de simultaneidad en algunos subprocesos. Sin embargo, no se garantiza una E/S no bloqueante total. Un cliente asincrónico puede realizar llamadas de bloqueo en algunos casos, como recuperación de credenciales, firma de solicitudes mediante AWS Signature Version 4 (SigV4) o detección de puntos de conexión.

Los métodos síncronos bloquean la ejecución de los subprocesos hasta que el cliente recibe una respuesta del servicio. Los métodos asíncronos terminan de ejecutarse inmediatamente, devolviendo el control al subproceso que realiza la llamada sin esperar una respuesta.

Como un método asíncrono termina de ejecutarse antes de que haya una respuesta disponible, necesita una forma de obtener la respuesta cuando esté lista. Los métodos para cliente asíncrono en 2.x del AWS SDK para Java devuelven objetos CompletableFuture que le permiten acceder a la respuesta cuando esté lista.

Uso de API de cliente asincrónicas

Las firmas de los métodos de cliente asincrónicos son las mismas que las de los métodos sincrónicos, pero los métodos asincrónicos devuelven un objeto CompletableFuture que contiene los resultados de la operación asincrónica en el futuro. Si se produce un error mientras se ejecuta el método asincrónico del SDK, el error se lanza como CompletionException.

Un método que se puede usar para obtener el resultado es encadenar un método whenComplete() al CompletableFuture devuelto por la llamada a métodos del SDK. El método whenComplete() recibe el resultado o un objeto del tipo Throwable de tipo CompletionException en función de cómo se haya completado la llamada asincrónica. Debe proporcionar una acción a whenComplete() para procesar o comprobar los resultados antes de que se devuelva al código de llamada.

Si quiere devolver algo distinto del objeto devuelto por el método del SDK, use el método handle() en su lugar. El método handle() utiliza los mismos parámetros que whenComplete(), pero se puede procesar el resultado y devolver un objeto.

Para esperar a que se complete la cadena asincrónica y recuperar los resultados de la finalización, puede llamar al método join(). Si el objeto Throwable no se administraba en la cadena, el método join() lanza un CompletionException no controlado que empaqueta la excepción original. Acceda a la excepción original con CompletionException#getCause(). También puede llamar al método CompletableFuture#get() para obtener los resultados de la finalización. Sin embargo, el método get() puede lanzar excepciones controladas.

El siguiente ejemplo muestra dos variantes de cómo se puede trabajar con el método listTables() del cliente asincrónico de DynamoDB. La acción pasada a whenComplete() registra simplemente una respuesta correcta, mientras que la versión handle() extrae la lista de nombres de tablas y la devuelve. En ambos casos, si se genera un error en la cadena asincrónica, el error se vuelve a lanzar para que el código del cliente pueda administrarlo.

Importaciones

import software.amazon.awssdk.regions.Region;
import software.amazon.awssdk.services.dynamodb.DynamoDbAsyncClient;
import software.amazon.awssdk.services.dynamodb.model.ListTablesRequest;
import software.amazon.awssdk.services.dynamodb.model.ListTablesResponse;

import java.util.List;
import java.util.concurrent.CompletableFuture;

Código de

whenComplete() variation

public class DynamoDbAsyncListTables {

    public static void main(String[] args) {
        Region region = Region.US_EAST_1;
        DynamoDbAsyncClient dynamoDbAsyncClient = DynamoDbAsyncClient.builder().region(region).build();
        try {
            ListTablesResponse listTablesResponse = listTablesWhenComplete(dynamoDbAsyncClient).join();  // The join() method may throw a CompletionException.
            if (listTablesResponse.hasTableNames()){
                System.out.println("Table exist in this region: " + region.id());
            }
        } catch (RuntimeException e) {
            // Handle as needed. Here we simply print out the class names.
            System.out.println(e.getClass()); // Prints 'class java.util.concurrent.CompletionException'.
            System.out.println(e.getCause().getClass()); // Prints 'class software.amazon.awssdk.services.dynamodb.model.DynamoDbException'.
        }
    }

    public static CompletableFuture<ListTablesResponse> listTablesWhenComplete(DynamoDbAsyncClient client) {
        return client.listTables(ListTablesRequest.builder().build())
            .whenComplete((listTablesResponse, throwable) -> {
                if (listTablesResponse != null) {  // Consume the response.
                    System.out.println("The SDK's listTables method completed successfully.");
                } else {
                    RuntimeException cause = (RuntimeException) throwable.getCause(); // If an error was thrown during the SDK's listTables method it is wrapped in a CompletionException.
                                                                                      // The SDK throws only RuntimeExceptions, so this is a safe cast.
                    System.out.println(cause.getMessage());  // Log error here, but rethrow so the calling code can handle as needed.
                    throw cause;
                }
            });
    }

handle() variation

public class DynamoDbAsyncListTables {

    public static void main(String[] args) {
        Region region = Region.US_EAST_1;
        DynamoDbAsyncClient dynamoDbAsyncClient = DynamoDbAsyncClient.builder().region(region).build();
        try {
            List<String> tableNames = listTablesHandle(dynamoDbAsyncClient).join(); // The join() method may throw a CompletionException.
            tableNames.forEach(System.out::println);
        } catch (RuntimeException e) {
            // Handle as needed. Here we simply print out the class names.
            System.out.println(e.getClass()); // Prints 'class java.util.concurrent.CompletionException'.
            System.out.println(e.getCause().getClass()); // Prints 'class software.amazon.awssdk.services.dynamodb.model.DynamoDbException'.
        }
    }

    public static CompletableFuture<List<String>> listTablesHandle(DynamoDbAsyncClient client) {
        return client.listTables(ListTablesRequest.builder().build())
            .handle((listTablesResponse, throwable) -> {
                if (listTablesResponse != null) {
                    return listTablesResponse.tableNames(); // Return the list of table names.
                } else {
                    RuntimeException cause = (RuntimeException) throwable.getCause(); // If an error was thrown during the SDK's listTables method it is wrapped in a CompletionException.
                                                                                      // The SDK throws only RuntimeExceptions, so this is a safe cast.
                    System.out.println(cause.getMessage());  // Log error here, but rethrow so the calling code can handle as needed.
                    throw cause;
                }
            });
    }
}

Administración de streaming en métodos asincrónicos

Para métodos asincrónicos con contenido de streaming, debe proporcionar un objeto AsyncRequestBody para proporcionar el contenido de forma incremental o un objeto AsyncResponseTransformer para recibir y procesar la respuesta.

El ejemplo siguiente carga un archivo en Amazon S3 de forma asincrónica utilizando la forma asincrónica de la operación PutObject.

Importaciones

import software.amazon.awssdk.core.async.AsyncRequestBody;
import software.amazon.awssdk.regions.Region;
import software.amazon.awssdk.services.s3.S3AsyncClient;
import software.amazon.awssdk.services.s3.model.PutObjectRequest;
import software.amazon.awssdk.services.s3.model.PutObjectResponse;
import java.nio.file.Paths;
import java.util.concurrent.CompletableFuture;

Código de

/**
 * To run this AWS code example, ensure that you have setup your development environment, including your AWS credentials.
 *
 * For information, see this documentation topic:
 *
 * https://docs.aws.amazon.com/sdk-for-java/latest/developer-guide/get-started.html
 */

public class S3AsyncOps {

     public static void main(String[] args) {

         final String USAGE = "\n" +
                 "Usage:\n" +
                 "    S3AsyncOps <bucketName> <key> <path>\n\n" +
                 "Where:\n" +
                 "    bucketName - the name of the Amazon S3 bucket (for example, bucket1). \n\n" +
                 "    key - the name of the object (for example, book.pdf). \n" +
                 "    path - the local path to the file (for example, C:/AWS/book.pdf). \n" ;

        if (args.length != 3) {
            System.out.println(USAGE);
             System.exit(1);
        }

        String bucketName = args[0];
        String key = args[1];
        String path = args[2];

        Region region = Region.US_WEST_2;
        S3AsyncClient client = S3AsyncClient.builder()
                .region(region)
                .build();

        PutObjectRequest objectRequest = PutObjectRequest.builder()
                .bucket(bucketName)
                .key(key)
                .build();

        // Put the object into the bucket
        CompletableFuture<PutObjectResponse> future = client.putObject(objectRequest,
                AsyncRequestBody.fromFile(Paths.get(path))
        );
        future.whenComplete((resp, err) -> {
            try {
                if (resp != null) {
                    System.out.println("Object uploaded. Details: " + resp);
                } else {
                    // Handle error
                    err.printStackTrace();
                }
            } finally {
                // Only close the client when you are completely done with it
                client.close();
            }
        });

        future.join();
    }
}

En el ejemplo siguiente se obtiene un archivo de Amazon S3 utilizando la forma asincrónica de la operación GetObject.

Importaciones

import software.amazon.awssdk.core.async.AsyncResponseTransformer;
import software.amazon.awssdk.regions.Region;
import software.amazon.awssdk.services.s3.S3AsyncClient;
import software.amazon.awssdk.services.s3.model.GetObjectRequest;
import software.amazon.awssdk.services.s3.model.GetObjectResponse;
import java.nio.file.Paths;
import java.util.concurrent.CompletableFuture;

Código de

/**
 * To run this AWS code example, ensure that you have setup your development environment, including your AWS credentials.
 *
 * For information, see this documentation topic:
 *
 * https://docs.aws.amazon.com/sdk-for-java/latest/developer-guide/get-started.html
 */

public class S3AsyncStreamOps {

    public static void main(String[] args) {

        final String USAGE = "\n" +
                "Usage:\n" +
                "    S3AsyncStreamOps <bucketName> <objectKey> <path>\n\n" +
                "Where:\n" +
                "    bucketName - the name of the Amazon S3 bucket (for example, bucket1). \n\n" +
                "    objectKey - the name of the object (for example, book.pdf). \n" +
                "    path - the local path to the file (for example, C:/AWS/book.pdf). \n" ;

        if (args.length != 3) {
            System.out.println(USAGE);
            System.exit(1);
         }

        String bucketName = args[0];
        String objectKey = args[1];
        String path = args[2];

        Region region = Region.US_WEST_2;
        S3AsyncClient client = S3AsyncClient.builder()
                .region(region)
                .build();

        GetObjectRequest objectRequest = GetObjectRequest.builder()
                .bucket(bucketName)
                .key(objectKey)
                .build();

        CompletableFuture<GetObjectResponse> futureGet = client.getObject(objectRequest,
                AsyncResponseTransformer.toFile(Paths.get(path)));

        futureGet.whenComplete((resp, err) -> {
            try {
                if (resp != null) {
                    System.out.println("Object downloaded. Details: "+resp);
                } else {
                    err.printStackTrace();
                }
            } finally {
               // Only close the client when you are completely done with it
                client.close();
            }
        });
        futureGet.join();
    }
}

Configuración de opciones asincrónicas avanzadas

El AWS SDK para Java 2.x usa Netty, una plataforma de aplicación de red asíncrona y basado en eventos, para gestionar subprocesos de E/S. El AWS SDK para Java 2.x crea un ExecutorService después de Netty para completar los futuros devueltos desde la solicitud del cliente HTTP al cliente Netty. Esta abstracción reduce el riesgo de que una aplicación interrumpa el proceso de sincronización si los desarrolladores deciden detener o suspender los subprocesos. De forma predeterminada, cada cliente asíncrono crea un grupo de subprocesos en función del número de procesadores y gestiona las tareas de una cola dentro del ExecutorService.

Puede especificar una implementación de JDK específica de ExecutorService cuando cree su cliente asincrónico. El siguiente fragmento crea un ExecutorService con un número fijo de subprocesos.

Código de

S3AsyncClient clientThread = S3AsyncClient.builder()
  .asyncConfiguration(
    b -> b.advancedOption(SdkAdvancedAsyncClientOption
      .FUTURE_COMPLETION_EXECUTOR,
      Executors.newFixedThreadPool(10)
    )
  )
  .build();

Para optimizar el rendimiento, puede administrar su propio ejecutor de grupo de subprocesos e incluirlo al configurar el cliente.

ThreadPoolExecutor executor = new ThreadPoolExecutor(50, 50,
    10, TimeUnit.SECONDS,
    new LinkedBlockingQueue<>(<custom_value>),
    new ThreadFactoryBuilder()
      .threadNamePrefix("sdk-async-response").build());

// Allow idle core threads to time out
executor.allowCoreThreadTimeOut(true);

S3AsyncClient clientThread = S3AsyncClient.builder()
  .asyncConfiguration(
    b -> b.advancedOption(SdkAdvancedAsyncClientOption
      .FUTURE_COMPLETION_EXECUTOR,
      executor
    )
  )
  .build();