新增动态数据源

hsweb-datasource/hsweb-datasource-api/pom.xml

@@ -0,0 +1,30 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <parent>
+        <artifactId>hsweb-datasource</artifactId>
+        <groupId>org.hswebframework.web</groupId>
+        <version>3.0-SNAPSHOT</version>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>hsweb-datasource-api</artifactId>
+
+
+    <dependencies>
+        <dependency>
+            <groupId>org.hswebframework.web</groupId>
+            <artifactId>hsweb-commons-utils</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.slf4j</groupId>
+            <artifactId>slf4j-api</artifactId>
+        </dependency>
+    </dependencies>
+</project>

hsweb-datasource/hsweb-datasource-api/src/main/java/org/hswebframework/web/datasource/DataSourceHolder.java

@@ -0,0 +1,37 @@
+package org.hswebframework.web.datasource;
+
+import org.hswebframework.web.datasource.switcher.DataSourceSwitcher;
+
+/**
+ * TODO 完成注释
+ *
+ * @author zhouhao
+ */
+public final class DataSourceHolder {
+
+    static DataSourceSwitcher dataSourceSwitcher;
+
+    static DynamicDataSourceService dynamicDataSourceService;
+
+    public static DataSourceSwitcher switcher() {
+        return dataSourceSwitcher;
+    }
+
+    public static DynamicDataSource getDefaultDataSource() {
+        return dynamicDataSourceService.getDefaultDataSource();
+    }
+
+    public static DynamicDataSource getActiveDataSource() {
+        String id = dataSourceSwitcher.currentDataSourceId();
+        if (id == null) return getDefaultDataSource();
+        return dynamicDataSourceService.getDataSource(id);
+    }
+
+    public static DatabaseType getActiveDatabaseType() {
+        return getActiveDataSource().getType();
+    }
+
+    public static DatabaseType getDefaultDatabaseType() {
+        return getDefaultDataSource().getType();
+    }
+}

hsweb-datasource/hsweb-datasource-api/src/main/java/org/hswebframework/web/datasource/DatabaseType.java

@@ -0,0 +1,73 @@
+/*
+ * Copyright 2015-2016 http://hsweb.me
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.hswebframework.web.datasource;
+
+import java.util.Arrays;
+import java.util.Objects;
+import java.util.function.Predicate;
+
+public enum DatabaseType {
+    unknown(null, null, null, String::isEmpty),
+    mysql("com.mysql.jdbc.Driver", "com.mysql.jdbc.jdbc2.optional.MysqlXADataSource", "select 1", createUrlPredicate("mysql")),
+    h2("org.h2.Driver", "org.h2.jdbcx.JdbcDataSource", "select 1", createUrlPredicate("h2")),
+    oracle("oracle.jdbc.driver.OracleDriver", "oracle.jdbc.xa.client.OracleXADataSource", "select 1 from dual", createUrlPredicate("oracle")),
+    jtds_sqlserver("net.sourceforge.jtds.jdbc.Driver", "net.sourceforge.jtds.jdbcx.JtdsDataSource", "select 1 t", createUrlPredicate("jtds:sqlserver")),
+    sqlserver("com.microsoft.sqlserver.jdbc.SQLServerDriver", "com.microsoft.sqlserver.jdbc.SQLServerXADataSource", "select 1 t", createUrlPredicate("sqlserver"));
+
+    static Predicate<String> createUrlPredicate(String name) {
+        return url -> {
+            String urlWithoutPrefix = url.substring("jdbc".length()).toLowerCase();
+            String prefix = ":" + name.toLowerCase() + ":";
+            return urlWithoutPrefix.startsWith(prefix);
+        };
+    }
+
+    DatabaseType(String driverClassName, String xaDataSourceClassName, String testQuery, Predicate<String> urlPredicate) {
+        this.driverClassName = driverClassName;
+        this.testQuery = testQuery;
+        this.xaDataSourceClassName = xaDataSourceClassName;
+        this.urlPredicate = urlPredicate;
+    }
+
+
+    private final String testQuery;
+
+    private final String driverClassName;
+
+    private final String xaDataSourceClassName;
+
+    private final Predicate<String> urlPredicate;
+
+    public String getDriverClassName() {
+        return driverClassName;
+    }
+
+    public String getXaDataSourceClassName() {
+        return xaDataSourceClassName;
+    }
+
+    public String getTestQuery() {
+        return testQuery;
+    }
+
+    public static DatabaseType fromJdbcUrl(String url) {
+        if (Objects.nonNull(url)) {
+            return Arrays.stream(values()).filter(type -> type.urlPredicate.test(url)).findFirst().orElse(unknown);
+        }
+        return unknown;
+    }
+}

hsweb-datasource/hsweb-datasource-api/src/main/java/org/hswebframework/web/datasource/DynamicDataSource.java

@@ -0,0 +1,15 @@
+package org.hswebframework.web.datasource;
+
+import javax.sql.DataSource;
+
+/**
+ * TODO 完成注释
+ *
+ * @author zhouhao
+ */
+public interface DynamicDataSource extends DataSource {
+    String getId();
+
+    DatabaseType getType();
+
+}

hsweb-datasource/hsweb-datasource-api/src/main/java/org/hswebframework/web/datasource/DynamicDataSourceAutoConfiguration.java

