Java 类org.springframework.jdbc.datasource.DataSourceTransactionManager 实例源码

public AbstractDbDialect(final JdbcTemplate jdbcTemplate, LobHandler lobHandler){
    this.jdbcTemplate = jdbcTemplate;
    this.lobHandler = lobHandler;
    // 初始化transction
    this.transactionTemplate = new TransactionTemplate();
    transactionTemplate.setTransactionManager(new DataSourceTransactionManager(jdbcTemplate.getDataSource()));
    transactionTemplate.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRES_NEW);

    // 初始化一些数据
    jdbcTemplate.execute(new ConnectionCallback() {

        public Object doInConnection(Connection c) throws SQLException, DataAccessException {
            DatabaseMetaData meta = c.getMetaData();
            databaseName = meta.getDatabaseProductName();
            databaseMajorVersion = meta.getDatabaseMajorVersion();
            databaseMinorVersion = meta.getDatabaseMinorVersion();

            return null;
        }
    });

    initTables(jdbcTemplate);
}

@Test
public void testSpringLocalTx() throws Exception {
    DataSource ds = wrap(createHsqlDataSource());
    JdbcTemplate jdbc = new JdbcTemplate(ds);
    TransactionTemplate tx = new TransactionTemplate(new DataSourceTransactionManager(ds));

    jdbc.execute(DROP_USER);
    jdbc.execute(CREATE_TABLE_USER);

    tx.execute(ts -> jdbc.update(INSERT_INTO_USER, 1, "user1"));
    User user = tx.execute(ts -> jdbc.queryForObject(SELECT_FROM_USER_BY_ID, new BeanPropertyRowMapper<>(User.class), 1));
    assertEquals(new User(1, "user1"), user);
    tx.execute(ts -> jdbc.update(DELETE_FROM_USER_BY_ID, 1));

    tx.execute(ts -> {
        int nb = jdbc.update(INSERT_INTO_USER, 1, "user1");
        ts.setRollbackOnly();
        return nb;
    });
    try {
        user = tx.execute(ts -> jdbc.queryForObject(SELECT_FROM_USER_BY_ID, new BeanPropertyRowMapper<>(User.class), 1));
        fail("Expected a EmptyResultDataAccessException");
    } catch (EmptyResultDataAccessException e) {
        // expected
    }
}

@Test
public void testSpringLocalTx() throws Exception {
    DataSource ds = wrap(createH2DataSource());
    JdbcTemplate jdbc = new JdbcTemplate(ds);
    TransactionTemplate tx = new TransactionTemplate(new DataSourceTransactionManager(ds));

    jdbc.execute(DROP_USER);
    jdbc.execute(CREATE_TABLE_USER);

    tx.execute(ts -> jdbc.update(INSERT_INTO_USER, 1, "user1"));
    User user = tx.execute(ts -> jdbc.queryForObject(SELECT_FROM_USER_BY_ID, new BeanPropertyRowMapper<>(User.class), 1));
    assertEquals(new User(1, "user1"), user);
    tx.execute(ts -> jdbc.update(DELETE_FROM_USER_BY_ID, 1));

    tx.execute(ts -> {
        int nb = jdbc.update(INSERT_INTO_USER, 1, "user1");
        ts.setRollbackOnly();
        return nb;
    });
    try {
        user = tx.execute(ts -> jdbc.queryForObject(SELECT_FROM_USER_BY_ID, new BeanPropertyRowMapper<>(User.class), 1));
        fail("Expected a EmptyResultDataAccessException");
    } catch (EmptyResultDataAccessException e) {
        // expected
    }
}

private void createBean(ConfigurableListableBeanFactory configurableListableBeanFactory,
    String prefixName, JdbcProperties jdbcProperties) {
  String jdbcUrl = jdbcProperties.getJdbcUrl();
  checkArgument(!Strings.isNullOrEmpty(jdbcUrl), prefixName + " url is null or empty");
  log.info("prefixName is {}, jdbc properties is {}", prefixName, jdbcProperties);

  HikariDataSource hikariDataSource = createHikariDataSource(jdbcProperties);
  DataSourceSpy dataSource = new DataSourceSpy(hikariDataSource);

  DataSourceTransactionManager transactionManager = new DataSourceTransactionManager(dataSource);
  AnnotationTransactionAspect.aspectOf().setTransactionManager(transactionManager);

  JdbcTemplate jdbcTemplate = new JdbcTemplate(dataSource);

  register(configurableListableBeanFactory, dataSource, prefixName + "DataSource",
      prefixName + "Ds");
  register(configurableListableBeanFactory, jdbcTemplate, prefixName + "JdbcTemplate",
      prefixName + "Jt");
  register(configurableListableBeanFactory, transactionManager, prefixName + "TransactionManager",
      prefixName + "Tx");
}

