使用 Spring 实现读写分离事务路由

1. 简介

在本教程中，我们将实现基于事务的路由，将写操作发送到主数据库，并将只读操作发送到副本数据库。这是在使用数据库复制来提高读取速度的应用程序中常见的一种模式。

2. 为什么要进行事务路由？

如果不进行路由，所有的查询都会指向同一个数据库。随着流量的增长，这可能会成为瓶颈，因此复制通过维护镜像主数据库的副本库来缓解这一问题。 这种策略提高了读取速度，但应用程序仍然需要知道将每个查询发送到哪里。我们将采用的方法是让事务元数据自动决定路由：

使用 Spring 的 AbstractRoutingDataSource，结合 @Transactional 注解的 readOnly 标志，我们可以拦截每个连接请求并将其引导至合适的数据库。

3. 场景设置

让我们从一个简单的实体和一个用于示例的仓库开始。这一层不需要任何特殊的配置。

3.1. 创建实体

首先创建一个具有基本属性的 Order 实体。我们将生成策略明确设置为 IDENTITY，以便使用数据库的自增列来提供 ID：

$ java

@Entity
public class Order {
    @Id
    @GeneratedValue(strategy = GenerationType.IDENTITY)
    private Long id;
    private String description;
    // 默认的 getters、setters 和构造函数
}

3.2. 创建仓库

接下来，创建一个 Spring Data 仓库：

$ java

public interface OrderRepository
  extends JpaRepository<Order, Long> {
}

现在我们已经具备了实现路由数据源所需的一切。

4. 实现路由 DataSource

我们将扩展 AbstractRoutingDataSource，根据当前事务决定使用哪个 DataSource。

4.1. 定义 DataSource 类型

首先，创建一个枚举来表示配置时使用的 DataSource 类型：

$ java

public enum DataSourceType {
    READ_WRITE, READ_ONLY
}

我们将使用 READ_WRITE 作为主 DataSource（所有 insert、update 和 delete 操作的目标）的查找键，使用 READ_ONLY 作为副本库（仅用于 select 查询）的查找键。

4.2. 创建路由逻辑

现在，我们将扩展 AbstractRoutingDataSource 并重写 determineCurrentLookupKey()。该方法通过 TransactionSynchronizationManager 检查当前事务是否为只读：

$ java

public class TransactionRoutingDataSource extends AbstractRoutingDataSource {
    @Override
    protected Object determineCurrentLookupKey() {
        boolean readOnly = TransactionSynchronizationManager
          .isCurrentTransactionReadOnly();
        if (readOnly) {
            return DataSourceType.READ_ONLY;
        }
        return DataSourceType.READ_WRITE;
    }
}

当需要连接时，Spring 会调用此方法，因此我们利用它将请求路由到主库或副本库。

5. 配置 DataSources

我们需要定义两个数据源的属性，并将它们配置为 Bean。请注意，此设置仅控制给定事务使用哪个连接。路由层无法防止向副本连接发出写操作，因为这种强制执行是在数据库层面配置的。

在实际生产中，这意味着通常只授予副本用户 SELECT 权限，或者将副本数据库本身配置为拒绝写操作。

5.1. 定义应用程序属性

使用 spring.datasource 前缀在 application.properties 中定义连接属性：

$ properties

spring.datasource.readwrite.url=jdbc:h2:mem:primary;DB_CLOSE_DELAY=-1
spring.datasource.readwrite.username=sa
spring.datasource.readwrite.driverClassName=org.h2.Driver
spring.datasource.readonly.url=jdbc:h2:mem:replica;DB_CLOSE_DELAY=-1
spring.datasource.readonly.username=sa
spring.datasource.readonly.driverClassName=org.h2.Driver

每个数据源都指向各自的内存 H2 数据库。我们将 DB_CLOSE_DELAY 设置为 -1，这样当最后一个连接关闭时，这些数据库不会被销毁。

5.2. 装配配置 Bean

由于我们定义了自定义的 DataSource Bean，因此不会使用 Spring Boot 的 JPA 自动配置。这意味着除非我们显式声明，否则 OrderRepository 将没有 EntityManagerFactory 或 TransactionManager。@EnableJpaRepositories 注解允许我们将仓库扫描指向我们在此类中定义的 Bean：