@@ -0,0 +1,41 @@
+package org.hswebframework.web.datasource;
+
+import org.hswebframework.web.datasource.switcher.DataSourceSwitcher;
+import org.hswebframework.web.datasource.switcher.DefaultDataSourceSwitcher;
+import org.springframework.beans.BeansException;
+import org.springframework.beans.factory.config.BeanPostProcessor;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+/**
+ * TODO 完成注释
+ *
+ * @author zhouhao
+ */
+@Configuration
+public class DynamicDataSourceAutoConfiguration implements BeanPostProcessor {
+
+    @Bean
+    @ConditionalOnMissingBean(DataSourceSwitcher.class)
+    public DataSourceSwitcher dataSourceSwitcher() {
+        return new DefaultDataSourceSwitcher();
+    }
+
+    @Override
+    public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException {
+        return bean;
+    }
+
+    @Override
+    public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
+        if (bean instanceof DynamicDataSourceService) {
+            DataSourceHolder.dynamicDataSourceService = ((DynamicDataSourceService) bean);
+        }
+        if (bean instanceof DataSourceSwitcher) {
+            DataSourceHolder.dataSourceSwitcher = ((DataSourceSwitcher) bean);
+        }
+        return bean;
+    }
+}

hsweb-datasource/hsweb-datasource-api/src/main/java/org/hswebframework/web/datasource/DynamicDataSourceProxy.java

@@ -0,0 +1,102 @@
+package org.hswebframework.web.datasource;
+
+import javax.sql.DataSource;
+import java.io.PrintWriter;
+import java.sql.Connection;
+import java.sql.SQLException;
+import java.sql.SQLFeatureNotSupportedException;
+import java.util.logging.Logger;
+
+/**
+ * TODO 完成注释
+ *
+ * @author zhouhao
+ */
+public class DynamicDataSourceProxy implements DynamicDataSource {
+
+    private String id;
+
+    private DatabaseType databaseType;
+
+    private DataSource proxy;
+
+    public DynamicDataSourceProxy(String id, DatabaseType databaseType, DataSource proxy) {
+        this.id = id;
+        this.databaseType = databaseType;
+        this.proxy = proxy;
+    }
+
+    public DynamicDataSourceProxy(String id, DataSource proxy) throws SQLException {
+        this.id = id;
+        this.proxy = proxy;
+    }
+
+    @Override
+    public String getId() {
+        return id;
+    }
+
+    @Override
+    public DatabaseType getType() {
+        if (databaseType == null) {
+            synchronized (this) {
+                if (databaseType == null) {
+                    try {
+                        try (Connection connection = proxy.getConnection()) {
+                            databaseType = DatabaseType.fromJdbcUrl(connection.getMetaData().getURL());
+                        }
+                    } catch (SQLException e) {
+                        e.printStackTrace();
+                    }
+                }
+            }
+        }
+
+        return databaseType;
+    }
+
+    @Override
+    public Connection getConnection() throws SQLException {
+        return proxy.getConnection();
+    }
+
+    @Override
+    public Connection getConnection(String username, String password) throws SQLException {
+        return proxy.getConnection(username, password);
+    }
+
+    @Override
+    public <T> T unwrap(Class<T> iface) throws SQLException {
+        return proxy.unwrap(iface);
+    }
+
+    @Override
+    public boolean isWrapperFor(Class<?> iface) throws SQLException {
+        return proxy.isWrapperFor(iface);
+    }
+
+    @Override
+    public PrintWriter getLogWriter() throws SQLException {
+        return proxy.getLogWriter();
+    }
+
+    @Override
+    public void setLogWriter(PrintWriter out) throws SQLException {
+        proxy.setLogWriter(out);
+    }
+
+    @Override
+    public void setLoginTimeout(int seconds) throws SQLException {
+        proxy.setLoginTimeout(seconds);
+    }
+
+    @Override
+    public int getLoginTimeout() throws SQLException {
+        return proxy.getLoginTimeout();
+    }
+
+    @Override
+    public Logger getParentLogger() throws SQLFeatureNotSupportedException {
+        return proxy.getParentLogger();
+    }
+}

hsweb-datasource/hsweb-datasource-api/src/main/java/org/hswebframework/web/datasource/DynamicDataSourceService.java

@@ -0,0 +1,12 @@
+package org.hswebframework.web.datasource;
+
+/**
+ * TODO 完成注释
+ *
+ * @author zhouhao
+ */
+public interface DynamicDataSourceService {
+    DynamicDataSource getDataSource(String dataSourceId);
+
+    DynamicDataSource getDefaultDataSource();
+}

hsweb-datasource/hsweb-datasource-api/src/main/java/org/hswebframework/web/datasource/annotation/UseDataSource.java

@@ -0,0 +1,14 @@
+package org.hswebframework.web.datasource.annotation;
+
+import java.lang.annotation.*;
+
+/**
+ * @author zhouhao
+ */
+@Target({ElementType.TYPE, ElementType.METHOD})
+@Retention(RetentionPolicy.RUNTIME)
+@Documented
+@Inherited
+public @interface UseDataSource {
+    String value();
+}

hsweb-datasource/hsweb-datasource-api/src/main/java/org/hswebframework/web/datasource/annotation/UseDefaultDataSource.java

@@ -0,0 +1,13 @@
+package org.hswebframework.web.datasource.annotation;
+
+import java.lang.annotation.*;
+
+/**
+ * @author zhouhao
+ */
+@Target({ElementType.TYPE, ElementType.METHOD})
+@Retention(RetentionPolicy.RUNTIME)
+@Documented
+@Inherited
+public @interface UseDefaultDataSource {
+}

hsweb-datasource/hsweb-datasource-api/src/main/java/org/hswebframework/web/datasource/exception/DataSourceClosedException.java

