Spring Batch CompositeItemWriter如何管理委托编写器的事务？ [英] How does Spring Batch CompositeItemWriter manage transaction for delegate writers?
本文介绍了Spring Batch CompositeItemWriter如何管理委托编写器的事务？的处理方法，对大家解决问题具有一定的参考价值，需要的朋友们下面随着小编来一起学习吧！
问题描述
在批处理作业步骤配置中，我计划在编写器中执行2个查询，第一个查询是更新表A中的记录，第二个查询是再次在表A中插入新记录。
到目前为止，我认为CompositeItemWriter可以实现上面的目标，即我需要创建两个JdbcBatchItemWriter，一个用于更新，另一个用于插入。
我的第一个问题是CompositeItemWriter是否符合上述要求？
如果是，则会引出关于交易的第二个问题。例如，如果第一次更新成功，而第二次插入失败。第一个更新事务是否会自动回滚？否则，如何在同一事务中手动拉取两个更新？提前感谢！
推荐答案
我的第一个问题是CompositeItemWriter是否符合上述要求？
是的，CompositeItemWriter是要走的路。
如果是，则会引出关于交易的第二个问题。例如，如果第一次更新成功，而第二次插入失败。第一个更新事务是否会自动回滚？否则，如何在同一事务中手动拉取两个更新？
问得好！是的，如果在第一个写入器中更新成功，然后在第二个写入器中插入失败，则所有语句都将自动回滚。您需要知道的是，事务围绕着面向块的微线程步骤的执行(因此围绕着复合项编写器的write方法)。因此，此方法中所有SQL语句的执行(在委托编写器中执行)将是原子的。
为了说明此用例，我编写了以下测试：
- 给定一个表
people，其中有两列id和name，其中只有一条记录：1,'foo' - 假设一个作业读取两条记录(
1,'foo'、2,'bar')，并尝试将foo更新为foo!!，然后在表中插入2,'bar'。这是通过CompositeItemWriter和UpdateItemWriter和InsertItemWriter两个项编写器 完成的
- 用例是
UpdateItemWriter成功但InsertItemWriter失败(抛出异常) - 预期结果是
foo没有更新为foo!!，表中也没有插入bar(由于InsertItemWriter中的异常，两条SQL语句都被回滚)
以下是代码(它是自包含的，因此您可以尝试并查看它是如何工作的，它使用一个应该位于类路径中的嵌入式hsqldb数据库)：
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.CompositeItemWriter;
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 = TransactionWithCompositeWriterTest.JobConfiguration.class)
public class TransactionWithCompositeWriterTest {
@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));");
jdbcTemplate.update("INSERT INTO people (id, name) VALUES (1, 'foo');");
}
@Test
public void testTransactionRollbackWithCompositeWriter() throws Exception {
// given
int peopleCount = JdbcTestUtils.countRowsInTable(jdbcTemplate, "people");
int fooCount = JdbcTestUtils.countRowsInTableWhere(jdbcTemplate, "people", "id = 1 and name = 'foo'");
int barCount = JdbcTestUtils.countRowsInTableWhere(jdbcTemplate, "people", "id = 2 and name = 'bar'");
Assert.assertEquals(1, peopleCount);
Assert.assertEquals(1, fooCount);
Assert.assertEquals(0, barCount);
// when
JobExecution jobExecution = jobLauncherTestUtils.launchJob();
// then
Assert.assertEquals(ExitStatus.FAILED.getExitCode(), jobExecution.getExitStatus().getExitCode());
Assert.assertEquals("Something went wrong!", jobExecution.getAllFailureExceptions().get(0).getMessage());
StepExecution stepExecution = jobExecution.getStepExecutions().iterator().next();
Assert.assertEquals(0, stepExecution.getCommitCount());
Assert.assertEquals(1, stepExecution.getRollbackCount());
Assert.assertEquals(0, stepExecution.getWriteCount());
peopleCount = JdbcTestUtils.countRowsInTable(jdbcTemplate, "people");
fooCount = JdbcTestUtils.countRowsInTableWhere(jdbcTemplate, "people", "id = 1 and name = 'foo'");
barCount = JdbcTestUtils.countRowsInTableWhere(jdbcTemplate, "people", "id = 2 and name = 'bar'");
Assert.assertEquals(1, peopleCount); // bar is not inserted
Assert.assertEquals(0, barCount); // bar is not inserted
Assert.assertEquals(1, fooCount); // foo is not updated to "foo!!"
}
@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");
return new ListItemReader<>(Arrays.asList(foo, bar));
}
@Bean
public ItemWriter<Person> updateItemWriter() {
return new UpdateItemWriter(dataSource());
}
@Bean
public ItemWriter<Person> insertItemWriter() {
return new InsertItemWriter(dataSource());
}
@Bean
public ItemWriter<Person> itemWriter() {
CompositeItemWriter<Person> compositeItemWriter = new CompositeItemWriter<>();
compositeItemWriter.setDelegates(Arrays.asList(updateItemWriter(), insertItemWriter()));
return compositeItemWriter;
}
@Bean
public Job job(JobBuilderFactory jobBuilderFactory, StepBuilderFactory stepBuilderFactory) {
return jobBuilderFactory.get("job")
.start(stepBuilderFactory
.get("step").<Person, Person>chunk(2)
.reader(itemReader())
.writer(itemWriter())
.build())
.build();
}
@Bean
public JobLauncherTestUtils jobLauncherTestUtils() {
return new JobLauncherTestUtils();
}
}
public static class UpdateItemWriter implements ItemWriter<Person> {
private JdbcTemplate jdbcTemplate;
public UpdateItemWriter(DataSource dataSource) {
this.jdbcTemplate = new JdbcTemplate(dataSource);
}
@Override
public void write(List<? extends Person> items) {
for (Person person : items) {
if ("foo".equalsIgnoreCase(person.getName())) {
jdbcTemplate.update("UPDATE people SET name = 'foo!!' WHERE id = 1");
}
}
}
}
public static class InsertItemWriter implements ItemWriter<Person> {
private JdbcTemplate jdbcTemplate;
public InsertItemWriter(DataSource dataSource) {
this.jdbcTemplate = new JdbcTemplate(dataSource);
}
@Override
public void write(List<? extends Person> items) {
for (Person person : items) {
if ("bar".equalsIgnoreCase(person.getName())) {
jdbcTemplate.update("INSERT INTO people (id, name) VALUES (?, ?)", person.getId(), person.getName());
throw new IllegalStateException("Something went wrong!");
}
}
}
}
public static class Person {
private long id;
private String name;
public Person() {
}
public 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;
}
}
}
我的示例使用自定义项编写器，但这应该也适用于两个JdbcBatchItemWriter。
我希望这会有帮助！
这篇关于Spring Batch CompositeItemWriter如何管理委托编写器的事务？的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！
查看全文
