Spring WebFlux分享---R2DBC多数据源自由切换

Hello，大家好，不知不觉离上一次发表文章已经4年了（抹汗）。因为工作和生活的问题，要么就是开了草稿一点没写，要么就是写了一半被打断然后续不上了或者已经过时了。不信你们看我草稿截图！真不是我偷懒啊！（抹汗）

但不管怎么说，这一次我会认真的坚持的写博客了！

为了避免同质化，我这次就分享一下R2DBC吧！

1.R2DBC介绍

R2DBC全称Reactive Relational Database Connectivity，也就是反应式关系型数据库连接。R2DBC是基于Reactive Streams标准来设计的。通过使用R2DBC，你可以使用reactive API来操作数据。

嗯？你问我为什么推荐R2DBC？嗯。。我并没有推荐哈，我只是纯粹的分享。

它确实有优点，他是支持响应式IO的，不同于JDBC的阻塞IO，性能确实有提升。但它不提供ORM框架的缓存，延迟加载，后写或其他许多功能。所以它是有好有坏的，各位需要按需使用，毕竟合适的才是最好的。

嗯？你以为我要讲R2DBC原理？不不不，这些东西你可以去别处找，我们今天就重点讲一下他的多数据源切换原理。

2.原理分析

和传统的spring-jdbc的AbstractRoutingDataSource一样，spring-r2dbc也有相同功能的AbstractRoutingConnectionFactory，以做到切换数据源。

嗯。先看下org.springframework.r2dbc.connection.lookup.AbstractRoutingConnectionFactory的核心源码。

