多Spring Boot实例轮询S3对象时如何避免重复读取文件？

方案一：利用S3原子移动操作（推荐） #

S3的对象移动本质是原子复制+删除操作，通过将待处理对象从轮询目录移动到专属处理目录，可实现独占性——只有成功完成移动的实例能获得处理权限，其他实例会因原对象已不存在而无法操作。

步骤：

1. 为S3桶划分两个目录：/pending/（存放待处理文件）和/processing/{instance-id}/（每个实例的专属处理目录，用实例唯一标识命名）
2. 实例轮询/pending/下的对象，对每个对象尝试执行原子移动：
   • 复制原对象到专属处理目录
   • 复制成功后删除原对象，后续实例将无法再获取该对象

代码示例（AWS SDK 2.x）：

@Autowired
private S3Client s3Client;
@Value("${spring.application.instance-id}")
private String instanceId;

private boolean tryClaimObject(String bucketName, String objectKey) {
    String targetKey = String.format("processing/%s/%s", instanceId, objectKey);
    try {
        // 原子复制原对象到处理目录
        CopyObjectRequest copyRequest = CopyObjectRequest.builder()
                .sourceBucket(bucketName)
                .sourceKey(objectKey)
                .destinationBucket(bucketName)
                .destinationKey(targetKey)
                .build();
        s3Client.copyObject(copyRequest);
        
        // 复制成功后删除原对象
        s3Client.deleteObject(DeleteObjectRequest.builder()
                .bucket(bucketName)
                .key(objectKey)
                .build());
        return true;
    } catch (S3Exception e) {
        // 复制失败：原对象已被其他实例移走
        return false;
    }
}

// 轮询处理逻辑
public void pollS3Objects() {
    ListObjectsV2Request listRequest = ListObjectsV2Request.builder()
            .bucket("your-bucket-name")
            .prefix("pending/")
            .build();
    ListObjectsV2Response response = s3Client.listObjectsV2(listRequest);
    
    for (S3Object object : response.contents()) {
        if (tryClaimObject("your-bucket-name", object.key())) {
            // 处理已获取的对象
            processObject("your-bucket-name", String.format("processing/%s/%s", instanceId, object.key()));
        }
    }
}

方案二：利用S3条件写入实现锁机制 #

为每个待处理对象创建对应锁文件，通过S3的条件写入特性，确保只有第一个创建锁文件的实例能获得处理权限。

步骤：

1. 实例轮询/pending/下的对象
2. 尝试创建/locks/{object-key}.lock文件，设置条件：仅当锁文件不存在时才写入
3. 创建成功的实例处理原对象，完成后删除原对象和锁文件
4. 创建失败的实例跳过该对象

代码示例：

private boolean tryAcquireLock(String bucketName, String objectKey) {
    String lockKey = String.format("locks/%s.lock", objectKey);
    try {
        PutObjectRequest putRequest = PutObjectRequest.builder()
                .bucket(bucketName)
                .key(lockKey)
                .contentType("text/plain")
                .metadata(Map.of("owner", instanceId))
                .build();
        // 添加条件：仅当锁文件不存在时才写入
        putRequest = putRequest.toBuilder()
                .overrideConfiguration(b -> b.putHeader("x-amz-if-none-match", "*"))
                .build();
        s3Client.putObject(putRequest, RequestBody.fromString("locked"));
        return true;
    } catch (S3Exception e) {
        // 锁文件已存在，获取锁失败
        return false;
    }
}

public void pollAndProcess() {
    ListObjectsV2Request listRequest = ListObjectsV2Request.builder()
            .bucket("your-bucket-name")
            .prefix("pending/")
            .build();
    ListObjectsV2Response response = s3Client.listObjectsV2(listRequest);
    
    for (S3Object object : response.contents()) {
        String objectKey = object.key();
        if (tryAcquireLock("your-bucket-name", objectKey)) {
            try {
                // 处理对象
                processObject("your-bucket-name", objectKey);
                // 清理资源
                s3Client.deleteObject(b -> b.bucket("your-bucket-name").key(objectKey));
                s3Client.deleteObject(b -> b.bucket("your-bucket-name").key(String.format("locks/%s.lock", objectKey)));
            } catch (Exception e) {
                // 处理失败，释放锁
                s3Client.deleteObject(b -> b.bucket("your-bucket-name").key(String.format("locks/%s.lock", objectKey)));
            }
        }
    }
}

方案三：结合分布式锁（如Redis） #

如果系统已存在Redis等分布式锁组件，可直接复用实现独占逻辑：

• 实例轮询到S3对象时，尝试获取以对象Key为标识的Redis分布式锁
• 获取锁成功的实例处理对象，完成后释放锁并删除S3对象
• 获取锁失败的实例跳过该对象

这种方式支持设置锁超时时间，能避免因实例崩溃导致的锁永久占用问题。

关键注意事项 #

• 处理实例崩溃场景：可通过S3生命周期规则，定期将/processing/目录下超时的对象移回/pending/；锁文件也可设置自动过期策略
• 轮询频率：避免过于频繁的S3列表请求，可通过延迟或定时任务控制轮询间隔

内容的提问来源于stack exchange，提问作者Gopal