@@ -0,0 +1,25 @@
+package org.hswebframework.web.datasource.exception;
+
+/**
+ * TODO 完成注释
+ *
+ * @author zhouhao
+ */
+public class DataSourceClosedException extends RuntimeException {
+
+    private String dataSourceId;
+
+    public DataSourceClosedException(String dataSourceId) {
+        this(dataSourceId, dataSourceId);
+    }
+
+    public DataSourceClosedException(String dataSourceId, String message) {
+        super(message);
+        this.dataSourceId = dataSourceId;
+    }
+
+    public String getDataSourceId() {
+        return dataSourceId;
+    }
+
+}

hsweb-datasource/hsweb-datasource-api/src/main/java/org/hswebframework/web/datasource/exception/DataSourceNotFoundException.java

@@ -0,0 +1,25 @@
+package org.hswebframework.web.datasource.exception;
+
+/**
+ * TODO 完成注释
+ *
+ * @author zhouhao
+ */
+public class DataSourceNotFoundException extends RuntimeException {
+
+    private String dataSourceId;
+
+    public DataSourceNotFoundException(String dataSourceId) {
+        this(dataSourceId, dataSourceId);
+    }
+
+    public DataSourceNotFoundException(String dataSourceId, String message) {
+        super(message);
+        this.dataSourceId = dataSourceId;
+    }
+
+    public String getDataSourceId() {
+        return dataSourceId;
+    }
+
+}

hsweb-datasource/hsweb-datasource-api/src/main/java/org/hswebframework/web/datasource/service/AbstractDynamicDataSourceService.java

@@ -0,0 +1,57 @@
+package org.hswebframework.web.datasource.service;
+
+import org.hswebframework.web.datasource.DynamicDataSource;
+import org.hswebframework.web.datasource.DynamicDataSourceProxy;
+import org.hswebframework.web.datasource.DynamicDataSourceService;
+
+import javax.sql.DataSource;
+import java.sql.SQLException;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * @author zhouhao
+ */
+public abstract class AbstractDynamicDataSourceService implements DynamicDataSourceService {
+    protected final Map<String, DataSourceCache> dataSourceStore = new ConcurrentHashMap<>(32);
+
+    private final DynamicDataSource defaultDataSource;
+
+    public AbstractDynamicDataSourceService(DynamicDataSource defaultDataSource) {
+        this.defaultDataSource = defaultDataSource;
+    }
+
+    public AbstractDynamicDataSourceService(DataSource dataSource) throws SQLException {
+        this(dataSource instanceof DynamicDataSource ? (DynamicDataSource) dataSource : new DynamicDataSourceProxy(null, dataSource));
+    }
+
+    @Override
+    public DynamicDataSource getDataSource(String dataSourceId) {
+        DataSourceCache cache = dataSourceStore.get(dataSourceId);
+        if (cache == null) {
+            cache = createCache(dataSourceId);
+            dataSourceStore.put(dataSourceId, cache);
+            return cache.getDataSource();
+        }
+        if (cache.getHash() != getHash(dataSourceId)) {
+            dataSourceStore.remove(dataSourceId);
+            cache.closeDataSource();
+            //重新获取
+            return getDataSource(dataSourceId);
+        }
+        return cache.getDataSource();
+    }
+
+    @Override
+    public DynamicDataSource getDefaultDataSource() {
+        return defaultDataSource;
+    }
+
+    protected abstract int getHash(String id);
+
+    protected abstract DataSourceCache createCache(String id);
+
+    public DataSourceCache removeCache(String id) {
+        return dataSourceStore.remove(id);
+    }
+}

hsweb-datasource/hsweb-datasource-api/src/main/java/org/hswebframework/web/datasource/service/DataSourceCache.java

@@ -0,0 +1,56 @@
+package org.hswebframework.web.datasource.service;
+
+import org.hswebframework.web.datasource.DynamicDataSource;
+import org.hswebframework.web.datasource.exception.DataSourceClosedException;
+
+import java.util.concurrent.CountDownLatch;
+
+/**
+ * 数据源缓存
+ *
+ * @author zhouhao
+ */
+public class DataSourceCache {
+    private long hash;
+
+    private volatile boolean closed;
+
+    private DynamicDataSource dataSource;
+
+    private volatile CountDownLatch initLatch;
+
+    public long getHash() {
+        return hash;
+    }
+
+    public DynamicDataSource getDataSource() {
+        if (initLatch != null) {
+            try {
+                //等待初始化完成
+                initLatch.await();
+            } catch (InterruptedException ignored) {
+            } finally {
+                initLatch = null;
+            }
+        }
+        if (closed) {
+            throw new DataSourceClosedException(dataSource.getId());
+        }
+        return dataSource;
+    }
+
+    public DataSourceCache(long hash, DynamicDataSource dataSource, CountDownLatch initLatch) {
+        this.hash = hash;
+        this.dataSource = dataSource;
+        this.initLatch = initLatch;
+    }
+
+    public boolean isClosed() {
+        return closed;
+    }
+
+
+    public void closeDataSource() {
+        closed = true;
+    }
+}

hsweb-datasource/hsweb-datasource-api/src/main/java/org/hswebframework/web/datasource/switcher/DataSourceSwitcher.java

@@ -0,0 +1,36 @@
+package org.hswebframework.web.datasource.switcher;
+
+/**
+ * TODO 完成注释
+ *
+ * @author zhouhao
+ */
+public interface DataSourceSwitcher {
+
+    /**
+     * 使用上一次调用的数据源
+     */
+    void useLast();
+
+    /**
+     * 选中参数(数据源ID)对应的数据源,如果数据源不存在,将使用默认数据源
+     *
+     * @param dataSourceId 数据源ID
+     */
+    void use(String dataSourceId);
+
+    /**
+     * 切换为默认数据源
+     */
+    void useDefault();
+
+    /**
+     * @return 当前选择的数据源ID
+     */
+    String currentDataSourceId();
+
+    /**
+     * 重置切换记录
+     */
+    void reset();
+}

