Spring 多线程中的事务管理？

Question

Spring事务不支持多线程，所以我尝试在Thread的运行()方法中手动管理事务。但是，它不起作用！

我想在下面的示例中回滚每个线程的运行() 方法，当其中有异常抛出时。（在下面的例子中，INSERT INTO UNKNOWN_TABLE）

我的预期结果是 'start, 1, 3, 5, end'。

而实际结果是'start, 1, 2, 3, 4, 5, end'。

欢迎任何回复！谢谢！

主要Class:

@SpringBootApplication
public class Application implements CommandLineRunner {

    @Autowired
    private TestService testService;

    public static void main(String[] args) {

        SpringApplication.run(Application.class, args);

    }

    @Bean
    public DriverManagerDataSource createDriverManagerDataSource() {

        HikariDataSource dataSource = new HikariDataSource();
        dataSource.setDriverClassName("oracle.jdbc.driver.OracleDriver");
        dataSource.setJdbcUrl("jdbc:oracle:thin:@url:port/schema");
        dataSource.setUsername("xxxx");
        dataSource.setPassword("xxxx");

        return dataSource;

    }

    @Bean
    public JdbcTemplate createJdbcTemplate() {

        JdbcTemplate jdbcTemplate = new JdbcTemplate();
        jdbcTemplate.setDataSource(createDriverManagerDataSource());

        return jdbcTemplate;

    }

    @Override
    public void run(String... args) throws Exception {

        testService.test();

    }

}

服务Class：

@Service
public class TestService {

    @Autowired
    private JdbcTemplate jdbcTemplate;

    @Transactional(rollbackFor = Exception.class)
    public void test() throws Exception {

        jdbcTemplate.batchUpdate("INSERT INTO TB_MYTEST(MYKEY, MYVALUE) VALUES ('start', 'start')");

        ExecutorService executorService = Executors.newFixedThreadPool(5);

        for (int i = 1; i <= 5; i++) {

            executorService.submit(new TestRunner(i));

        }

        executorService.shutdown();
        executorService.awaitTermination(Long.MAX_VALUE, TimeUnit.NANOSECONDS);

        jdbcTemplate.batchUpdate("INSERT INTO TB_MYTEST(MYKEY, MYVALUE) VALUES ('end', 'end')");
    }

    private class TestRunner implements Runnable {

        private Integer id;

        public TestRunner(Integer id) {

            this.id = id;

        }

        @Override
        public void run() {

            try (Connection connection = jdbcTemplate.getDataSource().getConnection()) {

                try {

                    connection.setAutoCommit(false);

                    String sqlString = String.format("INSERT INTO TB_MYTEST(MYKEY, MYVALUE) VALUES ('%d', '%d')", id, id);
                    jdbcTemplate.batchUpdate(sqlString);

                    if (id % 2 == 0) {
                        // Except the transaction been rollback when this.id is 2 or 4.
                        jdbcTemplate.batchUpdate("INSERT INTO UNKNOWN_TABLE(MYKEY, MYVALUE) VALUES ('no', 'no')");

                    }

                    connection.commit();

                } catch (Exception e) {

                    System.err.println("Failure: UNKNOWN_TABLE");
                    connection.rollback();

                } finally {

                    connection.close();

                }

            } catch (SQLException e2) {

                e2.printStackTrace();

            }

        }

    }

}

Answer 1

当你试图超越 Spring 和 Spring Boot 时，你的代码有一些事情。与其尝试这样做，不如使用框架而不是围绕它们工作。

放弃你的 @Configuration class 并让 Spring Boot 进行配置
使用 TransactionTemplate 而不是自己乱搞（错误！）Connection。
使用默认配置的SpringTaskExecutor，而不是手动访问一个Executor。

将此添加到您的 application.properties

spring.datasource.url=jdbc:oracle:thin:@url:port/schema
spring.datasource.username=xxxx
spring.datasource.password=xxxx