$ java

@Configuration
@EnableTransactionManagement
@EnableJpaRepositories(
  basePackageClasses = OrderRepository.class,
  entityManagerFactoryRef = "routingEntityManagerFactory",
  transactionManagerRef = "routingTransactionManager"
)
public class DataSourceConfiguration {
    // ...
}

basePackageClasses 属性从我们提供的类中提取包名以进行组件扫描。为了确保能扫描到所需的一切，OrderRepository 和 Order 必须在同一个包中。

我们首先从各自的属性前缀创建两个 DataSource Bean。先读取 spring.datasource.readwrite.* 属性：

$ java

@Bean
@ConfigurationProperties("spring.datasource.readwrite")
public DataSourceProperties readWriteProperties() {
    return new DataSourceProperties();
}

然后读取 spring.datasource.readonly.* 属性：

$ java

@Bean 
@ConfigurationProperties("spring.datasource.readonly") 
public DataSourceProperties readOnlyProperties() { 
    return new DataSourceProperties();
}

接下来，我们构建实际的 DataSource 实例。 首先是读写数据源：

$ java

@Bean
public DataSource readWriteDataSource() {
    return readWriteProperties()
      .initializeDataSourceBuilder()
      .build();
}

然后是只读数据源：

$ java

@Bean
public DataSource readOnlyDataSource() {
    return readOnlyProperties()
      .initializeDataSourceBuilder()
      .build();
}

5.3. 配置 TransactionRoutingDataSource

现在，我们将路由数据源装配到 TransactionRoutingDataSource Bean 中：

$ java

@Bean
public TransactionRoutingDataSource routingDataSource() {
    TransactionRoutingDataSource routingDataSource =
      new TransactionRoutingDataSource();
    // ...
    return routingDataSource;
}

我们将两个目标注册到一个 Map 中，并在调用 setTargetDataSources() 时使用它：

$ java

Map<Object, Object> dataSourceMap = new HashMap<>();
dataSourceMap.put(DataSourceType.READ_WRITE, readWriteDataSource());
dataSourceMap.put(DataSourceType.READ_ONLY, readOnlyDataSource());
routingDataSource.setTargetDataSources(dataSourceMap);
routingDataSource.setDefaultTargetDataSource(readWriteDataSource());

我们还调用了 setDefaultTargetDataSource()，它定义了例如代码在事务外运行时使用的回退数据源。

5.4. 定义延迟加载的 DataSource

由于 JPA 在事务同步设置只读标志之前就会获取连接，我们需要将路由数据源包装在 LazyConnectionDataSourceProxy 中。 这样，实际的连接会被推迟到执行第一条 SQL 语句时：

$ java

@Bean
@Primary
public DataSource dataSource() {
    return new LazyConnectionDataSourceProxy(routingDataSource());
}

我们将此 Bean 标记为 @Primary，以确保在自动装配 DataSource 时，不会错误地使用非延迟加载的 routingDataSource()。 然后，我们配置一个 EntityManagerFactory 来使用我们的延迟代理。当手动管理 JPA 配置时，LocalContainerEntityManagerFactoryBean 是标准选择，因为它与 Spring 的生命周期集成：

$ java

@Bean
public LocalContainerEntityManagerFactoryBean routingEntityManagerFactory(
  EntityManagerFactoryBuilder builder) {
    return builder
      .dataSource(dataSource())
      .packages(OrderRepository.class)
      .build();
}

我们还需要一个连接到我们工厂的 TransactionManager。我们返回一个 JpaTransactionManager，以便 @Transactional 方法能够将事务管理器的 readOnly 标志传播给我们的路由逻辑：

$ java

@Bean
public PlatformTransactionManager routingTransactionManager(
  LocalContainerEntityManagerFactoryBean routingEntityManagerFactory) {
    return new JpaTransactionManager(
      Objects.requireNonNull(routingEntityManagerFactory.getObject()));
}

完成所有这些配置后，我们就可以进入服务层了。

6. 创建服务层

我们将创建一个使用 @Transactional 注解来控制路由的服务：

$ java

@Service
public class OrderService {
    @Autowired
    OrderRepository orderRepository;
    // ...
}

标注了 readOnly = true 的方法会被路由到副本库，而其他事务则会进入主 DataSource：