public DataBaseTransactionLogWritterImpl(ObjectSerializer objectSerializer,DataSource dataSource) {
    super();
    this.objectSerializer = objectSerializer;

    this.dataSource = dataSource;
    transactionManager = new DataSourceTransactionManager(dataSource);
    transactionTemplate = new TransactionTemplate(transactionManager, new DefaultTransactionDefinition(TransactionDefinition.PROPAGATION_REQUIRES_NEW));

}

/**
 * 配置事物管理器
 *
 * @return
 */
@Bean(name = "read2TransactionManager")
public DataSourceTransactionManager read2TransactionManager(
        @Qualifier("read2DataSource") DataSource dataSource
) {
    DataSourceTransactionManager dataSourceTransactionManager = new DataSourceTransactionManager();
    dataSourceTransactionManager.setDataSource(dataSource);
    return dataSourceTransactionManager;
}

/**
 * 配置事物管理器
 *
 * @return
 */
@Bean(name = "writeTransactionManager")
public DataSourceTransactionManager writeTransactionManager(
        @Qualifier("writeDataSource") DataSource dataSource
) {
    DataSourceTransactionManager dataSourceTransactionManager = new DataSourceTransactionManager();
    dataSourceTransactionManager.setDataSource(dataSource);
    return dataSourceTransactionManager;
}

/**
 * 配置事物管理器
 *
 * @return
 */
@Bean(name = "read1TransactionManager")
public DataSourceTransactionManager read1TransactionManager(
        @Qualifier("read1DataSource") DataSource dataSource
) {
    DataSourceTransactionManager dataSourceTransactionManager = new DataSourceTransactionManager();
    dataSourceTransactionManager.setDataSource(dataSource);
    return dataSourceTransactionManager;
}

/**
 * 配置事物管理器
 *
 * @return
 */
@Bean(name = "masterTransactionManager")
@Primary
public DataSourceTransactionManager masterTransactionManager(
        @Qualifier("masterDataSource") DataSource dataSource
) {
    DataSourceTransactionManager dataSourceTransactionManager = new DataSourceTransactionManager();
    dataSourceTransactionManager.setDataSource(dataSource);
    return dataSourceTransactionManager;
}

/**
 * 配置事物管理器
 *
 * @return
 */
@Bean(name = "clusterTransactionManager")
public DataSourceTransactionManager clusterTransactionManager(
        @Qualifier("clusterDataSource") DataSource dataSource
) {
    DataSourceTransactionManager dataSourceTransactionManager = new DataSourceTransactionManager();
    dataSourceTransactionManager.setDataSource(dataSource);
    return dataSourceTransactionManager;
}

/**
 * 配置事物管理器
 *
 * @return
 */
@Bean(name = "cluster1TransactionManager")
public DataSourceTransactionManager cluster1TransactionManager(
        @Qualifier("cluster1DataSource") DataSource dataSource
) {
    DataSourceTransactionManager dataSourceTransactionManager = new DataSourceTransactionManager();
    dataSourceTransactionManager.setDataSource(dataSource);
    return dataSourceTransactionManager;
}

/**
 * 配置事物管理器
 *
 * @return
 */
@Bean
public DataSourceTransactionManager transactionManager() {
    DataSourceTransactionManager dataSourceTransactionManager = new DataSourceTransactionManager();
    dataSourceTransactionManager.setDataSource(dataSource);
    return dataSourceTransactionManager;
}

@Bean
@Override
public PlatformTransactionManager annotationDrivenTransactionManager() {
  try {
    return new DataSourceTransactionManager(datasource());
  } catch (SQLException e) {
    e.printStackTrace();
  }
  return null;
}

public AbstractDbDialect(JdbcTemplate jdbcTemplate, LobHandler lobHandler, String name, int majorVersion,
                         int minorVersion){
    this.jdbcTemplate = jdbcTemplate;
    this.lobHandler = lobHandler;
    // 初始化transction
    this.transactionTemplate = new TransactionTemplate();
    transactionTemplate.setTransactionManager(new DataSourceTransactionManager(jdbcTemplate.getDataSource()));
    transactionTemplate.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRES_NEW);

    this.databaseName = name;
    this.databaseMajorVersion = majorVersion;
    this.databaseMinorVersion = minorVersion;

    initTables(jdbcTemplate);
}

