如何用Java实现类CDC模式读取S3文件并感知新增文件？

针对你的需求，不用轮询S3接口、不依赖第三方中间件（如Kafka）的可行方案是利用AWS S3事件通知+SQS队列——S3主动推送新增文件事件到SQS，Java应用监听SQS即可高效获取新增文件信息并读取，完全符合你“类CDC”的摄入要求，具体实现步骤如下：

1. 配置S3桶事件通知 #

• 登录AWS控制台，找到目标S3桶，进入「属性」→「事件通知」
• 创建新通知规则：
  • 事件类型：勾选「所有对象创建事件」（对应s3:ObjectCreated:*，覆盖Put、Post、Copy等所有新增文件操作）
  • 目标选择：指定「SQS队列」，可新建队列或选择已有队列
  • 权限配置：在SQS的访问策略中添加S3服务主体的sqs:SendMessage权限，确保S3能向队列推送消息

2. Java应用监听SQS并处理S3文件 #

使用AWS SDK for Java v2（性能更优），先引入依赖：

<dependency>
    <groupId>software.amazon.awssdk</groupId>
    <artifactId>sqs</artifactId>
    <version>2.25.0</version>
</dependency>
<dependency>
    <groupId>software.amazon.awssdk</groupId>
    <artifactId>s3</artifactId>
    <version>2.25.0</version>
</dependency>
<dependency>
    <groupId>com.fasterxml.jackson.core</groupId>
    <artifactId>jackson-databind</artifactId>
    <version>2.15.2</version>
</dependency>

编写核心监听与处理逻辑（采用SQS长轮询，避免空轮询浪费资源）：

import software.amazon.awssdk.services.sqs.SqsClient;
import software.amazon.awssdk.services.sqs.model.*;
import software.amazon.awssdk.services.s3.S3Client;
import software.amazon.awssdk.services.s3.model.GetObjectRequest;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import java.io.InputStream;

public class S3NewFileConsumer {
    private static final String SQS_QUEUE_URL = "你的SQS队列URL";
    private static final SqsClient sqsClient = SqsClient.create();
    private static final S3Client s3Client = S3Client.create();
    private static final ObjectMapper objectMapper = new ObjectMapper();

    public static void main(String[] args) {
        while (true) {
            // 长轮询等待消息，最长阻塞20秒，无消息时不产生无效请求
            ReceiveMessageRequest receiveReq = ReceiveMessageRequest.builder()
                    .queueUrl(SQS_QUEUE_URL)
                    .waitTimeSeconds(20)
                    .maxNumberOfMessages(10)
                    .build();

            ReceiveMessageResponse response = sqsClient.receiveMessage(receiveReq);
            for (Message msg : response.messages()) {
                try {
                    // 解析S3事件通知的JSON消息体
                    JsonNode rootNode = objectMapper.readTree(msg.body());
                    JsonNode record = rootNode.get("records").get(0);
                    String bucketName = record.get("s3").get("bucket").get("name").asText();
                    String objectKey = record.get("s3").get("object").get("key").asText();

                    // 读取S3文件内容并处理
                    try (InputStream fileStream = s3Client.getObject(GetObjectRequest.builder()
                            .bucket(bucketName)
                            .key(objectKey)
                            .build())) {
                        processFileContent(fileStream, bucketName, objectKey);
                    }

                    // 处理完成后删除消息，避免重复消费
                    sqsClient.deleteMessage(DeleteMessageRequest.builder()
                            .queueUrl(SQS_QUEUE_URL)
                            .receiptHandle(msg.receiptHandle())
                            .build());
                } catch (Exception e) {
                    // 异常处理：日志记录、重试或死信队列转发
                    System.err.printf("处理消息失败：%s%n", e.getMessage());
                }
            }
        }
    }

    /**
     * 自定义文件内容处理逻辑
     */
    private static void processFileContent(InputStream fileStream, String bucketName, String objectKey) {
        // 示例：写入数据库、解析业务数据等
        System.out.printf("开始处理文件：s3://%s/%s%n", bucketName, objectKey);
    }
}

3. 关键权限说明 #

• Java应用使用的IAM身份（用户/角色）需具备：
  • sqs:ReceiveMessage、sqs:DeleteMessage（目标SQS队列权限）
  • s3:GetObject（目标S3桶权限）
• SQS队列的访问策略需允许S3推送消息，示例策略：

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Principal": {"Service": "s3.amazonaws.com"},
            "Action": "sqs:SendMessage",
            "Resource": "你的SQS队列ARN",
            "Condition": {
                "ArnLike": {"aws:SourceArn": "arn:aws:s3:::你的S3桶名"}
            }
        }
    ]
}

方案优势 #

• 完全避免轮询S3的低效操作，S3主动推送事件，实时性强
• 无需部署维护第三方中间件，仅依赖AWS原生服务与Java SDK
• SQS的长轮询机制大幅减少无效请求，降低资源消耗与成本

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