hsweb-datasource/hsweb-datasource-api/src/main/java/org/hswebframework/web/datasource/switcher/DefaultDataSourceSwitcher.java

@@ -0,0 +1,73 @@
+package org.hswebframework.web.datasource.switcher;
+
+import org.hswebframework.web.ThreadLocalUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Deque;
+import java.util.LinkedList;
+
+/**
+ * TODO 完成注释
+ *
+ * @author zhouhao
+ */
+public class DefaultDataSourceSwitcher implements DataSourceSwitcher {
+    private static final String DEFAULT_DATASOURCE_ID = DataSourceSwitcher.class.getName() + "_default_";
+
+    private Logger logger = LoggerFactory.getLogger(this.getClass());
+
+    private Deque<String> getUsedHistoryQueue() {
+        return ThreadLocalUtils.get(DefaultDataSourceSwitcher.class.getName() + "_queue", LinkedList::new);
+    }
+
+    @Override
+    public void useLast() {
+        if (getUsedHistoryQueue().size() == 0) {
+            return;
+        }
+        getUsedHistoryQueue().removeLast();
+
+        if (logger.isDebugEnabled()) {
+            String current = currentDataSourceId();
+            if (null != current)
+                logger.debug("try use last data source : {}", currentDataSourceId());
+            else logger.debug("try use default data source");
+        }
+    }
+
+    @Override
+    public void use(String dataSourceId) {
+        getUsedHistoryQueue().addLast(dataSourceId);
+        if (logger.isDebugEnabled()) {
+            logger.debug("try use data source : {}", dataSourceId);
+        }
+    }
+
+    @Override
+    public void useDefault() {
+        getUsedHistoryQueue().addLast(DEFAULT_DATASOURCE_ID);
+        if (logger.isDebugEnabled()) {
+            logger.debug("try use default data source");
+        }
+    }
+
+    @Override
+    public String currentDataSourceId() {
+        if (getUsedHistoryQueue().size() == 0) return null;
+
+        String activeId = getUsedHistoryQueue().getLast();
+        if (DEFAULT_DATASOURCE_ID.equals(activeId)) {
+            return null;
+        }
+        return activeId;
+    }
+
+    @Override
+    public void reset() {
+        getUsedHistoryQueue().clear();
+        if (logger.isDebugEnabled()) {
+            logger.debug("reset data source used history");
+        }
+    }
+}

hsweb-datasource/hsweb-datasource-api/src/main/resources/META-INF/spring.factories

@@ -0,0 +1,3 @@
+# Auto Configure
+org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
+org.hswebframework.web.datasource.DynamicDataSourceAutoConfiguration

hsweb-datasource/hsweb-datasource-api/src/test/java/org/hswebframework/web/datasource/switcher/DefaultDataSourceSwitcherTest.java

@@ -0,0 +1,44 @@
+package org.hswebframework.web.datasource.switcher;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+/**
+ * TODO 完成注释
+ *
+ * @author zhouhao
+ */
+public class DefaultDataSourceSwitcherTest {
+
+    DataSourceSwitcher switcher = new DefaultDataSourceSwitcher();
+
+    @Test
+    public void testChangeSwitcher() {
+
+        switcher.use("test");//切换为test
+        assertEquals(switcher.currentDataSourceId(), "test");
+        switcher.use("test2");//切换为test2
+        assertEquals(switcher.currentDataSourceId(), "test2");
+
+        switcher.useDefault();//切换默认数据源
+        assertTrue(switcher.currentDataSourceId() == null);
+
+        switcher.useLast(); //切换为上一次使用的数据源(test2)
+        assertEquals(switcher.currentDataSourceId(), "test2");
+
+        switcher.useLast(); //切换为上一次使用的数据源(test)
+        assertEquals(switcher.currentDataSourceId(), "test");
+
+        switcher.useLast(); //切换为上一次书用的数据源(无,默认为default)
+        assertTrue(switcher.currentDataSourceId() == null);
+
+
+        switcher.useLast();
+        assertTrue(switcher.currentDataSourceId() == null);
+
+    }
+
+}

hsweb-datasource/hsweb-datasource-jta/pom.xml

@@ -0,0 +1,78 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <parent>
+        <artifactId>hsweb-datasource</artifactId>
+        <groupId>org.hswebframework.web</groupId>
+        <version>3.0-SNAPSHOT</version>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>hsweb-datasource-jta</artifactId>
+
+    <dependencies>
+        <dependency>
+            <groupId>com.h2database</groupId>
+            <artifactId>h2</artifactId>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>com.alibaba</groupId>
+            <artifactId>druid</artifactId>
+            <version>1.0.26</version>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.springframework</groupId>
+            <artifactId>spring-jdbc</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>org.hswebframework</groupId>
+            <artifactId>hsweb-easy-orm-rdb</artifactId>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-activemq</artifactId>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>mysql</groupId>
+            <artifactId>mysql-connector-java</artifactId>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.springframework</groupId>
+            <artifactId>spring-jdbc</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>org.hswebframework.web</groupId>
+            <artifactId>hsweb-datasource-api</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-test</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-jta-atomikos</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>commons-beanutils</groupId>
+            <artifactId>commons-beanutils</artifactId>
+        </dependency>
+
+    </dependencies>
+</project>

hsweb-datasource/hsweb-datasource-jta/src/main/java/org/hswebframework/web/datasource/jta/AtomikosDataSourceConfig.java