$ java

@Transactional
public Order save(Order order) {
    return orderRepository.save(order);
}
@Transactional(readOnly = true)
public List<Order> findAllReadOnly() {
    return orderRepository.findAll();
}
@Transactional
public List<Order> findAllReadWrite() {
    return orderRepository.findAll();
}

在这个例子中，我们为每个 DataSource 都提供了一个查询方法。这在稍后的测试中非常有用。

7. 设置测试

在现实场景中，副本会持续从主库同步数据。但为了保持测试的简单性，我们将它们保持为不同步状态。利用这种隔离性，我们可以断言路由工作正常。

因此，在我们的案例中，如果只读查询返回了由读写事务写入的数据，我们就知道它命中了主库而不是副本库。

让我们设置测试类：

$ java

@SpringBootTest
class TransactionRoutingIntegrationTest {
    @Autowired
    OrderService orderService;
    // ...
}

首先，我们验证读写事务是否保留在主数据库上，以便保存的订单立即可见：

$ java

@Test
void whenSaveAndReadWithReadWrite_thenFindsOrder() {
    Order saved = orderService.save(new Order("laptop"));
    List<Order> result = orderService.findAllReadWrite();
    assertThat(result)
      .anyMatch(o -> o.getId().equals(saved.getId()));
}

然后，我们确认只读事务被路由到了副本库。由于副本是一个独立的、不同步的数据库，它无法访问保存到主 DataSource 中的订单：

$ java

@Test
void whenSaveAndReadWithReadOnly_thenOrderNotFound() {
    Order saved = orderService.save(new Order("keyboard"));
    List result = orderService.findAllReadOnly();
    assertThat(result)
      .noneMatch(o -> o.getId().equals(saved.getId()));
}

我们显式检查 findAllReadOnly() 是否不返回已保存订单的 ID，从而验证该订单没有被路由到副本库。

8. 注意事项

在使用这种路由模式时，需要考虑几个实际问题。

8.1. 复制延迟

从写入主库到同步到副本库之间存在延迟。延迟持续时间取决于基础设施。

因此，对于时间敏感的流程（例如写入实体后立即读取），将读取操作路由到副本库可能会返回陈旧数据。 在这种情况下，在同一个读写事务中执行这两个操作更为安全。

8.2. 嵌套事务

使用 Spring 的默认传播行为，从 @Transactional(readOnly = false) 方法内部调用的 @Transactional(readOnly = true) 方法会使用现有的事务。这意味着只有第一个创建的事务生效，后续 @Transactional 方法中的任何标志都会被忽略，因此查询仍会进入主数据库。要真正将内部调用路由到副本库，我们需要使用 Propagation.REQUIRES_NEW 来挂起外部事务并启动一个只读事务。

还要注意，如果两个方法都在同一个 Bean 中，Spring 将不会拦截内部调用。

8.3. 多个只读副本

也可以使用多个只读副本来分发流量。一种方法是扩展路由逻辑，通过使用只读 DataSource 列表，并在 determineCurrentLookupKey() 中使用轮询（round-robin）实现来从中选择一个，但这超出了本文的范围。

一种更稳健的方法是将只读连接委托给负载均衡的连接 URL。 根据基础设施的不同，有许多生产就绪的替代方案，如 PgBouncer 或 ProxySQL。

9. 结论

在本文中，我们使用 AbstractRoutingDataSource 实现了基于 Spring 的事务数据源路由。我们了解了如何分离读写和只读流量，为什么需要 LazyConnectionDataSourceProxy 来实现正确路由，以及如何通过集成测试来验证行为。

一如既往，源代码可在 GitHub 上获取。

分类导航

1. 简介

2. 为什么要进行事务路由？

3. 场景设置

3.1. 创建实体

3.2. 创建仓库

4. 实现路由 DataSource

4.1. 定义 DataSource 类型

4.2. 创建路由逻辑

5. 配置 DataSources

5.1. 定义应用程序属性

5.2. 装配配置 Bean

5.3. 配置 TransactionRoutingDataSource

5.4. 定义延迟加载的 DataSource

6. 创建服务层

7. 设置测试

8. 注意事项

8.1. 复制延迟

8.2. 嵌套事务

8.3. 多个只读副本

9. 结论