基于Springboot2.x+Mybatis Plus+Atomikos实现多数据源事务和DAP层多数据源动态切换

1
spring.datasource.druid.user.master.driver-class-name = com.mysql.cj.jdbc.Driver
2
spring.datasource.druid.user.master.initialSize = 5
3
spring.datasource.druid.user.master.maxActive = 100
4
spring.datasource.druid.user.master.maxPoolPreparedStatementPerConnectionSize = 20
5
spring.datasource.druid.user.master.maxWait = 60000
6
spring.datasource.druid.user.master.minEvictableIdleTimeMillis = 300000
7
spring.datasource.druid.user.master.minIdle = 5
8
spring.datasource.druid.user.master.password = root
9
spring.datasource.druid.user.master.poolPreparedStatements = true
10
spring.datasource.druid.user.master.remove-abandoned-timeout = 1800
11
spring.datasource.druid.user.master.testOnBorrow = false
12
spring.datasource.druid.user.master.testOnReturn = false
13
spring.datasource.druid.user.master.testWhileIdle = true
14
spring.datasource.druid.user.master.timeBetweenEvictionRunsMillis = 60000
15
spring.datasource.druid.user.master.url = jdbc:mysql://172.16.10.62:3306/user?useUnicode=true&characterEncoding=utf8&autoReconnect=true&useAffectedRows=true&useSSL=false&serverTimezone=Asia/Shanghai
16
spring.datasource.druid.user.master.username = root
17
spring.datasource.druid.user.master.validationQuery = SELECT 1
18
spring.datasource.druid.user.master.validationQueryTimeout = 10000
19
spring.datasource.druid.user.slave.driver-class-name = com.mysql.cj.jdbc.Driver
20
spring.datasource.druid.user.slave.initialSize = 5
21
spring.datasource.druid.user.slave.maxActive = 100
22
spring.datasource.druid.user.slave.maxPoolPreparedStatementPerConnectionSize = 20
23
spring.datasource.druid.user.slave.maxWait = 60000
24
spring.datasource.druid.user.slave.minEvictableIdleTimeMillis = 300000
25
spring.datasource.druid.user.slave.minIdle = 5
26
spring.datasource.druid.user.slave.password = root
27
spring.datasource.druid.user.slave.poolPreparedStatements = true
28
spring.datasource.druid.user.slave.remove-abandoned-timeout = 1800
29
spring.datasource.druid.user.slave.testOnBorrow = false
30
spring.datasource.druid.user.slave.testOnReturn = false
31
spring.datasource.druid.user.slave.testWhileIdle = true
32
spring.datasource.druid.user.slave.timeBetweenEvictionRunsMillis = 60000
33
spring.datasource.druid.user.slave.url = jdbc:mysql://172.16.10.62:3306/user?useUnicode=true&characterEncoding=utf8&autoReconnect=true&useAffectedRows=true&useSSL=false&serverTimezone=Asia/Shanghai
34
spring.datasource.druid.user.slave.username = root
35
spring.datasource.druid.user.slave.validationQuery = SELECT 1
36
spring.datasource.druid.user.slave.validationQueryTimeout = 10000
37
spring.datasource.druid.product.master.driver-class-name = com.mysql.cj.jdbc.Driver
38
spring.datasource.druid.product.master.initialSize = 5
39
spring.datasource.druid.product.master.maxActive = 100
40
spring.datasource.druid.product.master.maxPoolPreparedStatementPerConnectionSize = 20
41
spring.datasource.druid.product.master.maxWait = 60000
42
spring.datasource.druid.product.master.minEvictableIdleTimeMillis = 300000
43
spring.datasource.druid.product.master.minIdle = 5
44
spring.datasource.druid.product.master.password = root
45
spring.datasource.druid.product.master.poolPreparedStatements = true
46
spring.datasource.druid.product.master.remove-abandoned-timeout = 1800
47
spring.datasource.druid.product.master.testOnBorrow = false
48
spring.datasource.druid.product.master.testOnReturn = false
49
spring.datasource.druid.product.master.testWhileIdle = true
50
spring.datasource.druid.product.master.timeBetweenEvictionRunsMillis = 60000
51
spring.datasource.druid.product.master.url = jdbc:mysql://172.16.10.62:3306/product?useUnicode=true&characterEncoding=utf8&autoReconnect=true&useAffectedRows=true&useSSL=false&serverTimezone=Asia/Shanghai
52
spring.datasource.druid.product.master.username = root
53
spring.datasource.druid.product.master.validationQuery = SELECT 1
54
spring.datasource.druid.product.master.validationQueryTimeout = 10000
55
spring.datasource.druid.product.slave.driver-class-name = com.mysql.cj.jdbc.Driver
56
spring.datasource.druid.product.slave.initialSize = 5
57
spring.datasource.druid.product.slave.maxActive = 100
58
spring.datasource.druid.product.slave.maxPoolPreparedStatementPerConnectionSize = 20
59
spring.datasource.druid.product.slave.maxWait = 60000
60
spring.datasource.druid.product.slave.minEvictableIdleTimeMillis = 300000
61
spring.datasource.druid.product.slave.minIdle = 5
62
spring.datasource.druid.product.slave.password = root
63
spring.datasource.druid.product.slave.poolPreparedStatements = true
64
spring.datasource.druid.product.slave.remove-abandoned-timeout = 1800
65
spring.datasource.druid.product.slave.testOnBorrow = false
66
spring.datasource.druid.product.slave.testOnReturn = false
67
spring.datasource.druid.product.slave.testWhileIdle = true
68
spring.datasource.druid.product.slave.timeBetweenEvictionRunsMillis = 60000
69
spring.datasource.druid.product.slave.url = jdbc:mysql://172.16.10.62:3306/product?useUnicode=true&characterEncoding=utf8&autoReconnect=true&useAffectedRows=true&useSSL=false&serverTimezone=Asia/Shanghai
70
spring.datasource.druid.product.slave.username = root
71
spring.datasource.druid.product.slave.validationQuery = SELECT 1
72
spring.datasource.druid.product.slave.validationQueryTimeout = 10000
73
spring.datasource.type = com.alibaba.druid.pool.xa.DruidXADataSource