@@ -0,0 +1,182 @@
+package org.hswebframework.web.datasource.jta;
+
+import com.atomikos.jdbc.AtomikosDataSourceBean;
+
+import java.sql.SQLException;
+import java.util.Properties;
+
+/**
+ * TODO 完成注释
+ *
+ * @author zhouhao
+ */
+public class AtomikosDataSourceConfig {
+    private int        minPoolSize             = 1;
+    private int        maxPoolSize             = 20;
+    private int        borrowConnectionTimeout = 30;
+    private int        reapTimeout             = 0;
+    private int        maxIdleTime             = 60;
+    private int        maintenanceInterval     = 60;
+    private int        defaultIsolationLevel   = -1;
+    private String     xaDataSourceClassName   = null;
+    private int        loginTimeout            = 0;
+    private String     testQuery               = null;
+    private int        maxLifetime             = 0;
+    private Properties xaProperties            = null;
+    private int        initTimeOut             = 10 * 1000;
+
+    @Override
+    public int hashCode() {
+        return toString().hashCode();
+    }
+
+    @Override
+    public boolean equals(Object obj) {
+        return obj instanceof AtomikosDataSourceConfig && hashCode() == obj.hashCode();
+    }
+
+    @Override
+    public String toString() {
+        return "AtomikosDataSourceConfig{" +
+                "minPoolSize=" + minPoolSize +
+                ", maxPoolSize=" + maxPoolSize +
+                ", borrowConnectionTimeout=" + borrowConnectionTimeout +
+                ", reapTimeout=" + reapTimeout +
+                ", maxIdleTime=" + maxIdleTime +
+                ", maintenanceInterval=" + maintenanceInterval +
+                ", defaultIsolationLevel=" + defaultIsolationLevel +
+                ", xaDataSourceClassName='" + xaDataSourceClassName + '\'' +
+                ", loginTimeout=" + loginTimeout +
+                ", testQuery='" + testQuery + '\'' +
+                ", maxLifetime=" + maxLifetime +
+                ", xaProperties=" + xaProperties +
+                ", initTimeOut=" + initTimeOut +
+                '}';
+    }
+
+    public Properties getXaProperties() {
+        return xaProperties;
+    }
+
+    public void setXaProperties(Properties xaProperties) {
+        this.xaProperties = xaProperties;
+    }
+
+    public int getMinPoolSize() {
+        return minPoolSize;
+    }
+
+    public void setMinPoolSize(int minPoolSize) {
+        this.minPoolSize = minPoolSize;
+    }
+
+    public int getMaxPoolSize() {
+        return maxPoolSize;
+    }
+
+    public void setMaxPoolSize(int maxPoolSize) {
+        this.maxPoolSize = maxPoolSize;
+    }
+
+    public int getBorrowConnectionTimeout() {
+        return borrowConnectionTimeout;
+    }
+
+    public void setBorrowConnectionTimeout(int borrowConnectionTimeout) {
+        this.borrowConnectionTimeout = borrowConnectionTimeout;
+    }
+
+    public int getReapTimeout() {
+        return reapTimeout;
+    }
+
+    public void setReapTimeout(int reapTimeout) {
+        this.reapTimeout = reapTimeout;
+    }
+
+    public int getMaxIdleTime() {
+        return maxIdleTime;
+    }
+
+    public void setMaxIdleTime(int maxIdleTime) {
+        this.maxIdleTime = maxIdleTime;
+    }
+
+    public int getMaintenanceInterval() {
+        return maintenanceInterval;
+    }
+
+    public void setMaintenanceInterval(int maintenanceInterval) {
+        this.maintenanceInterval = maintenanceInterval;
+    }
+
+    public int getDefaultIsolationLevel() {
+        return defaultIsolationLevel;
+    }
+
+    public void setDefaultIsolationLevel(int defaultIsolationLevel) {
+        this.defaultIsolationLevel = defaultIsolationLevel;
+    }
+
+    public String getXaDataSourceClassName() {
+        return xaDataSourceClassName;
+    }
+
+    public void setXaDataSourceClassName(String xaDataSourceClassName) {
+        this.xaDataSourceClassName = xaDataSourceClassName;
+    }
+
+    public int getLoginTimeout() {
+        return loginTimeout;
+    }
+
+    public void setLoginTimeout(int loginTimeout) {
+        this.loginTimeout = loginTimeout;
+    }
+
+    public String getTestQuery() {
+        return testQuery;
+    }
+
+    public void setTestQuery(String testQuery) {
+        this.testQuery = testQuery;
+    }
+
+    public int getMaxLifetime() {
+        return maxLifetime;
+    }
+
+    public void setMaxLifetime(int maxLifetime) {
+        this.maxLifetime = maxLifetime;
+    }
+
+    public int getInitTimeOut() {
+        return initTimeOut;
+    }
+
+    public void setInitTimeOut(int initTimeOut) {
+        this.initTimeOut = initTimeOut;
+    }
+
+    public void putProperties(AtomikosDataSourceBean atomikosDataSourceBean) {
+        if (null != xaProperties) {
+            xaProperties.entrySet().forEach(entry -> entry.setValue(String.valueOf(entry.getValue())));
+        }
+        atomikosDataSourceBean.setXaDataSourceClassName(getXaDataSourceClassName());
+        atomikosDataSourceBean.setBorrowConnectionTimeout(getBorrowConnectionTimeout());
+        try {
+            atomikosDataSourceBean.setLoginTimeout(getLoginTimeout());
+        } catch (SQLException e) {
+            e.printStackTrace();
+        }
+        atomikosDataSourceBean.setMaxIdleTime(getMaxIdleTime());
+        atomikosDataSourceBean.setMaxPoolSize(getMaxPoolSize());
+        atomikosDataSourceBean.setMinPoolSize(getMinPoolSize());
+        atomikosDataSourceBean.setDefaultIsolationLevel(getDefaultIsolationLevel());
+        atomikosDataSourceBean.setMaintenanceInterval(getMaintenanceInterval());
+        atomikosDataSourceBean.setReapTimeout(getReapTimeout());
+        atomikosDataSourceBean.setTestQuery(getTestQuery());
+        atomikosDataSourceBean.setXaProperties(getXaProperties());
+        atomikosDataSourceBean.setMaxLifetime(getMaxLifetime());
+    }
+}

