Poller / Adapter в s3-inbound-channel-adapter не работает должным образом - PullRequest
Новая
       42

Poller / Adapter в s3-inbound-channel-adapter не работает должным образом

0 голосов
/ 10 октября 2018

Мой сценарий заключается в переносе одного файла из корзины AWS S3 в экземпляр EC2 только один раз, когда файл изменяется.Я использовал следующую конфигурацию и вручную запустил адаптер при запуске сервера.

Проблема заключается в том, что при запуске сервера выполнение повторяется 5 или 6 раз.Похоже, что выполняются разные потоки.Я могу видеть другого исполнителя задач в журнале, не знаю, является ли это проблемой поллера или адаптера.

Я использую активатор службы для выполнения какой-либо другой операции, основанной на изменении файла в местоположении S3.

Примечание. Эта проблема возникает только один раз при запуске.Он работает нормально, дальнейшее изменение файла.

Конфиг: 

<bean id="s3SessionFactory" 
            class="org.springframework.integration.aws.support.S3SessionFactory"></bean>
        <bean id="acceptOnceFilter"
            class="org.springframework.integration.file.filters.AcceptOnceFileListFilter" />
        <task:executor id="s3PollingExecutor" pool-size="1" queue-capacity="10" />
        <integration:channel id="s3FilesChannel"/>
        <int-aws:s3-inbound-channel-adapter id="s3FileInbound"
                     channel="s3FilesChannel" 
                     session-factory="s3SessionFactory" 
                     auto-create-local-directory="false"
                     delete-remote-files="false" 
                     preserve-timestamp="true"
                     filter="acceptOnceFilter"
                     local-directory="local_directory"
                     auto-startup="false" 
                     remote-directory="s3_bucket">
            <integration:poller id="s3FilesChannelPoller" 
                                fixed-rate="1000" 
                                max-messages-per-poll="1" time-unit="MILLISECONDS" 
                                task-executor="s3PollingExecutor">
            </integration:poller>
        </int-aws:s3-inbound-channel-adapter>
        <integration:service-activator id="s3FilesChannelWatcher" 
                                       input-channel="s3FilesChannel" 
                                       output-channel="nullChannel"
                                       ref="configurationFileWatcher" 
                                       method="getConfigurationFileWatcher">
        </integration:service-activator>

Как вы и предполагали, я попробовал следующее. 

<bean id="acceptOnceFilterRegion"
      class="cS3FileFilterOnLastModifiedTime">
    <constructor-arg index="0" ref="metaDataStoreRegion"/>
    <constructor-arg index="1" value="*"/>
</bean>

логика добавлена ​​для проверки последнейвремя модификации 

import org.springframework.integration.aws.support.filters.S3PersistentAcceptOnceFileListFilter;
import org.springframework.integration.metadata.ConcurrentMetadataStore;
import com.amazonaws.services.s3.model.S3ObjectSummary;
public class S3FileFilterOnLastModifiedTime extends S3PersistentAcceptOnceFileListFilter {

    Long delayTime = 1000L;

    public S3FileFilterOnLastModifiedTime(ConcurrentMetadataStore store, String prefix) {
        super(store, prefix);
    }

    @Override
    public boolean accept(S3ObjectSummary file) {
        long lastModified = modified(file);
        long currentTime = System.currentTimeMillis();
        long timeDifference = currentTime - lastModified;
        return timeDifference > delayTime;
    }   
}

Все еще нет надежды. Журналы похожи на это ....... 

[INFO ] 2018-10-11 11:22:10,888 [s3PollingExecutor-1] ConfigurationSettingWatcher {} - ConfigurationSettingWatcher Started succesfully
[INFO ] 2018-10-11 11:22:10,892 [s3PollingExecutor-2] ConfigurationSettingWatcher {} - ConfigurationSettingWatcher Started succesfully
[INFO ] 2018-10-11 11:22:10,892 [s3PollingExecutor-3] ConfigurationSettingWatcher {} - ConfigurationSettingWatcher Started succesfully
[INFO ] 2018-10-11 11:22:10,893 [s3PollingExecutor-4] ConfigurationSettingWatcher {} - ConfigurationSettingWatcher Started succesfully
[INFO ] 2018-10-11 11:22:10,893 [s3PollingExecutor-5] ConfigurationSettingWatcher {} - ConfigurationSettingWatcher Started succesfully
[INFO ] 2018-10-11 11:22:10,894 [s3PollingExecutor-6] ConfigurationSettingWatcher {} - ConfigurationSettingWatcher Started succesfully

1 Ответ

0 голосов
/ 10 октября 2018

Я думаю, что вам не хватает того факта, что ваш s3_bucket не пуст при запуске, и <int-aws:s3-inbound-channel-adapter> подхватывает все файлы при его запуске.Вот как вы видите эти несколько задач: каждая для каждого файла.

Если вы действительно беспокоитесь только о новых файлах в этом сегменте, вам следует подумать, не использовать в памяти AcceptOnceFileListFilter, а переключаться на некоторыепостоянная реализация, основанная на общей реализации MetadataStore.Для этого есть S3PersistentAcceptOnceFileListFilter в spring-integration-aws и DynamoDbMetadataStore для сохранения результатов фильтрации в DynamoDb на AWS: https://github.com/spring-projects/spring-integration-aws#metadata-store-for-amazon-dynamodb

...