1
public class DynamicDataSource extends AbstractRoutingDataSource {
2

3
    private static final ThreadLocal<String> contextHolder = new ThreadLocal<>();
4

5
    public DynamicDataSource(DataSource defaultTargetDataSource, Map<Object, Object> targetDataSources) {
6
        super.setDefaultTargetDataSource(defaultTargetDataSource);
7
        super.setTargetDataSources(targetDataSources);
8
        super.afterPropertiesSet();
9
    }
10

11
    @Override
12
    protected Object determineCurrentLookupKey() {
13
        return getDataSource();
14
    }
15

16
    public static void setDataSource(String dataSource) {
17
        contextHolder.set(dataSource);
18
    }
19

20
    public static String getDataSource() {
21
        return contextHolder.get();
22
    }
23

24
    public static void clearDataSource() {
25
        contextHolder.remove();
26
    }
27

28
}

1
@Configuration
2
public class DynamicDataSourceConfig {
3

4
    @Value("${spring.datasource.type}")
5
    String xaDataSourceClassName;
6

7
    @Bean
8
    @ConfigurationProperties("spring.datasource.druid.user.master")
9
    public DruidXADataSource userMasterDruidDataSource() {
10
        return new DruidXADataSource();
11
    }
12

13
    @Bean
14
    @ConfigurationProperties("spring.datasource.druid.product.master")
15
    public DruidXADataSource productMasterDruidDataSource() {
16
        return new DruidXADataSource();
17
    }
18

19
    @Bean
20
    @ConfigurationProperties("spring.datasource.druid.user.slave")
21
    public DruidXADataSource userSlaveDruidDataSource() {
22
        return new DruidXADataSource();
23
    }
24

25
    @Bean
26
    @ConfigurationProperties("spring.datasource.druid.product.slave")
27
    public DruidXADataSource productSlaveDruidDataSource() {
28
        return new DruidXADataSource();
29
    }
30

31
    @Bean(name = "userMasterDataSource")
32
    public DataSource userMasterDataSource(DruidXADataSource userMasterDruidDataSource) {
33
        return getXaDataSource(userMasterDruidDataSource, "userMasterDruidDataSource");
34
    }
35

36
    @Bean(name = "productMasterDataSource")
37
    public DataSource orderMasterDataSource(DruidXADataSource productMasterDruidDataSource) {
38
        return getXaDataSource(productMasterDruidDataSource, "productMasterDruidDataSource");
39
    }
40

41
    @Bean(name = "userSlaveDataSource")
42
    public DataSource userSlaveDataSource(DruidXADataSource userSlaveDruidDataSource) {
43
        return getXaDataSource(userSlaveDruidDataSource, "userSlaveDruidDataSource");
44
    }
45

46
    @Bean(name = "productSlaveDataSource")
47
    public DataSource orderSlaveDataSource(DruidXADataSource productSlaveDruidDataSource) {
48
        return getXaDataSource(productSlaveDruidDataSource, "productSlaveDruidDataSource");
49
    }
50

51
    @Bean(name = "dynamicDataSource")
52
    @DependsOn({"transactionManager"})
53
    public DynamicDataSource dataSource(@Qualifier("userMasterDataSource") DataSource userMasterDataSource,
54
                                        @Qualifier("productMasterDataSource") DataSource productMasterDataSource,
55
                                        @Qualifier("userSlaveDataSource") DataSource userSlaveDataSource,
56
                                        @Qualifier("productSlaveDataSource") DataSource productSlaveDataSource) {
57
        Map<Object, Object> targetDataSources = new HashMap<>();
58
        String MASTER_SUFFIX = "Master";
59
        targetDataSources.put(DataSourceKey.USER + MASTER_SUFFIX, userMasterDataSource);
60
        targetDataSources.put(DataSourceKey.PRODUCT + MASTER_SUFFIX, productMasterDataSource);
61
        String SLAVE_SUFFIX = "Slave";
62
        targetDataSources.put(DataSourceKey.USER + SLAVE_SUFFIX, userSlaveDataSource);
63
        targetDataSources.put(DataSourceKey.PRODUCT + SLAVE_SUFFIX, productSlaveDataSource);
64
        return new DynamicDataSource(userMasterDataSource, targetDataSources);
65
    }
66

67
    @Bean(name = "sqlSessionFactory")
68
    public SqlSessionFactory sqlSessionFactory(@Qualifier("dynamicDataSource") DataSource dataSource) throws Exception {
69
        MybatisSqlSessionFactoryBean sqlSessionFactoryBean = new MybatisSqlSessionFactoryBean();
70
        sqlSessionFactoryBean.setDataSource(dataSource);
71
        return sqlSessionFactoryBean.getObject();
72
    }
73

74
    @Bean(name = "sqlSessionTemplate")
75
    public SqlSessionTemplate sqlSessionTemplate(@Qualifier("sqlSessionFactory") SqlSessionFactory sqlSessionFactory) {
76
        return new SqlSessionTemplate(sqlSessionFactory);
77
    }
78

79
    /**
80
     * 分布式事务使用JTA管理，
81
     * 不管有多少个数据源只要配置一个 JtaTransactionManager
82
     */
83

84
    @Bean
85
    public TransactionManager atomikosTransactionManager() {
86
        UserTransactionManager userTransactionManager = new UserTransactionManager();
87
        userTransactionManager.setForceShutdown(true);
88
        return userTransactionManager;
89
    }
90

91
    @Bean(name = "userTransaction")
92
    public UserTransaction userTransaction() throws Throwable {
93
        UserTransactionImp userTransactionImp = new UserTransactionImp();
94
        userTransactionImp.setTransactionTimeout(10000);
95
        return userTransactionImp;
96
    }
97

98
    @Bean(name = "transactionManager")
99
    @DependsOn({"userTransaction", "atomikosTransactionManager"})
100
    public PlatformTransactionManager transactionManager() throws Throwable {
101
        JtaTransactionManager jtaTransactionManager = new JtaTransactionManager(userTransaction(), atomikosTransactionManager());
102
        jtaTransactionManager.setAllowCustomIsolationLevels(true);
103
        jtaTransactionManager.setDefaultTimeout(30);
104
        return jtaTransactionManager;
105
    }
106

107
    public DataSource getXaDataSource(DruidXADataSource druidXADataSource, String resourceName) {
108
        AtomikosDataSourceBean xaDataSource = new AtomikosDataSourceBean();
109
        xaDataSource.setXaDataSource(druidXADataSource);
110
        xaDataSource.setXaDataSourceClassName(xaDataSourceClassName);
111
        xaDataSource.setUniqueResourceName(resourceName);
112
        xaDataSource.setMinPoolSize(10);
113
        xaDataSource.setMaxPoolSize(20000);
114
        xaDataSource.setBorrowConnectionTimeout(60);
115
        xaDataSource.setTestQuery("SELECT 1");
116
        return xaDataSource;
117
    }
118

119
}