hsweb-datasource/hsweb-datasource-jta/src/main/java/org/hswebframework/web/datasource/jta/JtaDataSourceStore.java

@@ -0,0 +1,12 @@
+package org.hswebframework.web.datasource.jta;
+
+/**
+ * TODO 完成注释
+ *
+ * @author zhouhao
+ */
+public interface JtaDataSourceStore {
+
+    AtomikosDataSourceConfig getConfig(String id);
+
+}

hsweb-datasource/hsweb-datasource-jta/src/main/java/org/hswebframework/web/datasource/jta/JtaDynamicDataSourceService.java

@@ -0,0 +1,120 @@
+package org.hswebframework.web.datasource.jta;
+
+import org.hswebframework.web.datasource.DynamicDataSource;
+import org.hswebframework.web.datasource.DynamicDataSourceProxy;
+import org.hswebframework.web.datasource.exception.DataSourceNotFoundException;
+import org.hswebframework.web.datasource.service.AbstractDynamicDataSourceService;
+import org.hswebframework.web.datasource.service.DataSourceCache;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.jta.atomikos.AtomikosDataSourceBean;
+import org.springframework.jdbc.datasource.DataSourceUtils;
+
+import javax.sql.DataSource;
+import javax.sql.XADataSource;
+import java.io.Closeable;
+import java.io.IOException;
+import java.sql.SQLException;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Executor;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ForkJoinPool;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * TODO 完成注释
+ *
+ * @author zhouhao
+ */
+public class JtaDynamicDataSourceService extends AbstractDynamicDataSourceService {
+
+    private JtaDataSourceStore jtaDataSourceStore;
+
+    private Executor executor = Executors.newCachedThreadPool();
+
+    private Logger logger = LoggerFactory.getLogger(this.getClass());
+
+    @Autowired(required = false)
+    public void setExecutor(Executor executor) {
+        this.executor = executor;
+    }
+
+    public JtaDynamicDataSourceService(JtaDataSourceStore jtaDataSourceStore, DynamicDataSource defaultDataSource) {
+        super(defaultDataSource);
+        this.jtaDataSourceStore = jtaDataSourceStore;
+    }
+
+    public JtaDynamicDataSourceService(JtaDataSourceStore jtaDataSourceStore, DataSource dataSource) throws SQLException {
+        super(dataSource);
+        this.jtaDataSourceStore = jtaDataSourceStore;
+    }
+
+    @Override
+    protected int getHash(String id) {
+        AtomikosDataSourceConfig config = jtaDataSourceStore.getConfig(id);
+        if (null == config) return 0;
+        return config.hashCode();
+    }
+
+    @Override
+    protected DataSourceCache createCache(String id) {
+        AtomikosDataSourceConfig config = jtaDataSourceStore.getConfig(id);
+        if (config == null) {
+            throw new DataSourceNotFoundException(id);
+        }
+        AtomikosDataSourceBean atomikosDataSourceBean = new AtomikosDataSourceBean();
+        config.putProperties(atomikosDataSourceBean);
+        atomikosDataSourceBean.setBeanName("dynamic_ds_" + id);
+        atomikosDataSourceBean.setUniqueResourceName("dynamic_ds_" + id);
+        AtomicInteger successCounter = new AtomicInteger();
+        CountDownLatch downLatch = new CountDownLatch(1);
+        try {
+            DataSourceCache cache = new DataSourceCache(config.hashCode(), new DynamicDataSourceProxy(id, atomikosDataSourceBean), downLatch) {
+                @Override
+                public void closeDataSource() {
+                    super.closeDataSource();
+                    atomikosDataSourceBean.close();
+                    XADataSource dataSource = atomikosDataSourceBean.getXaDataSource();
+                    if (dataSource instanceof Closeable) {
+                        try {
+                            ((Closeable) dataSource).close();
+                        } catch (IOException e) {
+                            logger.error("close xa datasource error", e);
+                        }
+                    } else {
+                        logger.warn("XADataSource is not instanceof Closeable!", Thread.currentThread().getStackTrace());
+                    }
+                }
+            };
+            //异步初始化
+            executor.execute(() -> {
+                try {
+                    atomikosDataSourceBean.init();
+                    downLatch.countDown();
+                    successCounter.incrementAndGet();
+                } catch (Exception e) {
+                    //atomikosDataSourceBean.close();
+                }
+            });
+            //初始化状态判断
+            executor.execute(() -> {
+                try {
+                    Thread.sleep(config.getInitTimeOut());
+                } catch (InterruptedException ignored) {
+                } finally {
+                    if (successCounter.get() == 0) {
+                        // 初始化超时,认定为失败
+                        logger.error("init timeout ({}ms)", config.getInitTimeOut());
+                        cache.closeDataSource();
+                        if (downLatch.getCount() > 0)
+                            downLatch.countDown();
+                    }
+                }
+            });
+            return cache;
+        } catch (Exception e) {
+            throw new RuntimeException(e);
+        }
+    }
+}

