spring - 在多线程中进行Spring事务管理？

Question

Spring Transaction不支持多线程，因此我尝试在Thread的run()方法中手动管理事务。但是，它不起作用!

当下面有异常抛出时，我想在下面的示例中回滚每个线程的run()方法。 (在以下情况下，INSERT INTO UNKNOWN_TABLE)

我的预期结果是“开始，1、3、5，结束”。

实际结果是“开始，1、2、3、4、5，结束”。

欢迎任何回复!谢谢!

主类:

@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();

    }

}

服务等级:

@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类，让Spring Boot进行配置

使用TransactionTemplate而不是自己弄乱(错误!)Connection。

使用默认配置的Spring TaskExecutor，而不是手动访问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。