1
@Aspect
2
@Configuration
3
public class TransactionConfig {
4

5
    private static final String AOP_POINTCUT_EXPRESSION = "execution(* com.xxx.xxxx.service.*.*(..))";
6

7
    @Autowired
8
    private TransactionManager transactionManager;
9

10
    @Bean
11
    public TransactionInterceptor txAdvice() {
12
        NameMatchTransactionAttributeSource source = new NameMatchTransactionAttributeSource();
13
        /*只读事务，不做更新操作*/
14
        RuleBasedTransactionAttribute readOnlyTx = new RuleBasedTransactionAttribute();
15
        readOnlyTx.setReadOnly(true);
16
        readOnlyTx.setPropagationBehavior(TransactionDefinition.PROPAGATION_NOT_SUPPORTED);
17
        /*当前存在事务就使用当前事务，当前不存在事务就创建一个新的事务*/
18
        RuleBasedTransactionAttribute requiredTx = new RuleBasedTransactionAttribute();
19
        requiredTx.setRollbackRules(
20
                Collections.singletonList(new RollbackRuleAttribute(Exception.class)));
21
        requiredTx.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRED);
22
        Map<String, TransactionAttribute> txMap = new HashMap<>();
23
        txMap.put("get*", readOnlyTx);
24
        txMap.put("select*", readOnlyTx);
25
        txMap.put("find*", readOnlyTx);
26
        txMap.put("query*", readOnlyTx);
27
        txMap.put("search*", readOnlyTx);
28
        txMap.put("count*", readOnlyTx);
29
        txMap.put("list*", readOnlyTx);
30
        txMap.put("add*", requiredTx);
31
        txMap.put("save*", requiredTx);
32
        txMap.put("insert*", requiredTx);
33
        txMap.put("update*", requiredTx);
34
        txMap.put("delete*", requiredTx);
35
        txMap.put("existTable", readOnlyTx);
36
        txMap.put("*", requiredTx);
37
        source.setNameMap(txMap);
38
        TransactionInterceptor txAdvice = new TransactionInterceptor(transactionManager, source);
39
        return txAdvice;
40
    }
