首页 > 解决方案 > 同步 S3 存储桶并监听变化

问题描述

我有一个 AWS S3 存储桶,我每周都会在其中放置一个新的 ZIP 文件。

我想在我现有的 Web 服务中添加一个功能,用 Spring Boot 编写:在本地同步存储桶并观察变化。

目前,同步效果很好:每当将新文件添加到存储桶中时,它都会在本地下载。但是,我不知道要监听文件更新,这是一种在本地下载新文件时触发的方法。可以做到吗?

这是我的一段代码:

#  --------
# | AWS S3 |
#  --------
s3.credentials-access-key=***
s3.credentials-secret-key=****
s3.bucket = my-bucket
s3.remote-dir = zips
s3.local-dir = D:/s3-bucket/
@Log4j2
@Configuration
public class S3Config {

    public static final String OUT_CHANNEL_NAME = "s3filesChannel";

    @Value("${s3.credentials-access-key}") private String accessKey;
    @Value("${s3.credentials-secret-key}") private String secretKey;
    @Value("${s3.remote-dir}") private String remoteDir;
    @Value("${s3.bucket}") private String s3bucket;
    @Value("${s3.local-dir}") private String localDir;

    /*
     * AWS S3
     */
    @Bean
    public AmazonS3 getAmazonS3(

    ){
        BasicAWSCredentials creds = new BasicAWSCredentials(accessKey, secretKey);
        AmazonS3 s3client = AmazonS3ClientBuilder
                .standard()
                .withRegion(Regions.EU_WEST_1)
                .withCredentials(new AWSStaticCredentialsProvider(creds))
                .build();
        return s3client;        
    }

    @Bean
    public S3SessionFactory s3SessionFactory(AmazonS3 pAmazonS3) {
        return new S3SessionFactory(pAmazonS3);
    }

    @Bean
    public S3InboundFileSynchronizer s3InboundFileSynchronizer(S3SessionFactory pS3SessionFactory) {
        S3InboundFileSynchronizer sync = new S3InboundFileSynchronizer(pS3SessionFactory);
        sync.setPreserveTimestamp(true);
        sync.setDeleteRemoteFiles(false);
        String fullRemotePath = s3bucket.concat("/").concat(remoteDir);
        sync.setRemoteDirectory(fullRemotePath);
        sync.setFilter(new S3RegexPatternFileListFilter(".*\\.zip$"));
        return sync;
    }

    @Bean
    @InboundChannelAdapter(value = OUT_CHANNEL_NAME, poller = @Poller(fixedDelay = "30"))
    public S3InboundFileSynchronizingMessageSource s3InboundFileSynchronizingMessageSource(
        S3InboundFileSynchronizer pS3InboundFileSynchronizer
    ) {
        S3InboundFileSynchronizingMessageSource messageSource = new S3InboundFileSynchronizingMessageSource(pS3InboundFileSynchronizer);
        messageSource.setAutoCreateLocalDirectory(true);
        messageSource.setLocalDirectory(new File(localDir));
        messageSource.setLocalFilter(new AcceptOnceFileListFilter<File>());
        return messageSource;
    }

    @Bean("s3filesChannel")
    public PollableChannel s3FilesChannel() {
        return new QueueChannel();
    }

    @Bean
    public IntegrationFlow fileReadingFlow(
            S3InboundFileSynchronizingMessageSource pS3InboundFileSynchronizingMessageSource,
            GtfsBizkaibus pGtfsBizkaibus,
            @Qualifier("fileProcessor") MessageHandler pMessageHandler) {
        return IntegrationFlows
                .from(pS3InboundFileSynchronizingMessageSource, e -> e.poller(p -> p.fixedDelay(5, TimeUnit.SECONDS)))
                .handle(pMessageHandler)
                .get();
    }

    @Bean("fileProcessor")
    public MessageHandler fileProcessor() {
        FileWritingMessageHandler handler = new FileWritingMessageHandler(new File(localDir));
        handler.setExpectReply(false); // end of pipeline, reply not needed
        handler.setFileExistsMode(FileExistsMode.APPEND);
        handler.setNewFileCallback((file, msg) -> {
            log.debug("New file created... " + file.getAbsolutePath());
        });
        return handler;
    }

标签: spring-bootamazon-s3spring-integrationspring-integration-aws

解决方案


您可以使用 S3 事件通知和 SQS 队列。基本上,当一个对象添加到您的存储桶时,S3 可以将事件发布到已注册的 SQS 队列。然后,您可以让您的本地应用程序长轮询队列以查找新事件并处理添加的任何事件。

请参阅此处了解更多信息。


推荐阅读