Spring Boot SQL JPA 未使用正确的副本

Question

我有一个应用程序，我试图在两个副本之间分配读取和写入。由于某种原因，JPA 只使用我的只读副本，而不是写副本。写入副本是主副本。结果是，当我使用 JPA 尝试写入数据时，我得到了“更新命令被拒绝”错误，因为它使用的是只读数据源。我尝试过自己做注释并使用@Transactional 注释。这两个注解都是通过 AOP 调用的，具有正确的数据源，但 JPA 不会使用它。

仅供参考 Spring JDBC 通过自定义注解正常工作。这严格来说是一个 JPA 问题。下面是一些代码：

我的 AOP 课程：

@Aspect
@Order(20)
@Component
public class RouteDataSourceInterceptor {

    @Around("@annotation(com.kenect.db.common.annotations.UseDataSource) && execution(* *(..))")
    public Object proceed(ProceedingJoinPoint pjp) throws Throwable {
        try {
            MethodSignature signature = (MethodSignature) pjp.getSignature();
            Method method = signature.getMethod();
            UseDataSource annotation = method.getAnnotation(UseDataSource.class);
            RoutingDataSource.setDataSourceName(annotation.value());
            return pjp.proceed();
        } finally {
            RoutingDataSource.resetDataSource();
        }
    }

    @Around("@annotation(transactional)")
    public Object proceed(ProceedingJoinPoint proceedingJoinPoint, Transactional transactional) throws Throwable {
        try {
            if (transactional.readOnly()) {
                RoutingDataSource.setDataSourceName(SQL_READ_REPLICA);
                Klogger.info("Routing database call to the read replica");
            } else {
                RoutingDataSource.setDataSourceName(SQL_MASTER_REPLICA);
                Klogger.info("Routing database call to the primary replica");
            }
            return proceedingJoinPoint.proceed();
        } finally {
            RoutingDataSource.resetDataSource();
        }
    }
}

我的 RoutingDataSource 类：

public class RoutingDataSource extends AbstractRoutingDataSource {

    private static final ThreadLocal<String> currentDataSourceName = new ThreadLocal<>();

    public static synchronized void setDataSourceName(String name) {
        currentDataSourceName.set(name);
    }

    public static synchronized void resetDataSource() {
        currentDataSourceName.remove();
    }

    @Override
    protected Object determineCurrentLookupKey() {
        return currentDataSourceName.get();
    }
}

AbstractDynamicDataSourceConfig

public abstract class AbstractDynamicDataSourceConfig {

    private final ConfigurableEnvironment environment;

    public AbstractDynamicDataSourceConfig(ConfigurableEnvironment environment) {
        this.environment = environment;
    }

    protected DataSource getRoutingDataSource() {
        Map<String, String> props = DBConfigurationUtils.getAllPropertiesStartingWith("spring.datasource", environment);
        List<String> dataSourceNames = DBConfigurationUtils.getDataSourceNames(props.keySet());

        RoutingDataSource routingDataSource = new RoutingDataSource();
        Map<Object, Object> dataSources = new HashMap<>();
        DataSource masterDataSource = null;

        for (String name : dataSourceNames) {
            DataSource dataSource = getDataSource("spring.datasource." + name);
            dataSources.put(name, dataSource);

            if (masterDataSource == null && name.toLowerCase().contains("master")) {
                masterDataSource = dataSource;
            }
        }

        if (dataSources.isEmpty()) {
            throw new KenectInvalidParameterException("No datasources found.");
        }

        routingDataSource.setTargetDataSources(dataSources);

        if (masterDataSource == null) {
            masterDataSource = (DataSource) dataSources.get(dataSourceNames.get(0));
        }

        routingDataSource.setDefaultTargetDataSource(masterDataSource);

        return routingDataSource;
    }

    protected DataSource getDataSource(String prefix) {
        HikariConfig hikariConfig = new HikariConfig();
        hikariConfig.setJdbcUrl(environment.getProperty(prefix + ".jdbcUrl"));
        hikariConfig.setUsername(environment.getProperty(prefix + ".username"));
        hikariConfig.setPassword(environment.getProperty(prefix + ".password"));

        return new HikariDataSource(hikariConfig);
    }
}

application.yaml

spring:
  datasource:
    master:
      jdbcUrl: jdbc:mysql://my-main-replica
      username: some-user
      password: some-password
    read-replica:
      jdbcUrl: jdbc:mysql://my-read-replica
      username: another-user
      password: another-password

如果我将注释与 JDBC 模板一起使用，那么它会按预期工作：

这行得通：

// Uses main replica as it is not specified
public Message insertMessage(Message message) {

    String sql = "INSERT INTO message(" +
            " `conversationId`," +
            " `body`)" +
            " VALUE (" +
            " :conversationId," +
            " :body" +
            ")";
    MapSqlParameterSource parameters = new MapSqlParameterSource();
    parameters.addValue("conversationId", message.getConversationId());
    parameters.addValue("body", message.getBody());
       
    namedJdbcTemplate.update(sql, parameters);
}

// Uses read replica
@UseDataSource(SQL_READ_REPLICA)
public List<Message> getMessage(long id) {

    MapSqlParameterSource parameters = new MapSqlParameterSource();
    parameters.addValue("id", id);
    String sql = "SELECT " +
            " conversationId," +
            " body" +
            " FROM message"
            " WHERE id = :id";

    return namedJdbcTemplate.query(sql, parameters, new BeanPropertyRowMapper<>(Message.class));
}

如果我使用 JPA 接口，它总是使用只读副本：

这失败了：

@Repository
public interface MessageJpaRepository extends JpaRepository<MessageEntity, Long> {

    // Should use the main-replica but always uses the read-replica
    @Modifying
    @Query(value =
            "UPDATE clarioMessage SET" +
                    " body = :body" +
                    " WHERE id = :id" +
                    " AND organizationId = :organizationId",
            nativeQuery = true)
    @Transactional
    int updateMessageBodyByIdAndOrganizationId(@Param("body") String body, @Param("id")long id, @Param("organizationId")long organizationId);
}

所以当我尝试使用主副本时，我只是收到以下错误。我试过使用@UseDataSource 注释，AOP 确实拦截了它。但是，它仍然使用只读副本。

java.sql.SQLSyntaxErrorException: UPDATE command denied to user 'read-replica-user'@'read replica IP' for table 'message'

我错过了什么？

Answer 1

• 当您使用 @UseDataSource 时，它正在工作，因此它似乎排除了方面实现的任何问题。

• 当您@Transactional 时，它使用辅助副本，而不管您的 AOP 是否被调用。我的怀疑是由 spring 创建的TransactionInterceptor 在您的RouteDataSourceInterceptor 之前调用。您可以尝试以下方法：

• 在您的 aop 方法中放置一个断点以及在 org.springframework.transaction.interceptor.TransactionInterceptor.invoke 方法中放置一个断点，以查看哪个首先调用。您希望首先调用您的拦截器

• 如果你的拦截器没有被首先调用，我会修改你的拦截器，使其具有如下高阶。

    @Aspect
    @Order(Ordered.HIGHEST_PRECEDENCE)
    @Component
    public class RouteDataSourceInterceptor {

• 我仍然不明白你是如何告诉TransactionInterceptor 选择你在RouteDataSourceInterceptor 中设置的DataSource。我没有使用多租户设置，但最近我遇到了一个我帮助解决的问题，我可以看到它正在实施AbstractDataSourceBasedMultiTenantConnectionProviderImpl。所以我希望你有类似的东西。 Not able to switch database after defining Spring AOP