public abstract class AbstractRoutingConnectionFactory implements ConnectionFactory, InitializingBean {private static final Object FALLBACK_MARKER = new Object();@Nullableprivate Map<?, ?> targetConnectionFactories;@Nullableprivate Object defaultTargetConnectionFactory;private boolean lenientFallback = true;private ConnectionFactoryLookup connectionFactoryLookup = new MapConnectionFactoryLookup();@Nullableprivate Map<Object, ConnectionFactory> resolvedConnectionFactories;@Nullableprivate ConnectionFactory resolvedDefaultConnectionFactory;// 需要在实现类中调用这个方法，设置一个Map<?,?>// 一般我们会将 数据源名称作为key， 数据源作为value// determineTargetConnectionFactory()就是通过这个Map来查询数据源public void setTargetConnectionFactories(Map<?, ?> targetConnectionFactories) {this.targetConnectionFactories = targetConnectionFactories;}// 需要在实现类中调用这个方法// determineTargetConnectionFactory()当获取不到数据源时// 返回这个默认的ConnectionFactorypublic void setDefaultTargetConnectionFactory(Object defaultTargetConnectionFactory) {this.defaultTargetConnectionFactory = defaultTargetConnectionFactory;}// 省略。。。// 经典spring初始化bean的必走的一个方法// 将targetConnectionFactories设置进resolvedConnectionFactories@Overridepublic void afterPropertiesSet() {Assert.notNull(this.targetConnectionFactories, "Property 'targetConnectionFactories' must not be null");this.resolvedConnectionFactories = CollectionUtils.newHashMap(this.targetConnectionFactories.size());this.targetConnectionFactories.forEach((key, value) -> {Object lookupKey = resolveSpecifiedLookupKey(key);ConnectionFactory connectionFactory = resolveSpecifiedConnectionFactory(value);this.resolvedConnectionFactories.put(lookupKey, connectionFactory);});if (this.defaultTargetConnectionFactory != null) {this.resolvedDefaultConnectionFactory = resolveSpecifiedConnectionFactory(this.defaultTargetConnectionFactory);}}// 省略。。。// 最最核心的方法protected Mono<ConnectionFactory> determineTargetConnectionFactory() {Assert.state(this.resolvedConnectionFactories != null, "ConnectionFactory router not initialized");// 通过我们实现的方法获取数据源名称Mono<Object> lookupKey = determineCurrentLookupKey().defaultIfEmpty(FALLBACK_MARKER);return lookupKey.handle((key, sink) -> {// 根据lookupKey从resolvedConnectionFactories中获取ConnectionFactoryConnectionFactory connectionFactory = this.resolvedConnectionFactories.get(key);if (connectionFactory == null && (key == FALLBACK_MARKER || this.lenientFallback)) {connectionFactory = this.resolvedDefaultConnectionFactory;}if (connectionFactory == null) {sink.error(new IllegalStateException(String.format("Cannot determine target ConnectionFactory for lookup key '%s'", key == FALLBACK_MARKER ? null : key)));return;}sink.next(connectionFactory);});}// 我们需要实现的方法，函数返回值是一个对应着自己数据源的名称protected abstract Mono<Object> determineCurrentLookupKey();}

可以看到，逻辑就是，需要先将一个Map<数据源名，数据源>通过setTargetConnectionFactories()设置好，然后再通过determineCurrentLookupKey()获取当前上下文中的数据源名称，spring就会通过determineTargetConnectionFactory()将对应的数据源注入你的dao对象中。

3.多数据源切换实现

下面贴一下我实现的子类CCRoutingConnectionFactory的代码：

public class CCRoutingConnectionFactory extends AbstractRoutingConnectionFactory {private final static Logger logger = LoggerFactory.getLogger(CCRoutingConnectionFactory.class);private final static String DB_KEY = "my_r2dbc_content_key";private final String defaultConnectionFactoryKey;private static CCConnectionFactory ccConnectionFactory;CCRoutingConnectionFactory(CCConnectionFactory ccConnectionFactory) {// 这个是我编写的自定义的ConnectionFactory，下期分享！CCRoutingConnectionFactory.ccConnectionFactory = ccConnectionFactory;Map<String, ConnectionFactory> connectionFactories = ccConnectionFactory.getConnectionFactories();this.defaultConnectionFactoryKey = ccConnectionFactory.getDefaultConnectionFactoryKey();// set target MapssetTargetConnectionFactories(connectionFactories);// set default factorysetDefaultTargetConnectionFactory(connectionFactories.get(this.defaultConnectionFactoryKey));}public static <T> Mono<T> putR2dbcSource(Mono<T> mono, String group) {return mono.contextWrite(ctx -> ctx.put(DB_KEY, ccConnectionFactory.getConnectionFactoryKey(group)));}public static <T> Flux<T> putR2dbcSource(Flux<T> flux, String group) {return flux.contextWrite(ctx -> ctx.put(DB_KEY, ccConnectionFactory.getConnectionFactoryKey(group)));}@Overrideprotected Mono<Object> determineCurrentLookupKey() {return Mono.deferContextual(Mono::just).map(ctx -> {if (ctx.hasKey(DB_KEY)) {if (logger.isDebugEnabled()) {logger.debug("determine datasource: " + ctx.get(DB_KEY));}return ctx.get(DB_KEY);} else {return this.defaultConnectionFactoryKey;}});}
}

这里补充个知识点，Webflux可以通过contextWrite(key,value)写入数据到上下文中，并且通过Mono.deferContextual()获取上下文，可以理解为以前的ThreadLocal吧（只是功能近似）。所以在上面的代码中，我是通过putR2dbcSource()的方法，将数据源名称写入到context中，然后在determineCurrentLookupKey()方法中取出上下文中存储的值，再获取到相应的ConnectionFactory。

只需要下面这样调用，就可以选择执行sql的数据源了：

1.选择“master”数据源

Flux<User> users = CCRoutingConnectionFactory.putR2dbcSource(userDAO.findAll(), "master");

2.选择“slave”数据源

Flux<User> users = CCRoutingConnectionFactory.putR2dbcSource(userDAO.findAll(), "slave");

4.添加注解和AOP，解决代码入侵问题

虽然这时候数据源已经可以自由切换了，但严谨，温柔，漂亮，帅气的同学肯定要说了：你这个对代码有入侵，老项目修改起来不方便，你个辣鸡！

博主：好好好，那我改改。。。（苦笑）

1.编写注解类，嗯，就一个value属性，用来填写数据源的名称。

@Retention(RetentionPolicy.RUNTIME)
@Target({ElementType.METHOD})
@Documented
public @interface R2DBCSource {String value();
}

2.编写切面代码，嗯，也就是取注解的值，放进Context中。

@Aspect
@Order(Ordered.HIGHEST_PRECEDENCE)
public class R2DBCSourceAOP {@Pointcut(value = "@annotation(cn.git_chinwin.cc.plugins.R2DBCSource)")public void point() {}@Around(value = "point()")public Object dynamicSelectSource(ProceedingJoinPoint pjp) {MethodSignature signature = (MethodSignature) pjp.getSignature();Method method = signature.getMethod();R2DBCSource r2DBCSource = method.getAnnotation(R2DBCSource.class);if (method.getReturnType() == Mono.class) {return CCRoutingConnectionFactory.putR2dbcSource((Mono<?>) pjp.proceed(), r2DBCSource.value());} else if (method.getReturnType() == Flux.class) {return CCRoutingConnectionFactory.putR2dbcSource((Flux<?>) pjp.proceed(), r2DBCSource.value());} else {throw new RuntimeException("不支持别的发布类型");}}
}

3.通过在方法上添加注解，即可选择数据源

    @GetMapping("/user")@R2DBCSource("master")public Flux<User> getUsers() {logger.info("~~~~~~~~~~~~~~~~~~~~~~~~");return userDAO.findAll();}

今天就先分享到这里吧，上面代码中的CCConnectionFactory这是我编写的规划数据源的初始化和分组的类，这里就不详细深入讲解了，有兴趣的同学就去我的demo看看吧。
https://github.com/chinwin94/DynamicDbSrouceWithR2dbchttps://github.com/chinwin94/DynamicDbSrouceWithR2dbc

很久没写博客了，可能写的有点乱，还望包容。

觉得写的好的话，请别吝啬你的点赞，当然，写的不好，开喷便是。

祝大家周末愉快！

2022.06.24