41

42
    @Bean
43
    public Advisor txAdviceAdvisor() {
44
        AspectJExpressionPointcut pointcut = new AspectJExpressionPointcut();
45
        pointcut.setExpression(AOP_POINTCUT_EXPRESSION);
46
        return new DefaultPointcutAdvisor(pointcut, txAdvice());
47
    }
48
}

1
@Slf4j
2
public class DBAspect {
3

4
    private final String MASTER_SUFFIX = "Master";
5

6
    private final String SLAVE_SUFFIX = "Slave";
7

8
    private final String[] methodArray = {"select", "find", "get", "query", "search", "list", "count"};
9

10
    @Pointcut("@annotation(com.xxx.xxxx.annotation.DS)")
11
    protected void dataSourcePointCut() {
12

13
    }
14

15
    @Around("dataSourcePointCut()")
16
    public Object around(ProceedingJoinPoint point) throws Throwable {
17
        String methodName = point.getSignature().getName();
18
        boolean readOnly = false;
19
        for (String methodPrefix : methodArray) {
20
            if (methodName.startsWith(methodPrefix)) {
21
                readOnly = true;
22
                break;
23
            }
24
        }
25
        String dataSourceName = determineDatasource(point);
26
        if (StringUtils.isBlank(dataSourceName)) {
27
            log.info("设置为默认数据源：{}", DataSourceKey.USER);
28
            DynamicDataSource.setDataSource(DataSourceKey.USER);
29
        } else {
30
            if (readOnly) {
31
                if (DataSourceKey.PRODUCT.equals(dataSourceName)) {
32
                    log.info("切换到{}数据源", DataSourceKey.PRODUCT + SLAVE_SUFFIX);
33
                    DynamicDataSource.setDataSource(DataSourceKey.PRODUCT + SLAVE_SUFFIX);
34
                } else {
35
                    log.info("切换到{}数据源", DataSourceKey.USER + SLAVE_SUFFIX);
36
                    DynamicDataSource.setDataSource(DataSourceKey.USER + SLAVE_SUFFIX);
37
                }
38
            } else {
39
                if (DataSourceKey.PRODUCT.equals(dataSourceName)) {
40
                    log.info("切换到{}数据源", DataSourceKey.PRODUCT + MASTER_SUFFIX);
41
                    DynamicDataSource.setDataSource(DataSourceKey.PRODUCT + MASTER_SUFFIX);
42
                } else {
43
                    log.info("切换到{}数据源", DataSourceKey.USER + MASTER_SUFFIX);
44
                    DynamicDataSource.setDataSource(DataSourceKey.USER + MASTER_SUFFIX);
45
                }
46
            }
47
        }
48
        return point.proceed();
49
    }
50

51
    private String determineDatasource(ProceedingJoinPoint point) {
52
        String methodName = point.getSignature().getName();
53
        Class targetClass = point.getSignature().getDeclaringType();
54
        String dataSourceForTargetClass = resolveDataSourceFromClass(point);
55
        String dataSourceForTargetMethod = resolveDataSourceFromMethod(targetClass, methodName);
56
        String resultDS = determinateDataSource(dataSourceForTargetClass, dataSourceForTargetMethod);
57
        return resultDS;
58
    }
59

60
    private String resolveDataSourceFromClass(ProceedingJoinPoint point) {
61
        //获取当前类上的注解
62
        DS classAnnotation = (DS) ((Class) AopUtils.getTargetClass(point.getTarget()).getGenericInterfaces()[0]).getAnnotation(DS.class);
63
        return null != classAnnotation ? resolveDataSourceName(classAnnotation) : null;
64
    }
65

66
    private String resolveDataSourceFromMethod(Class targetClass, String methodName) {
67
        Method m = findUniqueMethod(targetClass, methodName);
68
        if (m != null) {
69
            DS choDs = m.getAnnotation(DS.class);
70
            return resolveDataSourceName(choDs);
71
        }
72
        return null;
73
    }
74

75
    private String resolveDataSourceName(DS ds) {
76
        return ds == null ? null : ds.value();
77
    }
78

79
    private String determinateDataSource(String classDS, String methodDS) {
80
        return methodDS == null ? classDS : methodDS;
81
    }
82

83
    @After("dataSourcePointCut()")
84
    public void restoreActivityDataSource(JoinPoint point) {
85
        DynamicDataSource.clearDataSource();
86
    }
87

88
    private static Method findUniqueMethod(Class<?> clazz, String name) {
89
        Class<?> searchType = clazz;
90
        while (searchType != null) {
91
            Method[] methods = (searchType.isInterface() ? searchType.getMethods() : searchType.getDeclaredMethods());
92
            for (Method method : methods) {
93
                if (name.equals(method.getName())) {
94
                    return method;
95
                }
96
            }
97
            searchType = searchType.getSuperclass();
98
        }
99
        return null;
100
    }
101
}