Добавить пакетные возможности для потока интеграции

Question

В настоящее время у меня есть поток интеграции Spring, который отлично работает (см. Ссылку на диаграмму). Я хотел бы добавить Batch поверх моей текущей конфигурации, чтобы позволить повторить попытку с экспоненциальным отклонением, шаблон автоматического выключателя и сохраняющиеся задания для базы данных для перезапуска.

поток Интеграция состоит из шлюза, который принимает сообщение <MyObj>, который в конечном итоге направляется в трансформатор, который преобразует сообщение <MyObj> на сообщение < Строка >. Затем агрегатор принимает сообщение <String> и в конечном итоге освобождает конкатенированное сообщение <String> (используя как стратегию выпуска размера, так и MessageGroupStoreReaper с таймаутом). Конкатенированная строка является полезной нагрузкой файла, загружаемого с использованием адаптера исходящего канала SFTP.

Я искал, прочитал документы, просмотрел множество примеров, и я не могу понять, как инкапсулировать последний шаг процесса в пакетное задание. Мне нужна возможность повторить загрузку String (как полезную нагрузку файла), если есть проблема с SFTP-подключением или другое исключение, возникшее во время загрузки. Я также хочу, чтобы иметь возможность перезагрузки (с использованием базы данных, поддерживаемой JobRepository) в случае некоторого сбоя, поэтому я не думаю, что использование Retry Advice достаточно.

Пожалуйста, объясните и помогите мне понять, как соединять куски и какие использовать (запуск задания-шлюз, MessageToJobRequest Transformer, ItemReader, ItemWriter?). Я также не знаю, как получить доступ к каждому сообщению <String> и отправить в SFTP-канал-адаптер внутри Job, Step или Tasklet.

Текущий поток:

Answer 1

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

1. <int-sftp:outbound-channel-adapter> имеет <request-handler-advice-chain>, где вы можете настроить RequestHandlerRetryAdvice и RequestHandlerCircuitBreakerAdvice.

2. Для достижения опции restartable вы можете сделать входной канал этого адаптера в качестве постоянной очереди с помощью message-store.

Теперь о пакете.

Чтобы начать работу из потока интеграции, вы должны написать MessageToJobRequest и использовать после этого <batch-int:job-launching-gateway>. Здесь, конечно, вы можете разместить payload до jobParameters.

Чтобы отправить сообщение от Job на какой-либо канал (например, на sftp-адаптер), вы можете использовать org.springframework.batch.integration.chunk.ChunkMessageChannelItemWriter.

Подробнее здесь: http://docs.spring.io/spring-batch/reference/html/springBatchIntegration.html