aws-sdk-java-v2-dynamodb

AWS SDK for Java 2.x - Amazon DynamoDB

概述

使用 AWS SDK for Java 2.x 和 Enhanced Client 提供 DynamoDB 模式，支持类型安全的 CRUD、查询、批量操作、事务和 Spring Boot 集成。

适用场景

• 对 DynamoDB 项目执行 CRUD 操作
• 使用排序键或 GSI 查询表
• 对多个项目执行批量操作
• 跨表执行原子事务
• 将 DynamoDB 与 Spring Boot 集成

说明

1. 将 AWS SDK DynamoDB 依赖项添加到 pom.xml
2. 配置客户端设置（低级客户端或 Enhanced Client）
3. 使用 @DynamoDbBean 注解定义实体类
4. 使用 DynamoDbTable 执行操作（CRUD、查询、扫描、批量操作、事务）
5. 使用重试逻辑和指数退避处理部分失败
6. 在 Spring Boot 集成中使用仓库模式

依赖项

添加到 pom.xml：

<dependency>
    <groupId>software.amazon.awssdk</groupId>
    <artifactId>dynamodb</artifactId>
</dependency>


<dependency>
    <groupId>software.amazon.awssdk</groupId>
    <artifactId>dynamodb-enhanced</artifactId>
</dependency>

客户端设置

低级客户端

import software.amazon.awssdk.regions.Region;
import software.amazon.awssdk.services.dynamodb.DynamoDbClient;

DynamoDbClient dynamoDb = DynamoDbClient.builder()
    .region(Region.US_EAST_1)
    .build();

Enhanced Client（推荐）

import software.amazon.awssdk.enhanced.dynamodb.DynamoDbEnhancedClient;

DynamoDbEnhancedClient enhancedClient = DynamoDbEnhancedClient.builder()
    .dynamoDbClient(dynamoDb)
    .build();

实体映射

@DynamoDbBean
public class Customer {

    @DynamoDbPartitionKey
    private String customerId;

    @DynamoDbAttribute("customer_name")
    private String name;

    private String email;

    @DynamoDbSortKey
    private String orderId;

    // Getters and setters
}

如需了解使用 GSI 和自定义转换器进行复杂实体映射，请参阅实体映射参考。

CRUD 操作

基本操作

// Create or update item
DynamoDbTable<Customer> table = enhancedClient.table("Customers", TableSchema.fromBean(Customer.class));
table.putItem(customer);

// Get item
Customer result = table.getItem(Key.builder().partitionValue(customerId).build());

// Update item
return table.updateItem(customer);

// Delete item
table.deleteItem(Key.builder().partitionValue(customerId).build());

复合键操作

// Get item with composite key
Order order = table.getItem(Key.builder()
    .partitionValue(customerId)
    .sortValue(orderId)
    .build());

查询操作

基本查询

import software.amazon.awssdk.enhanced.dynamodb.model.QueryConditional;

QueryConditional queryConditional = QueryConditional
    .keyEqualTo(Key.builder()
        .partitionValue(customerId)
        .build());

List<Order> orders = table.query(queryConditional).items().stream()
    .collect(Collectors.toList());

使用筛选条件的高级查询

import software.amazon.awssdk.enhanced.dynamodb.Expression;

Expression filter = Expression.builder()
    .expression("status = :pending")
    .putExpressionValue(":pending", AttributeValue.builder().s("PENDING").build())
    .build();

List<Order> pendingOrders = table.query(r -> r
    .queryConditional(queryConditional)
    .filterExpression(filter))
    .items().stream()
    .collect(Collectors.toList());

如需了解详细的查询模式，请参阅高级操作参考。

扫描操作

警告：扫描会读取整个表，并消耗所有项目的读取容量。请尽可能优先使用带分区键或 GSI 的查询操作。

扫描前的验证：

• 确认对于当前访问模式，无法使用分区键进行查询
• 确认表有足够的预置读取容量，或使用按需模式
• 考虑使用 limit() 分页，以控制容量消耗

// Scan all items
List<Customer> allCustomers = table.scan().items().stream()
    .collect(Collectors.toList());

