带有AWS DynamoDB的反应式Spring Webflux

AWS已经发布了Java版本2的AWS开发工具包，该开发工具包现在支持针对不同AWS服务的API调用的非阻塞IO。在本文中，我将探讨如何使用AWS开发工具包2.x的DynamoDB API以及如何使用Spring Webflux堆栈公开响应式端点-这种方式，应用程序是端对端响应式的，大概应该非常有效地使用资源（我有计划在此设置上做一些测试作为后续步骤）。

申请详情

简单地看一下代码并在那儿跟随它可能会更容易-在我的GitHub存储库中可以找到它。

该应用程序很简单-使用以下Kotlin代码对酒店实体执行CRUD操作：

data class Hotel(val id: String = UUID.randomUUID().toString(),val name: String? = null,val address: String? = null,val state: String? = null,val zip: String? = null
)

我想公开端点以保存和检索酒店实体，并按州获取酒店列表。

AWS开发工具包2的详细信息

AWS SDK 2 api的所有软件包名称现在都以“ software.amazon.awssdk”前缀开头，使用以下代码创建与DynamoDB进行交互的客户端：

import software.amazon.awssdk.auth.credentials.DefaultCredentialsProvider
import software.amazon.awssdk.regions.Region
import software.amazon.awssdk.services.dynamodb.DynamoDbAsyncClientval client: DynamoDbAsyncClient = DynamoDbAsyncClient.builder().region(Region.of(dynamoProperties.region)).credentialsProvider(DefaultCredentialsProvider.builder().build()).build()

创建DynamoDbAsyncClient实例后，使用此客户端的任何操作都将返回Java 8 CompletableFuture类型。例如。保存酒店实体时：

val putItemRequest = PutItemRequest.builder().tableName("hotels").item(HotelMapper.toMap(hotel)).build()val result: CompletableFuture<PutItemResponse> =dynamoClient.putItem(putItemRequest)

并通过ID检索记录：

val getItemRequest: GetItemRequest = GetItemRequest.builder().key(mapOf(Constants.ID to AttributeValue.builder().s(id).build())).tableName(Constants.TABLE_NAME).build()val response: CompletableFuture<GetItemResponse> = dynamoClient.getItem(getItemRequest)

CompletableFuture提供了一组全面的功能，以在可用时转换结果。

与Spring Webflux集成

Spring Webflux是一个反应式Web框架。现在，借助AWS开发工具包2中的非阻塞IO支持，可以使用DynamoDB编写端到端的响应式和非阻塞应用程序。 Spring Webflux使用反应堆核心提供反应流支持，与AWS开发工具包2集成的技巧是将Java 8 CompletableFuture转换为反应堆核心类型，方法是通过ID从DynamoDB检索项目时采用以下方式：

val getItemRequest: GetItemRequest = GetItemRequest.builder().key(mapOf(Constants.ID to AttributeValue.builder().s(id).build())).tableName(Constants.TABLE_NAME).build()return Mono.fromCompletionStage(dynamoClient.getItem(getItemRequest)).map { resp ->HotelMapper.fromMap(id, resp.item())}

Spring Webflux期望不同Web终结点方法签名的返回类型为响应类型，因此，用于说出酒店列表的典型终结点如下：

@RequestMapping(value = ["/hotels"], method = [RequestMethod.GET])
fun getHotelsByState(@RequestParam("state") state: String): Flux<Hotel> {return hotelRepo.findHotelsByState(state)
}

Spring Webflux还支持描述应用程序API的功能性方法，因此，一个等效的API可通过其ID检索酒店，但表示为功能性DSL，如下所示：

@Configuration
class HotelAdditionalRoutes {@Beanfun routes(hotelRepo: HotelRepo) = router {GET("/hotels/{id}") { req ->val id = req.pathVariable("id")val response: Mono<ServerResponse> = hotelRepo.getHotel(id).flatMap { hotel ->ServerResponse.ok().body(BodyInserters.fromObject(hotel))}response.switchIfEmpty(ServerResponse.notFound().build())}}
}

结论

AWS SDK 2使编写端到端反应性和非阻塞应用程序变得简单。我已使用Spring Webflux和AWS开发工具包2发电机客户端在此处编写此类应用程序。整个工作示例可在我的GitHub存储库中找到-https ://github.com/bijukunjummen/boot-with-dynamodb，其中包含有关如何启动DynamoDB本地版本并将其用于测试应用程序的说明。