hsweb-datasource/hsweb-datasource-jta/src/main/java/org/hswebframework/web/datasource/jta/MemoryJtaDataSourceStore.java

@@ -0,0 +1,29 @@
+package org.hswebframework.web.datasource.jta;
+
+import org.springframework.boot.context.properties.ConfigurationProperties;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * TODO 完成注释
+ *
+ * @author zhouhao
+ */
+@ConfigurationProperties(prefix = "hsweb.dynamic")
+public class MemoryJtaDataSourceStore implements JtaDataSourceStore {
+    private Map<String, AtomikosDataSourceConfig> datasource = new HashMap<>();
+
+    @Override
+    public AtomikosDataSourceConfig getConfig(String id) {
+        return datasource.get(id);
+    }
+
+    public Map<String, AtomikosDataSourceConfig> getDatasource() {
+        return datasource;
+    }
+
+    public void setDatasource(Map<String, AtomikosDataSourceConfig> datasource) {
+        this.datasource = datasource;
+    }
+}

hsweb-datasource/hsweb-datasource-jta/src/test/java/org/hswebframework/web/datasource/jta/SimpleAtomikosTests.java

@@ -0,0 +1,215 @@
+package org.hswebframework.web.datasource.jta;
+
+import com.alibaba.druid.pool.DruidDataSource;
+import org.hsweb.ezorm.rdb.RDBDatabase;
+import org.hsweb.ezorm.rdb.executor.AbstractJdbcSqlExecutor;
+import org.hsweb.ezorm.rdb.executor.SQL;
+import org.hsweb.ezorm.rdb.executor.SqlExecutor;
+import org.hsweb.ezorm.rdb.meta.RDBDatabaseMetaData;
+import org.hsweb.ezorm.rdb.render.SqlRender;
+import org.hsweb.ezorm.rdb.render.dialect.Dialect;
+import org.hsweb.ezorm.rdb.render.dialect.H2RDBDatabaseMetaData;
+import org.hsweb.ezorm.rdb.render.dialect.MysqlRDBDatabaseMetaData;
+import org.hsweb.ezorm.rdb.render.dialect.OracleRDBDatabaseMetaData;
+import org.hsweb.ezorm.rdb.simple.SimpleDatabase;
+import org.hswebframework.web.datasource.DataSourceHolder;
+import org.hswebframework.web.datasource.DatabaseType;
+import org.hswebframework.web.datasource.DynamicDataSourceProxy;
+import org.hswebframework.web.datasource.service.DataSourceCache;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.boot.jta.atomikos.AtomikosDataSourceBean;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.jdbc.datasource.DataSourceUtils;
+import org.springframework.jms.annotation.EnableJms;
+import org.springframework.jms.core.JmsTemplate;
+import org.springframework.test.annotation.Rollback;
+import org.springframework.test.context.junit4.SpringRunner;
+import org.springframework.transaction.annotation.Propagation;
+import org.springframework.transaction.annotation.Transactional;
+
+import javax.sql.DataSource;
+import java.sql.Connection;
+import java.sql.SQLException;
+import java.util.*;
+import java.util.stream.Collectors;
+
+/**
+ * TODO 完成注释
+ *
+ * @author zhouhao
+ */
+@SpringBootTest(properties = "application.yml", classes = SimpleAtomikosTests.Config.class)
+@RunWith(SpringRunner.class)
+public class SimpleAtomikosTests {
+
+    @Configuration
+    @SpringBootApplication
+    @EnableJms
+    public static class Config {
+        @Bean
+        public SqlExecutor sqlExecutor() {
+            return new AbstractJdbcSqlExecutor() {
+
+                @Override
+                @Transactional(propagation = Propagation.NOT_SUPPORTED)
+                public void exec(SQL sql) throws SQLException {
+                    super.exec(sql);
+                }
+
+                @Override
+                public Connection getConnection() {
+                    return DataSourceUtils.getConnection(DataSourceHolder.getActiveDataSource());
+                }
+
+                @Override
+                public void releaseConnection(Connection connection) throws SQLException {
+                    DataSourceUtils.releaseConnection(connection, DataSourceHolder.getActiveDataSource());
+                }
+            };
+        }
+
+        @Bean(initMethod = "init", destroyMethod = "destroy")
+        public AtomikosDataSourceBean atomikosDataSourceBean() {
+            AtomikosDataSourceBean dataSourceBean = new AtomikosDataSourceBean();
+            return dataSourceBean;
+        }
+
+        @Bean
+        public MemoryJtaDataSourceStore memoryJtaDataSourceStore() {
+            return new MemoryJtaDataSourceStore();
+        }
+
+        @Bean
+        public MemoryJtaDataSourceStore jtaDynamicDataSourceService() {
+            return new MemoryJtaDataSourceStore();
+        }
+
+        @Bean
+        public JtaDynamicDataSourceService jtaDynamicDataSourceService(JtaDataSourceStore jtaDataSourceStore, DataSource dataSource) throws SQLException {
+            return new JtaDynamicDataSourceService(jtaDataSourceStore, dataSource);
+        }
+
+        @Bean
+        public DynDsTest transTest(SqlExecutor sqlExecutor) {
+            return new DynDsTest(new SimpleDatabase(new DynDatabaseMeta(), sqlExecutor));
+        }
+
+        public class DynDatabaseMeta extends RDBDatabaseMetaData {
+            private Map<DatabaseType, Dialect>             dialectMap;
+            private Map<DatabaseType, RDBDatabaseMetaData> metaDataMap;
+
+            public DynDatabaseMeta() {
+                dialectMap = new HashMap<>();
+                metaDataMap = new HashMap<>();
+                dialectMap.put(DatabaseType.h2, Dialect.H2);
+                dialectMap.put(DatabaseType.mysql, Dialect.MYSQL);
+                dialectMap.put(DatabaseType.oracle, Dialect.ORACLE);
+
+                metaDataMap.put(DatabaseType.h2, new H2RDBDatabaseMetaData());
+                metaDataMap.put(DatabaseType.mysql, new MysqlRDBDatabaseMetaData());
+                metaDataMap.put(DatabaseType.oracle, new OracleRDBDatabaseMetaData());
+            }
+
+            @Override
+            public Dialect getDialect() {
+                return dialectMap.get(DataSourceHolder.getActiveDatabaseType());
+            }
+
+            @Override
+            public void init() {
+                metaDataMap.values().forEach(RDBDatabaseMetaData::init);
+            }
+
+            @Override
+            public SqlRender getRenderer(SqlRender.TYPE type) {
+                return metaDataMap.get(DataSourceHolder.getActiveDatabaseType()).getRenderer(type);
+            }
+
+            @Override
+            public String getName() {
+                return metaDataMap.get(DataSourceHolder.getActiveDatabaseType()).getName();
+            }
+        }
+    }
+
+    @Autowired
+    private DynDsTest dynDsTest;
+
+
+    @Test
+    @Transactional
+    public void testForEach() throws InterruptedException {
+        for (int i = 0; i < 100; i++) {
+            Thread.sleep(1000);
+            test();
+        }
+    }
+
+    @Autowired
+    private JmsTemplate jmsTemplate;
+
+    @Test
+    @Rollback(false)
+    public void test() {
+        try {
+            dynDsTest.testCreateTable();
+            dynDsTest.testInsert();
+            DataSourceHolder.switcher().use("test_ds");
+
+            dynDsTest.testCreateTable();
+
+            DataSourceHolder.switcher().use("test_ds2");
+
+            dynDsTest.testCreateTable();
+            System.out.println(DataSourceHolder.getActiveDatabaseType());
+            System.out.println(dynDsTest.testQuery());
+
+            DataSourceHolder.switcher().useLast();
+            System.out.println(DataSourceHolder.getActiveDatabaseType());
+            System.out.println(dynDsTest.testQuery());
+
+            DataSourceHolder.switcher().useLast();
+            System.out.println(DataSourceHolder.getActiveDatabaseType());
+            System.out.println(dynDsTest.testQuery());
+            jmsTemplate.convertAndSend("test", "hello");
+            Thread.sleep(1000);
+            //   throw new RuntimeException();
+        } catch (SQLException e) {
+            e.printStackTrace();
+        } catch (InterruptedException e) {
+            e.printStackTrace();
+        }
+    }
+
+
+    @Transactional
+    public static class DynDsTest {
+        private RDBDatabase database;
+
+        public void testCreateTable() throws SQLException {
+            database.createOrAlter("s_user")
+                    .addColumn().name("name").varchar(32).commit()
+                    .commit();
+        }
+
+        public DynDsTest(RDBDatabase database) {
+            this.database = database;
+        }
+
+        public void testInsert() throws SQLException {
+            database.getTable("s_user").createInsert()
+                    .value(Collections.singletonMap("name", "test"))
+                    .exec();
+        }
+
+        public List testQuery() throws SQLException {
+            return database.getTable("s_user").createQuery().list();
+        }
+    }
+
+}