PGSynchronizedQuery(@NonNull JdbcTemplate jdbcTemplate, @NonNull String sql,
        @NonNull PreparedStatementSetter setter, @NonNull RowCallbackHandler rowHandler,
        AtomicLong serialToContinueFrom, PGLatestSerialFetcher fetcher) {
    this.serialToContinueFrom = serialToContinueFrom;
    latestFetcher = fetcher;
    this.jdbcTemplate = jdbcTemplate;
    this.sql = sql;
    this.setter = setter;
    this.rowHandler = rowHandler;

    DataSourceTransactionManager transactionManager = new DataSourceTransactionManager(
            jdbcTemplate.getDataSource());
    transactionTemplate = new TransactionTemplate(transactionManager);
}

/**
 * puts the block into the database.
 *
 * @param blocks
 *            the blocks to use.
 */
@Override
public void put(final Block... blocks) {
    synchronized (this) {
        if (closed) {
            return;
        }
    }
    if (LOG.isInfoEnabled()) {
        LOG.info("STARTED put, {} blocks", NumberFormat.getIntegerInstance().format(blocks.length));
    }
    final DataSourceTransactionManager tsMan = new DataSourceTransactionManager(ds);

    final TransactionTemplate txTemplate = new TransactionTemplate(tsMan);
    // set behavior
    txTemplate.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRES_NEW);
    try {
        txTemplate.execute(new TransactionCallback(blocks));
    } catch (final DataAccessException e) {
        if (LOG.isErrorEnabled()) {
            LOG.error("data access exception", e);
        }
    }
    if (LOG.isInfoEnabled()) {
        LOG.info("SUCCESS put, {} blocks", NumberFormat.getIntegerInstance().format(blocks.length));
    }
}

/**
 * 开启事物
 * @param transactionManager
 */
public void start(DataSourceTransactionManager transactionManager) {
  DefaultTransactionDefinition def = new DefaultTransactionDefinition();
  /*PROPAGATION_REQUIRES_NEW:  事物隔离级别，开启新事务*/
  def.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRES_NEW);
  status =  transactionManager.getTransaction(def);
}

@Bean
public DataSourceTransactionManager wholeTransactionManager(DataSource whole){
    return new DataSourceTransactionManager(whole);
}

@Bean
public DataSourceTransactionManager buySthTransactionManager(DataSource buySth){
    return new DataSourceTransactionManager(buySth);
}

@Bean
public DataSourceTransactionManager payTransactionManager(DataSource pay){
    return new DataSourceTransactionManager(pay);
}

@Bean
public PlatformTransactionManager annotationDrivenTransactionManager() {
    return new DataSourceTransactionManager((DataSource) SpringContextUtils.getBean("roundRobinDataSouceProxy"));
}

@Bean
public DataSourceTransactionManager ReliableOrderMsgTransactionManager(DataSource ReliableOrderMsg){
    return new DataSourceTransactionManager(ReliableOrderMsg);
}

@Bean
public PlatformTransactionManager transactionManager() {
    return new DataSourceTransactionManager(dataSource());
}

@Bean
public DataSourceTransactionManager noticeExpressTransactionManager(DataSource noticeExpress){
    return new DataSourceTransactionManager(noticeExpress);
}

@Bean(name="transactionManager")
PlatformTransactionManager transactionManager(){
    return new DataSourceTransactionManager(dataSource());
}

@Bean
public DataSourceTransactionManager getTransactionManager() {
    DataSourceTransactionManager txManager = new DataSourceTransactionManager();
    txManager.setDataSource(getDataSource());
    return txManager;
}

@Bean
public DataSourceTransactionManager transactionManager(DataSource ds) {
    return new DataSourceTransactionManager(ds);
}