// Scan with filter
Expression filter = Expression.builder()
    .expression("points >= :minPoints")
    .putExpressionValue(":minPoints", AttributeValue.builder().n("1000").build())
    .build();

List<Customer> vipCustomers = table.scan(r -> r.filterExpression(filter))
    .items().stream()
    .collect(Collectors.toList());

批量操作

批量读取

import software.amazon.awssdk.enhanced.dynamodb.model.*;

List<Key> keys = customerIds.stream()
    .map(id -> Key.builder().partitionValue(id).build())
    .collect(Collectors.toList());

ReadBatch.Builder<Customer> batchBuilder = ReadBatch.builder(Customer.class)
    .mappedTableResource(table);

keys.forEach(batchBuilder::addGetItem);

BatchGetResultPageIterable result = enhancedClient.batchGetItem(r ->
    r.addReadBatch(batchBuilder.build()));

List<Customer> customers = result.resultsForTable(table).stream()
    .collect(Collectors.toList());

带错误处理的批量写入

WriteBatch.Builder<Customer> batchBuilder = WriteBatch.builder(Customer.class)
    .mappedTableResource(table);

customers.forEach(batchBuilder::addPutItem);

BatchWriteItemEnhancedRequest request = BatchWriteItemEnhancedRequest.builder()
    .addWriteBatch(batchBuilder.build())
    .build();

BatchWriteResult result = enhancedClient.batchWriteItem(request);

// Validate: check for unprocessed items
if (!result.writeResponsesForTable(table).isEmpty()) {
    // Retry unprocessed items with exponential backoff
    Map<String, AttributeValue> unprocessed = result.writeResponsesForTable(table).get(0)
        .unprocessedAttributes();
    if (unprocessed != null && !unprocessed.isEmpty()) {
        enhancedClient.batchWriteItem(r -> r
            .addWriteBatch(WriteBatch.builder(Customer.class)
                .mappedTableResource(table)
                .addPutItemFromItem(unprocessed)
                .build()));
    }
}

事务

带重试的事务写入

public void placeOrderWithRetry(Order order, Customer customer, int maxRetries) {
    int attempt = 0;
    while (attempt < maxRetries) {
        try {
            enhancedClient.transactWriteItems(r -> r
                .addPutItem(customerTable, customer)
                .addPutItem(orderTable, order));
            return;
        } catch (TransactionCanceledException e) {
            if (e.cancellationReasons().stream()
                .anyMatch(r -> r.code().equals("TransactionCanceledException")
                    && r.message().contains("throughput"))) {
                attempt++;
                if (attempt < maxRetries) {
                    try { Thread.sleep((long) Math.pow(2, attempt) * 100); }
                    catch (InterruptedException ie) { Thread.currentThread().interrupt(); }
                }
            } else {
                throw e; // Non-retryable error
            }
        }
    }
}

事务读取

TransactGetItemsEnhancedRequest request = TransactGetItemsEnhancedRequest.builder()
    .addGetItem(customerTable, customerKey)
    .addGetItem(orderTable, orderKey)
    .build();

List<Document> results = enhancedClient.transactGetItems(request);

Spring Boot 集成

配置

@Configuration
public class DynamoDbConfiguration {

    @Bean
    public DynamoDbClient dynamoDbClient() {
        return DynamoDbClient.builder()
            .region(Region.US_EAST_1)
            .build();
    }

    @Bean
    public DynamoDbEnhancedClient dynamoDbEnhancedClient(DynamoDbClient dynamoDbClient) {
        return DynamoDbEnhancedClient.builder()
            .dynamoDbClient(dynamoDbClient)
            .build();
    }
}

仓库模式

@Repository
public class CustomerRepository {

    private final DynamoDbTable<Customer> customerTable;

    public CustomerRepository(DynamoDbEnhancedClient enhancedClient) {
        this.customerTable = enhancedClient.table("Customers", TableSchema.fromBean(Customer.class));
    }

    public void save(Customer customer) {
        customerTable.putItem(customer);
    }

    public Optional<Customer> findById(String customerId) {
        Key key = Key.builder().partitionValue(customerId).build();
        return Optional.ofNullable(customerTable.getItem(key));
    }
}