hsweb-datasource/hsweb-datasource-jta/src/test/resources/application.yml

@@ -0,0 +1,44 @@
+
+spring:
+  jta:
+    enabled: true
+    atomikos:
+      datasource:
+        xa-data-source-class-name: com.alibaba.druid.pool.xa.DruidXADataSource
+        xa-properties:
+          url : jdbc:h2:mem:core;DB_CLOSE_ON_EXIT=FALSE
+          username : sa
+          password :
+        max-pool-size: 20
+        borrow-connection-timeout: 1000
+      connectionfactory:
+        max-pool-size: 20
+        local-transaction-mode: true
+
+  activemq:
+    broker-url: tcp://localhost:61616
+    in-memory: false
+
+logging:
+  level:
+      com.atomikos: WARN
+hsweb:
+  dynamic:
+    datasource:
+        test_ds:
+            xa-data-source-class-name: com.alibaba.druid.pool.xa.DruidXADataSource
+            xa-properties:
+              url: jdbc:h2:mem:test;DB_CLOSE_ON_EXIT=FALSE
+              username: sa
+              password:
+            max-pool-size: 20
+            borrow-connection-timeout: 1000
+        test_ds2:
+          xa-data-source-class-name: com.alibaba.druid.pool.xa.DruidXADataSource
+          xa-properties:
+#            url: jdbc:mysql://localhost:3306/hsweb?pinGlobalTxToPhysicalConnection=true&useSSL=false&useUnicode=true&characterEncoding=utf-8&autoReconnect=true&failOverReadOnly=false
+            url: jdbc:h2:mem:test2;DB_CLOSE_ON_EXIT=FALSE
+            username: sa
+            password:
+          max-pool-size: 20
+          borrow-connection-timeout: 1000

hsweb-datasource/pom.xml

@@ -10,6 +10,11 @@
     <modelVersion>4.0.0</modelVersion>
 
     <artifactId>hsweb-datasource</artifactId>
+    <packaging>pom</packaging>
+    <modules>
+        <module>hsweb-datasource-api</module>
+        <module>hsweb-datasource-jta</module>
+    </modules>
 
 
 </project>