public static ObjectNode getPagedDatasetComments(String userName, int id, int page, int size)
{
    ObjectNode result = Json.newObject();

    javax.sql.DataSource ds = getJdbcTemplate().getDataSource();
    DataSourceTransactionManager tm = new DataSourceTransactionManager(ds);
    TransactionTemplate txTemplate = new TransactionTemplate(tm);

    result = txTemplate.execute(new TransactionCallback<ObjectNode>() {
        public ObjectNode doInTransaction(TransactionStatus status) {

            List<DatasetComment> pagedComments = getJdbcTemplate().query(
                        GET_COMMENTS_BY_DATASET_ID,
                        new DatasetCommentRowMapper(),
                        id,
                        (page - 1) * size, size);

            long count = 0;
            try {
                count = getJdbcTemplate().queryForObject(
                        "SELECT FOUND_ROWS()",
                        Long.class);
            } catch (EmptyResultDataAccessException e) {
                Logger.error("Exception = " + e.getMessage());
            }

            if (pagedComments != null)
            {
                for(DatasetComment dc : pagedComments)
                {
                    if(StringUtils.isNotBlank(userName) && userName.equalsIgnoreCase(dc.authorUserName))
                    {
                        dc.isAuthor = true;
                    }
                }
            }

            ObjectNode resultNode = Json.newObject();
            resultNode.set("comments", Json.toJson(pagedComments));
            resultNode.put("count", count);
            resultNode.put("page", page);
            resultNode.put("itemsPerPage", size);
            resultNode.put("totalPages", (int) Math.ceil(count / ((double) size)));

            return resultNode;
        }
    });
    return result;
}

public static ObjectNode getPagedMetricByKeyword(final String category, String keyword, int page, int size)
{
    List<Metric> pagedMetrics = new ArrayList<Metric>();
    final JdbcTemplate jdbcTemplate = getJdbcTemplate();
    javax.sql.DataSource ds = jdbcTemplate.getDataSource();
    DataSourceTransactionManager tm = new DataSourceTransactionManager(ds);

    TransactionTemplate txTemplate = new TransactionTemplate(tm);

    ObjectNode result;
    result = txTemplate.execute(new TransactionCallback<ObjectNode>()
    {
        public ObjectNode doInTransaction(TransactionStatus status)
        {
            String query = SEARCH_METRIC_WITH_PAGINATION.replace("$keyword", keyword);
            List<Map<String, Object>> rows = null;
            rows = jdbcTemplate.queryForList(query, (page-1)*size, size);
            for (Map row : rows) {

                Metric metric = new Metric();
                metric.id = (Integer)row.get(MetricRowMapper.METRIC_ID_COLUMN);
                metric.name = (String)row.get(MetricRowMapper.METRIC_NAME_COLUMN);
                metric.refID = (String)row.get(MetricRowMapper.METRIC_REF_ID_COLUMN);
                metric.refIDType = (String)row.get(MetricRowMapper.METRIC_REF_ID_TYPE_COLUMN);
                metric.description = (String)row.get(MetricRowMapper.METRIC_DESCRIPTION_COLUMN);
                metric.dashboardName = (String)row.get(MetricRowMapper.METRIC_DASHBOARD_NAME_COLUMN);
                metric.category = (String)row.get(MetricRowMapper.METRIC_CATEGORY_COLUMN);
                metric.group = (String)row.get(MetricRowMapper.METRIC_GROUP_COLUMN);
                metric.source = "metric";
                metric.urn = "";
                if (StringUtils.isNotBlank(metric.dashboardName))
                {
                    metric.urn += metric.dashboardName + "/";
                }
                if (StringUtils.isNotBlank(metric.group))
                {
                    metric.urn += metric.group + "/";
                }
                if (StringUtils.isNotBlank(metric.name))
                {
                    metric.urn += metric.name;
                }

                ObjectNode schema = Json.newObject();
                schema.put(MetricRowMapper.METRIC_REF_ID_COLUMN, metric.refID);
                schema.put(MetricRowMapper.METRIC_REF_ID_TYPE_COLUMN, metric.refIDType);
                schema.put(MetricRowMapper.METRIC_DESCRIPTION_COLUMN, metric.description);
                schema.put(MetricRowMapper.METRIC_DASHBOARD_NAME_COLUMN, metric.dashboardName);
                schema.put(MetricRowMapper.METRIC_CATEGORY_COLUMN, metric.category);
                schema.put(MetricRowMapper.METRIC_GROUP_COLUMN, metric.group);
                metric.schema = schema.toString();
                pagedMetrics.add(metric);
            }
            long count = 0;
            try {
                count = jdbcTemplate.queryForObject(
                        "SELECT FOUND_ROWS()",
                        Long.class);
            }
            catch(EmptyResultDataAccessException e)
            {
                Logger.error("Exception = " + e.getMessage());
            }

            ObjectNode resultNode = Json.newObject();
            resultNode.put("count", count);
            resultNode.put("page", page);
            resultNode.put("category", category);
            resultNode.put("itemsPerPage", size);
            resultNode.put("totalPages", (int)Math.ceil(count/((double)size)));
            resultNode.set("data", Json.toJson(pagedMetrics));

            return resultNode;
        }
    });

    return result;
}

