我有一个挑战,我需要从SQL Server数据库中读取“未处理”数据,处理数据,然后有选择地更新DB2数据库中的两到六个表,然后将这些数据标记为在SQL Server上的原始数据库中处理。在任何时候,如果有任何失败,我希望所有更新都回滚。如果我有10个未处理的项目,9个是好的,但是一个失败,我仍然希望9个好的项目完成,第10个项目返回原始状态,直到我们可以研究问题并进行修正。
总体架构是一个输入实例可能导致插入至少3个DB2表和多达7个表。一些DB2表最终可能会从一个输入中进行多次插入。我必须为每个表更新开发一个不同的编写器,并弄清楚如何向每个编写器传递该表所需的特定数据。我还需要分别利用2个数据源来更新DB2和SQL Server。
我不是一个经验丰富的Spring Batch开发人员。我很少有一个项目,我可以“读取1,处理1,写1”并重复。通常我需要读取几个文件/数据库,处理该数据,然后写入一个或多个报告,文件和/或数据库。我看到这种应用程序提供了哪些支持,但它更复杂,需要更多的研究,只能找到有限的例子。
在我尝试实施解决方案时,我采取了轻松的道路。我开发了一个实现Tasklet的类,并按照我的实时过程的方式编写代码。它使用JDBCTemplate从SQL获取输入数据,然后将数据传递给处理数据的代码并确定需要更新的内容。我有一个事务管理器类,使用REQUIRES_NEW实现@Transactional并回滚我的自定义未经检查的异常。 Transactional类捕获所有DataAccessException事件,并将抛出自定义异常。目前我只使用DB2数据源,以免使情况过于复杂。
在我的测试中,我在更新过程结束时添加了代码,引发了未经检查的异常。我希望可以回滚更新。但它没有发生。如果我重新运行该过程,我会在DB2上获得803个错误。
最后一件事。在我们的商店中,我们需要使用DB2上的存储过程来进行所有访问。所以我使用SimpleJdbcCall来执行SP。
这是我的代码:
Tasklet的主要java类:
public class SynchronizeDB2WithSQL implements Tasklet
{
private static final BatchLogger logger = BatchLogger.getLogger();
private Db2UpdateTranManager tranMgr;
public void setTranMgr(Db2UpdateTranManager tranMgr) {
this.tranMgr = tranMgr;
}
private AccessPaymentIntegrationDAO pmtIntDAO;
public void setPmtIntDAO(AccessPaymentIntegrationDAO pmtIntDAO) {
this.pmtIntDAO = pmtIntDAO;
}
@Override
public RepeatStatus execute(StepContribution arg0, ChunkContext arg1) throws Exception {
logger.logInfoMessage("=============================================");
logger.logInfoMessage(" EB0255IA - Synchronize DB2 with SQL");
logger.logInfoMessage("=============================================");
List<UnprocessedPaymentDataBean> orderList = this.pmtIntDAO.fetchUnprocessedEntries();
if(CollectionUtils.isNotEmpty(orderList)) {
for(UnprocessedPaymentDataBean ent: orderList) {
logger.logDebugMessage(" Processing payment ");
logger.logDebugMessage(ent.toString());
Map<String, List<PaymentTransactionDetailsBean>> paymentDetails = arrangePayments(this.pmtIntDAO.getDetailsByOrder(ent.getOrderNbr()));
try {
this.tranMgr.createNewAuthorizedPayment(ent, paymentDetails);
} catch (DataException e) {
logger.logErrorMessage("Encountered a Data Exception: "+e);
}
}
} else {
logger.logInfoMessage("=============================================");
logger.logInfoMessage("No data was encountered that needed to be processed");
logger.logInfoMessage("=============================================");
}
return RepeatStatus.FINISHED;
}
和Spring Batch xml:
<job id="EB0255IA" parent="baseJob" job-repository="jobRepository"
xmlns="http://www.springframework.org/schema/batch" restartable="true"
incrementer="parameterIncrementer">
<description>Job to maintain the DB2 updates for payment activity</description>
<step id="SynchronizeDB2WithSQL">
<tasklet ref="synchronizeTasklet" />
</step>
</job>
<bean id="synchronizeTasklet" class="com.ins.pmtint.synchdb2.SynchronizeDB2WithSQL" >
<property name="pmtIntDAO" ref="pmtIntDAO" />
<property name="tranMgr" ref="db2TranMgr" />
</bean>
<bean id="jdbcUpdateDB2" class="com.ins.pmtint.db.JDBCUpdateDB2">
<property name="dataSource" ref="dataSourceBnkDB2" />
</bean>
<bean id="updateDB2DataDAO" class="com.ins.pmtint.db.dao.UpdateDB2DataDAOImpl">
<property name="jdbcUpdateDB2" ref="jdbcUpdateDB2" />
</bean>
<bean id="db2TranMgr" class="com.ins.pmtint.db.tranmgr.Db2UpdateTranManagerImpl">
<property name="updateDB2DataDAO" ref="updateDB2DataDAO" />
</bean>
<bean id="jdbcPaymentIntegration" class="com.ins.pmtint.db.JDBCPaymentIntegration" >
<property name="dataSource" ref="dataSourcePmtIntegration" />
</bean>
<bean id="pmtIntDAO" class="com.ins.pmtint.db.dao.AccessPaymentIntegrationDAOImpl">
<property name="jdbcPaymentIntegration" ref="jdbcPaymentIntegration" />
</bean>
部分事务管理器实现。
public class Db2UpdateTranManagerImpl implements Db2UpdateTranManager, DB2FieldNames {
private static final BatchLogger logger = BatchLogger.getLogger();
UpdateDB2DataDAO updateDB2DataDAO;
public void setUpdateDB2DataDAO(UpdateDB2DataDAO updateDB2DataDAO) {
this.updateDB2DataDAO = updateDB2DataDAO;
}
@Override
@Transactional(propagation = Propagation.REQUIRES_NEW, readOnly = false, rollbackFor = DataException.class)
public void createNewAuthorizedPayment(UnprocessedPaymentDataBean dataBean, Map<String, List<PaymentTransactionDetailsBean>> paymentDetails) {
logger.logDebugMessage("At Db2UpdateTranManagerImpl.createNewAuthorizedPayment(");
logger.logDebugMessage(dataBean.toString());
String orderNbr = String.valueOf(dataBean.getOrderNbr());
String eventCode = TranTypeCode.fromValue(dataBean.getTransactionTypeCode()).getDB2Event();
if(eventCode == null) {
try {
KFBDistBatchEMail.createAndSendMessage("There is no event code for current entry\n\nOrder: "+orderNbr+" Tran type: "+dataBean.getTransactionTypeCode(), "EB0255IA - Database error" ,EnhancedPropertyPlaceholderConfigurer.getEmailFrom(), EnhancedPropertyPlaceholderConfigurer.getEmailTo(), null);
throw new DataException("Update failed: No event code to apply");
} catch (EMailExcpetion e2) {
logger.logErrorMessage("Generating email", e2);
}
}
String orginatingSystemId;
if (dataBean.getPaymentTypeCode().equalsIgnoreCase("EFT"))
orginatingSystemId = "FS";
else
orginatingSystemId = "IN";
try {
if(dataBean.getTransactionTypeCode().equalsIgnoreCase("A")) {
this.updateDB2DataDAO.updatePaymentDetails(orderNbr, DB_INITIAL_EVENT_CODE, "", dataBean.getTransactionAmt(), orginatingSystemId);
}
**** FOR TESTING - AT THE END I HAVE ADDED ****
throw new DataException("I finished processing and backed out. \n\n"+dataBean);
}
这是JDBC代码的一部分:
public class JDBCUpdateDB2 extends JdbcDaoSupport
implements DB2FieldNames
{
private static final BatchLogger logger = KFBBatchLogger.getLogger();
public void updatePaymentDetails(String orderNbr, String eventCd, String authnbr, Double amount, String orginatingSystemId) {
SimpleJdbcCall jdbcCall = new SimpleJdbcCall(getDataSource()).withSchemaName(EnhancedPropertyPlaceholderConfigurer.getDB2Schema()).withProcedureName(UPDATE_PAYMENT_TRANSACTION_DB2_PROC);
MapSqlParameterSource sqlIn = new MapSqlParameterSource();
sqlIn.addValue(SP_BNKCRD_PMT_ORD_NBR, orderNbr);
sqlIn.addValue(SP_CLUSTERING_NBR_2, new StringBuilder(orderNbr.substring(Math.max(orderNbr.length() - 2, 0))).reverse().toString());
sqlIn.addValue(SP_BNKCRD_EVNT_CD, eventCd);
sqlIn.addValue(SP_CCTRAN_ERR_CD, "N");
sqlIn.addValue(SP_BNKCRD_PROC_RET_CD, "");
sqlIn.addValue(SP_BNKCRD_AUTH_CD, "G");
sqlIn.addValue(SP_ORIG_SYS_ID_TXT, orginatingSystemId);
sqlIn.addValue(SP_BNKCRD_TRAN_AMT, amount);
try {
jdbcCall.execute(sqlIn);
} catch (DataAccessException e) {
logger.logErrorMessage("Database error in updatePaymentDetails", e);
throw e;
}
}
由于您需要写入多个表,因此可以使用具有每个表的委托项编写器的CompositeItemWriter
。在这种情况下,代表应该是步骤中的registered as streams。您还可以创建一个单独的项目编写器,将3个(或更多)插入语句发布到不同的表(但我不建议这样做)。
如果我有10个未处理的项目,9个是好的但是一个失败了我仍然希望9个好的项目完成而第10个项目恢复到它的原始状态
如果使用容错步骤并且在写入块期间抛出可跳过的异常,则Spring Batch将扫描块以查找有缺陷的项(因为它无法知道哪个项导致了错误)。从技术上讲,Spring Batch会将块大小设置为1并且每个项目使用一个事务,因此只会回滚有缺陷的项目。这使您可以达到上述要求。这是一个自包含的示例,向您展示它是如何工作的:
import java.util.Arrays;
import java.util.List;
import javax.sql.DataSource;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.batch.core.ExitStatus;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.support.ListItemReader;
import org.springframework.batch.test.JobLauncherTestUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseBuilder;
import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseType;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringRunner;
import org.springframework.test.jdbc.JdbcTestUtils;
@RunWith(SpringRunner.class)
@ContextConfiguration(classes = ChunkScanningTest.JobConfiguration.class)
public class ChunkScanningTest {
@Autowired
private JobLauncherTestUtils jobLauncherTestUtils;
@Autowired
private JdbcTemplate jdbcTemplate;
@Before
public void setUp() {
jdbcTemplate.update("CREATE TABLE people (id INT IDENTITY NOT NULL PRIMARY KEY, name VARCHAR(20));");
}
@Test
public void testChunkScanningWhenSkippableExceptionInWrite() throws Exception {
// given
int peopleCount = JdbcTestUtils.countRowsInTable(jdbcTemplate, "people");
Assert.assertEquals(0, peopleCount);
// when
JobExecution jobExecution = jobLauncherTestUtils.launchJob();
// then
peopleCount = JdbcTestUtils.countRowsInTable(jdbcTemplate, "people");
int fooCount = JdbcTestUtils.countRowsInTableWhere(jdbcTemplate, "people", "id = 1 and name = 'foo'");
int bazCount = JdbcTestUtils.countRowsInTableWhere(jdbcTemplate, "people", "id = 3 and name = 'baz'");
Assert.assertEquals(1, fooCount); // foo is inserted
Assert.assertEquals(1, bazCount); // baz is inserted
Assert.assertEquals(2, peopleCount); // bar is not inserted
Assert.assertEquals(ExitStatus.COMPLETED.getExitCode(), jobExecution.getExitStatus().getExitCode());
StepExecution stepExecution = jobExecution.getStepExecutions().iterator().next();
Assert.assertEquals(3, stepExecution.getCommitCount()); // one commit for foo + one commit for baz + one commit for the last (empty) chunk
Assert.assertEquals(2, stepExecution.getRollbackCount()); // initial rollback for whole chunk + one rollback for bar
Assert.assertEquals(2, stepExecution.getWriteCount()); // only foo and baz have been written
}
@Configuration
@EnableBatchProcessing
public static class JobConfiguration {
@Bean
public DataSource dataSource() {
return new EmbeddedDatabaseBuilder()
.setType(EmbeddedDatabaseType.HSQL)
.addScript("/org/springframework/batch/core/schema-drop-hsqldb.sql")
.addScript("/org/springframework/batch/core/schema-hsqldb.sql")
.build();
}
@Bean
public JdbcTemplate jdbcTemplate(DataSource dataSource) {
return new JdbcTemplate(dataSource);
}
@Bean
public ItemReader<Person> itemReader() {
Person foo = new Person(1, "foo");
Person bar = new Person(2, "bar");
Person baz = new Person(3, "baz");
return new ListItemReader<>(Arrays.asList(foo, bar, baz));
}
@Bean
public ItemWriter<Person> itemWriter() {
return new PersonItemWriter(dataSource());
}
@Bean
public Job job(JobBuilderFactory jobBuilderFactory, StepBuilderFactory stepBuilderFactory) {
return jobBuilderFactory.get("job")
.start(stepBuilderFactory.get("step")
.<Person, Person>chunk(3)
.reader(itemReader())
.writer(itemWriter())
.faultTolerant()
.skip(IllegalStateException.class)
.skipLimit(10)
.build())
.build();
}
@Bean
public JobLauncherTestUtils jobLauncherTestUtils() {
return new JobLauncherTestUtils();
}
}
public static class PersonItemWriter implements ItemWriter<Person> {
private JdbcTemplate jdbcTemplate;
PersonItemWriter(DataSource dataSource) {
this.jdbcTemplate = new JdbcTemplate(dataSource);
}
@Override
public void write(List<? extends Person> items) {
System.out.println("Writing items: "); items.forEach(System.out::println);
for (Person person : items) {
if ("bar".equalsIgnoreCase(person.getName())) {
System.out.println("Throwing exception: No bars here!");
throw new IllegalStateException("No bars here!");
}
jdbcTemplate.update("INSERT INTO people (id, name) VALUES (?, ?)", person.getId(), person.getName());
}
}
}
public static class Person {
private long id;
private String name;
public Person() {
}
Person(long id, String name) {
this.id = id;
this.name = name;
}
public long getId() {
return id;
}
public void setId(long id) {
this.id = id;
}
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
@Override
public String toString() {
return "Person{" +
"id=" + id +
", name='" + name + '\'' +
'}';
}
}
}
此示例打印:
Writing items:
Person{id=1, name='foo'}
Person{id=2, name='bar'}
Person{id=3, name='baz'}
Throwing exception: No bars here!
Writing items:
Person{id=1, name='foo'}
Writing items:
Person{id=2, name='bar'}
Throwing exception: No bars here!
Writing items:
Person{id=3, name='baz'}
如您所见,在抛出skippable之后,每个块只包含一个项目(Spring Batch逐个扫描项目以确定有问题的项目),并且只写入有效项目。
有限的例子可以找到
我希望这个例子能够清楚地说明这个特征如果你想要一个复合项目编写器的例子,请看一下这个问题/答案:How does Spring Batch CompositeItemWriter manage transaction for delegate writers?
希望这可以帮助。