使用 TransactionTemplate 而不是搞乱连接。

@@SpringBootApplication
public class Application {

    private static final String SQL = "INSERT INTO TB_MYTEST(MYKEY, MYVALUE) VALUES (?, ?)";
    private static final String ERROR_SQL = "INSERT INTO UNKNOWN_TABLE(MYKEY, MYVALUE) VALUES (?, ?)";


    public static void main(String[] args) {

        SpringApplication.run(Application.class, args);
    }

    @Bean
    public CommandLineRunner testRunner(JdbcTemplate jdbc, TransactionTemplate tx, TaskExecutor tasks) {
        return (args) -> {
            jdbc.update(SQL, "start", "start");
            IntStream.range(1, 6)
                    .forEach(id -> {
                        try {
                            tasks.execute(() -> tx.executeWithoutResult((s) -> {
                                jdbc.update(SQL, id, id);
                                if (id % 2 == 0) {
                                    jdbc.update(ERROR_SQL, "no", "no");
                                }
                            }));
                        } catch (DataAccessException e) {
                            e.printStackTrace();
                        }
                    });
            jdbc.update(SQL, "end", "end");
        };
    }
}

像上面这样的东西会产生你想要的结果。请注意，您现在使用框架提供的 JdbcTemplate、TransactionTemplate 和 TaskExecutor。

Answer 2

参考@M后。 Deinum 回答，我已将我的代码更改为以下内容，它满足了我的需求。

application.properties

spring.datasource.url=jdbc:oracle:thin:@ip:port/schema
spring.datasource.username=xxxx
spring.datasource.password=xxxx
spring.datasource.driver-class-name=oracle.jdbc.OracleDriver

主要Class

@SpringBootApplication
public class Application implements CommandLineRunner {

    @Autowired
    private TestService testService;

    public static void main(String[] args) {

        SpringApplication.run(Application.class, args);

    }

    @Override
    public void run(String... args) throws Exception {

        testService.test();
        System.exit(0);

    }

}

测试服务

@Service
public class TestService {

    @Autowired
    private JdbcTemplate jdbcTemplate;

    @Autowired
    private PlatformTransactionManager transactionManager;

    @Transactional(rollbackFor = Exception.class)
    public void test() throws Exception {

        jdbcTemplate.batchUpdate("INSERT INTO TB_MYTEST(MYKEY, MYVALUE) VALUES ('start', 'start')");

        ExecutorService executorService = Executors.newFixedThreadPool(5);

        for (int i = 1; i <= 5; i++) {

            executorService.submit(new TestRunner(i));

        }

        executorService.shutdown();
        executorService.awaitTermination(Long.MAX_VALUE, TimeUnit.NANOSECONDS);

        jdbcTemplate.batchUpdate("INSERT INTO TB_MYTEST(MYKEY, MYVALUE) VALUES ('end', 'end')");
    }

    private class TestRunner implements Runnable {

        private Integer id;

        public TestRunner(Integer id) {

            this.id = id;

        }

        @Override
        public void run() {

            TransactionTemplate transactionTemplate = new TransactionTemplate(transactionManager);

            transactionTemplate.execute(new TransactionCallbackWithoutResult() {

                @Override
                protected void doInTransactionWithoutResult(TransactionStatus status) {

                    String sqlString = String.format("INSERT INTO TB_MYTEST(MYKEY, MYVALUE) VALUES ('%d', '%d')", id, id);
                    jdbcTemplate.batchUpdate(sqlString);

                    if (id % 2 == 0) {

                        jdbcTemplate.batchUpdate("INSERT INTO UNKNOWN_TABLE(MYKEY, MYVALUE) VALUES ('no', 'no')");

                    }

                }

            });

        }

    }

}

结果'start, 1, 3, 5, end'.

Spring 多线程中的事务管理？

Spring Transaction Management in Multi-Thread?

spring

multithreading

transactions