public static ObjectNode getPagedFlowByKeyword(String category, String keyword, int page, int size)
{
    final List<FlowJob> pagedFlows = new ArrayList<FlowJob>();
    final JdbcTemplate jdbcTemplate = getJdbcTemplate();
    javax.sql.DataSource ds = jdbcTemplate.getDataSource();
    DataSourceTransactionManager tm = new DataSourceTransactionManager(ds);

    TransactionTemplate txTemplate = new TransactionTemplate(tm);

    ObjectNode result;
    result = txTemplate.execute(new TransactionCallback<ObjectNode>()
    {
        public ObjectNode doInTransaction(TransactionStatus status)
        {
            String query = SEARCH_FLOW_WITH_PAGINATION.replace("$keyword", keyword);
            List<Map<String, Object>> rows = null;

            rows = jdbcTemplate.queryForList(query, (page-1)*size, size);
            for (Map row : rows) {

                FlowJob flow = new FlowJob();
                flow.flowId = (Long)row.get(FlowRowMapper.FLOW_ID_COLUMN);
                flow.flowName = (String)row.get(FlowRowMapper.FLOW_NAME_COLUMN);
                flow.flowPath = (String)row.get(FlowRowMapper.FLOW_PATH_COLUMN);
                flow.flowGroup = (String)row.get(FlowRowMapper.FLOW_GROUP_COLUMN);
                flow.appCode = (String)row.get(FlowRowMapper.APP_CODE_COLUMN);
                flow.appId = (Integer)row.get(FlowRowMapper.APP_ID_COLUMN);
                flow.displayName = flow.flowName;
                flow.link = "#/flows/name/" + flow.appCode + "/" +
                        Long.toString(flow.flowId) + "/page/1?urn=" + flow.flowGroup;
                flow.path = flow.appCode + "/" + flow.flowPath;
                pagedFlows.add(flow);
            }
            long count = 0;
            try {
                count = jdbcTemplate.queryForObject(
                        "SELECT FOUND_ROWS()",
                        Long.class);
            }
            catch(EmptyResultDataAccessException e)
            {
                Logger.error("Exception = " + e.getMessage());
            }

            ObjectNode resultNode = Json.newObject();
            resultNode.put("count", count);
            resultNode.put("isFlowJob", true);
            resultNode.put("page", page);
            resultNode.put("category", category);
            resultNode.put("itemsPerPage", size);
            resultNode.put("totalPages", (int)Math.ceil(count/((double)size)));
            resultNode.set("data", Json.toJson(pagedFlows));

            return resultNode;
        }
    });

    return result;
}

@Bean(name = "txManagerTest")
public PlatformTransactionManager txManagerTest() {
    return new DataSourceTransactionManager(dataSourceTest());
}

@Bean(name = {ConfigConstant.NAME_DS_TM_SLAVE})
public DataSourceTransactionManager mafSlaveTransactionManager() {
    logger.info("----- MAFIA slave data source transaction manager INIT -----");
    return new DataSourceTransactionManager(mafSlaveDataSource());
}

public static TransactionTemplate getTransactionTemplate() {
    PlatformTransactionManager txManager = new DataSourceTransactionManager(
            DataSource.getDataSource(MYSQL_DB_WHEREHOWS_OPENSOURCE));
    return new TransactionTemplate(txManager);
}

@Bean
public PlatformTransactionManager transactionManager() {
    return new DataSourceTransactionManager(dataSource());
}

@Bean
public PlatformTransactionManager transactionManager() {
    return new DataSourceTransactionManager(selectAnyDataSource);
}

@Bean
public PlatformTransactionManager transactionManager(DataSource dataSource) {
    return new DataSourceTransactionManager(dataSource);
}

@Bean
public PlatformTransactionManager inspektrAuditTransactionManager() {
    return new DataSourceTransactionManager(inspektrAuditTrailDataSource());
}

/**
 * 初始化事务对象并开启事务
 * @param transactionManager
 */
public Transactioner(DataSourceTransactionManager transactionManager) {
  this.transactionManager = transactionManager;
  start(transactionManager);
}

@Bean
public PlatformTransactionManager transactionManager() {
    return new DataSourceTransactionManager(dataSource());
}

@Bean(name = "userTransactionManager")
@Primary
public DataSourceTransactionManager userTransactionManager() {
    return new DataSourceTransactionManager(userDataSource());
}