Как обрабатывать файлы в отдельных потоках?

Question

Как обрабатывать файлы в отдельных потоках?

Существует локальный каталог, в котором файлы помещаются другими способами, а новые файлы с тем же именем заменяют старые. Я хочу переместить файлы из/local в/обрабатывать dir и активировать некоторые службы. Далее в конце цепочки фильтров задача очистки удалит файлы из/обработки. Я сделал это работает 1 на 1, но обработка занимает минут, так что я хотел бы

1. Добавить многопоточность: то есть несколько файлов перемещаются и обработаны одновременно.

2. Если файл еще не обработан, скажем, «File1.abc», и новая версия этого файла была помещена в/local, тогда нет необходимости обрабатывать старое сообщение со старой версией файла. То есть Сообщения должны быть посланы только для версии файлов в момент, когда они перемещаются из/локального до/обработки

Я пытаюсь что-то вроде этого:

<file:inbound-channel-adapter channel="processingChannel" 
           directory="#{localDir}" 
           prevent-duplicates="false" filter="acceptAllFileListFilter"> 
    <int:poller fixed-rate="20" max-messages-per-poll="3" task-executor="executor"/> 
</file:inbound-channel-adapter> 

<task:executor id="executor" pool-size="3" queue-capacity="0" rejection-policy="ABORT"/> 

<file:outbound-gateway  request-channel="processingChannel" reply-channel="serviceChannel" 
           directory="#{processing}" 
           auto-create-directory="true" 
           filename-generator-expression="payload.name + '_' + { T(java.lang.System).currentTimeMillis()}" 
           delete-source-files="true" 
           mode="FAIL" /> 

<int:service-activator input-channel="serviceChannel" output-channel="furtherChannels" 
         ref="someService" method="process"> 
</int:service-activator> 

<bean id="someService" class="com.dot.SomeService"/> 

Но это не работает, и я не могу выяснить, как это исправить. Я пробовал разные способы, но всегда есть ошибки, такие как сообщения для уже удаленных файлов или некоторые другие проблемы. Сама задача кажется простой. Как заставить файлы обрабатывать в 3 потоках и отправлять сообщения только для актуальных версий файлов? Может быть, проблема здесь с опросом потребителей, но входящий адаптер используется только с этим потребителем, не так ли?

Answer 1

Я пока не вижу решения для вас, но, возможно, вы просто не объясняете проблему должным образом ... Попробуйте разделить деловые требования.

Плюс я не вижу причины для <file:outbound-gateway>. Вы можете просто читать файлы из каталога /local и обрабатывать их. Для параллелизма и отброса логики процессов в полете можно использовать некоторые настраиваемые FileListFilter, откуда вы должны определить новую версию файла и по его ключу cancel() запущенный процесс для запуска нового в конце текущего poll().

Может быть какое-то другое решение, но давайте начнем с бизнес-требований!