Как переопределить транзакцию управления Spring Batch CompositeItemWriter для авторов делегатов в случае возникновения исключения?

Question

Я расширяю это Как Spring Batch CompositeItemWriter управляет транзакциями для авторов делегатов? вопрос здесь:

В моем случае у меня ниже CompositeItemWriter, который записывает данные в несколько таблицы той же базы данных, перед записью данных он преобразует данные, реализуя различные бизнес-правила. Здесь одна запись может удовлетворять разным бизнес-правилам и т. Д. c. Следовательно, один писатель может получить больше данных, чем другие.

@Bean
public CompositeItemWriter<Employee> EmployeeCompositeWriter() throws Exception {
    List<ItemWriter<? super Employee>> employee = new ArrayList<>();
    employee.add(employeeWriter());
    employee.add(departmentWriter());
    employee.add(stockWriter());
    employee.add(purchaseWriter());

    CompositeItemWriter<Employee> compositeItemWriter = new CompositeItemWriter<>();
    compositeItemWriter.setDelegates(employee);
    compositeItemWriter.afterPropertiesSet();
    return compositeItemWriter;
}

Сценарий - Предположим, 1-й писатель работает очень хорошо, 2-й писатель генерирует исключение, тогда 3-й и 4-й писатели не получают вызова, это что Automic характер по умолчанию в Spring Batch происходит из-за отката транзакции.

Здесь, даже если какое-либо исключение возникает во 2-м писателе, я хочу успешно вызвать 3-й и 4-й писатели и сохранить данные, я также хотел успешно сохранить данные 1-го писателя и 2-го писателя ... только данные исключения, которые я хочу сохранить в таблице ошибок с помощью SkipListener, чтобы определить, какие записи были нежелательными или мусорными.

Решение - Чтобы достичь вышеупомянутого сценария , мы добавили @Transactional(propagation = Propagation.REQUIRES_NEW) для каждого метода записи писателей, первый писатель сохранил данные сейчас, а второй писатель генерирует исключение (используя namedJdbcTemplate.batchUpdate() для массового обновления данных), мы кэшируем его и повторно выбрасывая его, но мы могли видеть, что уровень фиксации снижен до 1 (отклонение от курса для определения точной записи мусора) и мама Исключение ent возникает из-за 2-го писателя, снова вызывается 1-й писатель, и он сохраняет повторяющиеся данные, и вызываются писатели 2-й, 3-й и 4-й, но также эта нежелательная запись не поступает на 3-й и 4-й писатели.

Здесь Я не хочу, чтобы все пакетное задание останавливалось, если одна или несколько записей являются мусором, потому что это задание критически важно для нас, чтобы запускать его каждый раз. Есть ли способ сохранить все данные там, где исключение не возникает, и только сохранить данные исключения в таблице ошибок с помощью SkipListener, если возможно, или любым другим способом?

Есть ли способ можем ли мы повторно использовать компоненты пакета, такие как (ЧИТАТЕЛЬ или ПРОЦЕССОР) любого шага, на другом шаге?

Answer 1

Я не вижу, как вы могли бы согласовать единую транзакцию spring -batch для записи всего фрагмента как atomi c с вашей идеей сохранения атомарности для отдельных писателей, пока вы хотите skiplistener.

Я не уверен, возможно ли это, но, возможно, вы сможете быстро это проверить. Вот как сообщение переносит исключение в некоторых средах интеграции, таких как camel от одного процессора к потоку обработки ошибок.

• Считыватель элементов должен вернуть EmployeeWrapper, который содержит запись employee и имеет поле для хранения исключения.

• ваш CompositeItemWriter получает List<EmployeeWrapper>, а составной писатель имеет 5 писателей вместо 4. И 5-й писатель будет делать то, что сделал бы ваш SkipListener.

    List<ItemWriter<? super EmployeeWrapper>> employee = new ArrayList<>();
    employee.add(employeeWriter());
    employee.add(departmentWriter());
    employee.add(stockWriter());
    employee.add(purchaseWriter());
    employee.add(errorRecordWriter());

• Ваши первые 4 отдельных писателя никогда не генерируют исключение, вместо этого отмечают его как обработанное, но добавляют пойманное исключение как атрибут EmployeeWrapper.

• Ваш 5-й errorRecordWriter получает все записи, проверяет все записи с добавленным атрибутом исключения и записывает их в таблицу ошибок. Если не удалось записать запись об ошибке, вы можете выбросить исключение, и все 5 писателей будут повторены.

• Относительно того, как вы узнаете, какая запись является записью об ошибке при сбое пакетного обновления. Кажется, что когда в фрагменте возникает ошибка, пружина откатывает фрагмент и начинает повторять попытку записи за записью в этом фрагменте, чтобы знать, какая запись является проблемной c. Таким образом, вы можете делать то же самое с отдельными писателями. Т.е. перехватить исключение пакетного обновления и затем повторить их одну за другой, чтобы разделить записи об ошибках

Answer 2

Основная root причина проблемы заключалась в том, что мы пытались использовать два разных ItemWriter для записи данных в одну и ту же таблицу, и это приводило к странному поведению транзакции.

Мы ' ve реализовал SkipListenets (учитывая тот факт, что использование может не часто получать мусор или нежелательные данные, поскольку мы выполняем проверку при начальной загрузке данных.)

Поскольку мы реализовали «Spring Методика пакетного пропуска » в пакетных заданиях, это помогает нам указать определенные типы исключений и максимальное количество. пропущенных элементов и всякий раз, когда возникает одно из этих пропускаемых исключений, пакетное задание не завершается неудачно, а пропускает этот конкретный элемент и переходит к следующему элементу. Только когда максимум нет. пропущенных элементов, пакетное задание завершится ошибкой. Мы использовали skip logi c с «Отказоустойчивость». функции Spring Batch применяются к элементам на этапах, ориентированных на фрагменты, а не на весь шаг.

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

Answer 3

Здесь пара вещей:

1. Не используйте @Transactional с Spring Batch - Spring Batch управляет транзакциями за вас, поэтому использование этой аннотации вызовет проблемы. Не используйте его.
2. Управляйте исключениями самостоятельно - В сценарии, который вы описываете, вы хотите вызвать четыре ItemWriter реализации для одного и того же элемента, но хотите пропустить исключения на делегированном уровне ItemWriter вам нужно будет написать свою собственную реализацию CompositeItemWriter. Spring Batch обеспечивает такой уровень композиции (где мы делегируем каждой реализации ItemWriter один и тот же элемент) для удобства, но с точки зрения фреймворка это всего лишь один ItemWriter. Чтобы обрабатывать исключения на дочернем уровне ItemWriter, вам нужно будет написать свою собственную оболочку и самостоятельно управлять исключениями.

UPDATE: Пример реализации пользовательского ItemWriter, о котором я говорю (обратите внимание, что приведенный ниже код не протестирован):

public class MyCompositeItemWriter<T> implements ItemWriter<T> {
      private List<ItemWriter<? super T>> delegates;
 
    @Override
      public void write(List<? extends T> items) throws Exception {
            for(ItemWriter delegate : delegates) {
               try {
                  delegate.write(items);
               }
               catch (Exception e) {
                  // Do logging/error handling here
               }
            }
    }

    @Override
    public void setDelegates(List<ItemWriter<? super T>> delegates) {
        super.setDelegates(delegates);
        this.delegates = delegates;
    }
}