如需了解全面的 Spring Boot 集成模式，请参阅 Spring Boot 集成参考。

测试

使用模拟对象进行单元测试

@ExtendWith(MockitoExtension.class)
class CustomerServiceTest {

    @Mock
    private DynamoDbClient dynamoDbClient;

    @Mock
    private DynamoDbEnhancedClient enhancedClient;

    @Mock
    private DynamoDbTable<Customer> customerTable;

    @InjectMocks
    private CustomerService customerService;

    @Test
    void saveCustomer_ShouldReturnSavedCustomer() {
        // Arrange
        when(enhancedClient.table(anyString(), any(TableSchema.class)))
            .thenReturn(customerTable);

        Customer customer = new Customer("123", "John Doe", "[email protected]");

        // Act
        Customer result = customerService.saveCustomer(customer);

        // Assert
        assertNotNull(result);
        verify(customerTable).putItem(customer);
    }
}

使用 LocalStack 进行集成测试

@Testcontainers
@SpringBootTest
class DynamoDbIntegrationTest {

    @Container
    static LocalStackContainer localstack = new LocalStackContainer(
        DockerImageName.parse("localstack/localstack:3.0"))
        .withServices(LocalStackContainer.Service.DYNAMODB);

    @DynamicPropertySource
    static void configureProperties(DynamicPropertyRegistry registry) {
        registry.add("aws.endpoint",
            () -> localstack.getEndpointOverride(LocalStackContainer.Service.DYNAMODB).toString());
    }

    @Autowired
    private DynamoDbEnhancedClient enhancedClient;

    @Test
    void testCustomerCRUDOperations() {
        // Test implementation
    }
}

如需了解详细的测试策略，请参阅测试策略。

最佳实践

• 使用 Enhanced Client：以更少的样板代码执行类型安全的操作
• 设计分区键以实现均匀分布：避免热点分区
• 优先使用查询而非扫描：使用 GSI 满足访问模式
• 按 25/100 个项目分批处理：BatchGetItem 每次最多处理 100 个项目，BatchWriteItem 每张表最多处理 25 个项目
• 处理部分失败：针对 ProvisionedThroughputExceeded 实现带指数退避的重试
• 使用条件写入：通过 attribute_not_exists(pk) 防止竞态条件

示例

完整的 CRUD 仓库

@Repository
public class UserRepository {

    private final DynamoDbTable<User> userTable;

    public UserRepository(DynamoDbEnhancedClient enhancedClient) {
        this.userTable = enhancedClient.table("Users", TableSchema.fromBean(User.class));
    }

    public User save(User user) {
        userTable.putItem(user);
        return user;
    }

    public Optional<User> findById(String userId) {
        Key key = Key.builder().partitionValue(userId).build();
        return Optional.ofNullable(userTable.getItem(key));
    }

    public void deleteById(String userId) {
        userTable.deleteItem(Key.builder().partitionValue(userId).build());
    }
}

带重试的条件写入

public boolean createIfNotExists(User user) {
    PutItemEnhancedRequest<User> request = PutItemEnhancedRequest.builder(User.class)
        .item(user)
        .conditionExpression("attribute_not_exists(userId)")
        .build();

    try {
        userTable.putItemWithRequest(request);
        return true;
    } catch (ConditionalCheckFailedException e) {
        return false; // Item already exists
    }
}

限制和警告

• 项目大小限制：DynamoDB 项目大小上限为 400KB
• 分区键设计：设计不当会导致热点分区
• 批量限制：BatchGetItem 最多处理 100 个项目，BatchWriteItem 每张表最多处理 25 个项目
• 事务成本：事务消耗的读写容量单位是普通操作的 2 倍
• 扫描操作：扫描会消耗大量读取容量；仅在必要时使用

参考资料

• AWS DynamoDB 文档
• AWS SDK for Java 文档
• DynamoDB 示例
• 用于测试的 LocalStack

如需了解详细实现，请参阅 references 文件夹：

• 实体映射参考
• 高级操作参考
• Spring Boot 集成参考
• 测试策略