JOOQ和交易

Question

我一直在阅读关于交易和jooq但我很难看到如何在实践中实现它.

假设我为JOOQ提供了一个自定义ConnectionProvider,它恰好使用自动提交设置为false的连接池.

实施大致如下:

@Override public Connection acquire() throws DataAccessException {
    return pool.getConnection();
}

@Override public void release(Connection connection) throws DataAccessException {
    connection.commit();
    connection.close();
}

如何将两个jooq查询包装到单个事务中？

使用DefaultConnectionProvider很容易,因为只有一个连接 - 但是对于一个池我不知道如何去做.

Answer 1

jOOQ 3.4交易API

使用jOOQ 3.4,添加了一个事务API来抽象JDBC,Spring或JTA事务管理器.此API可以与Java 8一起使用:

DSL.using(configuration)
   .transaction(ctx -> {
       DSL.using(ctx)
          .update(TABLE)
          .set(TABLE.COL, newValue)
          .where(...)
          .execute();
   });

或者使用pre-Java 8语法

DSL.using(configuration)
   .transaction(new TransactionRunnable() {
       @Override
       public void run(Configuration ctx) {
           DSL.using(ctx)
              .update(TABLE)
              .set(TABLE.COL, newValue)
              .where(...)
              .execute();
       }
   });

想法是lambda表达式(或匿名类)形成事务代码,其中:

• 正常完成后提交
• 回滚例外

该org.jooq.TransactionProviderSPI可用于覆盖默认行为,它通过JDBC使用实现嵌套事务Savepoints.

一个春天的例子

当前文档显示了使用Spring进行事务处理时的示例:

• http://www.jooq.org/doc/latest/manual/getting-started/tutorials/jooq-with-spring/

这个例子基本上归结为使用Spring TransactionAwareDataSourceProxy

<!-- Using Apache DBCP as a connection pooling library.
     Replace this with your preferred DataSource implementation -->
<bean id="dataSource" class="org.apache.commons.dbcp.BasicDataSource"
    init-method="createDataSource" destroy-method="close">
    <property name="driverClassName" value="org.h2.Driver" />
    <property name="url" value="jdbc:h2:~/maven-test" />
    <property name="username" value="sa" />
    <property name="password" value="" />
</bean>

<!-- Using Spring JDBC for transaction management -->
<bean id="transactionManager"
    class="org.springframework.jdbc.datasource.DataSourceTransactionManager">
    <property name="dataSource" ref="dataSource" />
</bean>

<bean id="transactionAwareDataSource"
    class="org.springframework.jdbc.datasource.TransactionAwareDataSourceProxy">
    <constructor-arg ref="dataSource" />
</bean>

<!-- Bridging Spring JDBC data sources to jOOQ's ConnectionProvider -->
<bean class="org.jooq.impl.DataSourceConnectionProvider" 
      name="connectionProvider">
    <constructor-arg ref="transactionAwareDataSource" />
</bean>

GitHub提供了一个运行示例:

• https://github.com/jOOQ/jOOQ/tree/master/jOOQ-examples/jOOQ-spring-example

一个Spring和Guice的例子

虽然我个人不会推荐它,但有些用户已经成功用Guice取代Spring的DI部分并处理与Guice的交易.在GitHub上还有针对此用例的集成测试运行示例:

• https://github.com/jOOQ/jOOQ/tree/master/jOOQ-examples/jOOQ-spring-guice-example

Answer 2

这可能不是最好的方法，但它似乎有效。需要说明的是，它不是release，但commit它关闭连接，并将其返回到池，这是相当混乱，并可能导致问题，如果一些代码“忘记”提交方法...

所以客户端代码看起来像：

final PostgresConnectionProvider postgres =
            new PostgresConnectionProvider("localhost", 5432, params.getDbName(), params.getUser(), params.getPass())

private static DSLContext sql = DSL.using(postgres, SQLDialect.POSTGRES, settings);

//execute some statements here
sql.execute(...);

//and don't forget to commit or the connection will not be returned to the pool
PostgresConnectionProvider p = (PostgresConnectionProvider) sql.configuration().connectionProvider();
p.commit();

和 ConnectionProvider：

public class PostgresConnectionProvider implements ConnectionProvider {
    private static final Logger LOG = LoggerFactory.getLogger(PostgresConnectionProvider.class);

    private final ThreadLocal<Connection> connections = new ThreadLocal<>();
    private final BoneCP pool;

    public PostgresConnectionProvider(String serverName, int port, String schema, String user, String password) throws SQLException {
        this.pool = new ConnectionPool(getConnectionString(serverName, port, schema), user, password).pool;
    }

    private String getConnectionString(String serverName, int port, String schema) {
        return "jdbc:postgresql://" + serverName + ":" + port + "/" + schema;
    }

    public void close() {
        pool.shutdown();
    }

    public void commit() {
        LOG.debug("Committing transaction in {}", Thread.currentThread());
        try {
            Connection connection = connections.get();
            if (connection != null) {
                connection.commit();
                connection.close();
                connections.set(null);
            }
        } catch (SQLException ex) {
            throw new DataAccessException("Could not commit transaction in postgres pool", ex);
        }
    }

    @Override
    public Connection acquire() throws DataAccessException {
        LOG.debug("Acquiring connection in {}", Thread.currentThread());
        try {
            Connection connection = connections.get();
            if (connection == null) {
                connection = pool.getConnection();
                connection.setAutoCommit(false);
                connections.set(connection);
            }
            return connection;
        } catch (SQLException ex) {
            throw new DataAccessException("Can't acquire connection from postgres pool", ex);
        }
    }

    @Override
    //no-op => the connection won't be released until it is commited
    public void release(Connection connection) throws DataAccessException {
        LOG.debug("Releasing connection in {}", Thread.currentThread());
    }
}