在springboot项目集成r2dbc，集成mysql的流式代码DAO层

我引用的是springboot 2.7.0版本。

在pom.xml里引入r2dbc的包，和mysql的驱动包：

<dependency>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-boot-starter-webflux</artifactId>
		</dependency>
        <dependency>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-boot-starter-actuator</artifactId>
		</dependency>
        <dependency>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-boot-configuration-processor</artifactId>
			<optional>true</optional>
		</dependency>
        
        <!-- MySQL -->
		<dependency>
			<groupId>mysql</groupId>
			<artifactId>mysql-connector-java</artifactId>
		</dependency>
		<dependency>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-boot-starter-data-r2dbc</artifactId>
		</dependency>
		<dependency>
			<groupId>dev.miku</groupId>
			<artifactId>r2dbc-mysql</artifactId>
			<version>0.8.2.RELEASE</version>
		</dependency>
        

配置applicaion.properties:

spring.r2dbc.url=r2dbc:mysql://localhost:3306/db_name?autoReconnect=true&useUnicode=true&characterEncoding=UTF-8&zeroDateTimeBehavior=convertToNull&useSSL=false&useTimezone=true&serverZoneId=Asia/Shanghai
spring.r2dbc.username=root
spring.r2dbc.password=123456
spring.r2dbc.pool.enabled=true
spring.r2dbc.pool.initial-size=50
spring.r2dbc.pool.max-size=200
spring.r2dbc.pool.max-life-time=10
spring.r2dbc.pool.max-acquire-time.second=10
spring.r2dbc.pool.max-create-connection-time=10
spring.r2dbc.pool.max-idle-time.minutes=10
# s = second   m = minutes
#spring.r2dbc.pool.max-acquire-time=10s
#spring.r2dbc.pool.max-idle-time=10m
spring.r2dbc.pool.validation-query=select 1 from dual
spring.data.r2dbc.repositories.enabled=true

配置R2DBC到项目里：

package com.my.config;

import com.disney.wdpro.service.ecoupon.redemption.service.util.LocalDateTimeToZonedDateTimeConverter;
import com.disney.wdpro.service.ecoupon.redemption.service.util.ZonedDateTimeToLocalDateTimeConverter;
import io.r2dbc.pool.ConnectionPool;
import io.r2dbc.pool.ConnectionPoolConfiguration;
import io.r2dbc.spi.ConnectionFactory;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.actuate.r2dbc.ConnectionFactoryHealthIndicator;
import org.springframework.boot.autoconfigure.r2dbc.R2dbcProperties;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;
import org.springframework.data.convert.CustomConversions;
import org.springframework.data.r2dbc.convert.MappingR2dbcConverter;
import org.springframework.data.r2dbc.convert.R2dbcCustomConversions;
import org.springframework.data.r2dbc.core.R2dbcEntityTemplate;
import org.springframework.data.r2dbc.dialect.DialectResolver;
import org.springframework.data.r2dbc.dialect.R2dbcDialect;
import org.springframework.data.r2dbc.mapping.R2dbcMappingContext;
import org.springframework.data.r2dbc.repository.config.EnableR2dbcRepositories;
import org.springframework.data.relational.core.mapping.NamingStrategy;
import org.springframework.r2dbc.connection.R2dbcTransactionManager;
import org.springframework.r2dbc.core.DatabaseClient;
import org.springframework.transaction.ReactiveTransactionManager;

import java.time.Duration;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;


@Slf4j
@Configuration
@EnableR2dbcRepositories(
        basePackages = "com.my.repository"
        , entityOperationsRef = "r2dbcEntityTemplate"
)
public class ConfigR2dbcForErs {

    @Primary
    @Bean("ersR2dbcProperties")
    @ConfigurationProperties(prefix = "spring.r2dbc")
    public R2dbcProperties ersR2dbcProperties() {
        return new R2dbcProperties();
    }

    @Primary
    @Bean("ersConnectionFactory")
    public ConnectionFactory getErsConnectionFactory(
            @Qualifier("ersR2dbcProperties") R2dbcProperties ersR2dbcProperties
            , @Value("${spring.r2dbc.pool.initial-size}") Integer poolInitialSize
            , @Value("${spring.r2dbc.pool.max-size}") Integer poolMaxSize
            , @Value("${spring.r2dbc.pool.max-acquire-time.second}") Integer poolMaxAcquireTimeSecond
            , @Value("${spring.r2dbc.pool.max-idle-time.minutes}") Integer poolMaxIdleTimeMinutes
            , @Value("${spring.r2dbc.pool.max-life-time}") Integer poolMaxLifeTimeMinutes
            , @Value("${spring.r2dbc.pool.max-create-connection-time}") Integer maxCreateConnectionTimeSecond
    ) {
        ConnectionFactory factory = org.springframework.boot.r2dbc.ConnectionFactoryBuilder
                .withUrl(ersR2dbcProperties.getUrl())
                .username(ersR2dbcProperties.getUsername())
                .password(ersR2dbcProperties.getPassword())
                .build();
        ConnectionPoolConfiguration configuration = ConnectionPoolConfiguration
                .builder()
                .connectionFactory(factory)
                .maxSize(poolMaxSize)
                .initialSize(poolInitialSize)
                .maxAcquireTime(Duration.ofSeconds(poolMaxAcquireTimeSecond))
                .maxIdleTime(Duration.ofMinutes(poolMaxIdleTimeMinutes))
                .maxLifeTime(Duration.ofMinutes(poolMaxLifeTimeMinutes))
                .maxCreateConnectionTime(Duration.ofSeconds(maxCreateConnectionTimeSecond))
                .validationQuery("select 1 from dual")
                .build();
        ConnectionPool pool = new ConnectionPool(configuration);
        return pool;
    }

    @Primary
    @Bean("ersR2dbcCustomConversions")
    public R2dbcCustomConversions getR2dbcCustomConversions(
            @Qualifier("ersConnectionFactory") ConnectionFactory ersConnectionFactory
            , @Autowired LocalDateTimeToZonedDateTimeConverter timeConverter
            , @Autowired ZonedDateTimeToLocalDateTimeConverter zonedDateTimeToLocalDateTimeConverter
    ) {
        R2dbcDialect dialect = DialectResolver.getDialect(ersConnectionFactory);
        List<Object> converters = new ArrayList<>(dialect.getConverters());
        converters.addAll(R2dbcCustomConversions.STORE_CONVERTERS);

        // add custom datetime converter
        converters.add(timeConverter);
        converters.add(zonedDateTimeToLocalDateTimeConverter);

        R2dbcCustomConversions customConversions = new R2dbcCustomConversions(
                CustomConversions.StoreConversions.of(dialect.getSimpleTypeHolder(), converters),
                Collections.emptyList());
        return customConversions;
    }

    @Primary
    @Bean(name = "databaseClient")
    public DatabaseClient r2dbcDatabaseClient(@Qualifier("ersConnectionFactory") ConnectionFactory connectionFactory) {
        return DatabaseClient.builder().connectionFactory(connectionFactory).build();
    }

    @Primary
    @Bean("r2dbcEntityTemplate")
    public R2dbcEntityTemplate getR2dbcEntityTemplate(
            @Qualifier("databaseClient") DatabaseClient databaseClient
            , @Qualifier("ersConnectionFactory") ConnectionFactory ersConnectionFactory
            , @Qualifier("ersR2dbcCustomConversions") R2dbcCustomConversions ersR2dbcCustomConversions

    ) {
        R2dbcMappingContext context = new R2dbcMappingContext(NamingStrategy.INSTANCE);
        context.setSimpleTypeHolder(ersR2dbcCustomConversions.getSimpleTypeHolder());
        MappingR2dbcConverter converter = new MappingR2dbcConverter(context, ersR2dbcCustomConversions);
        R2dbcDialect dialect = DialectResolver.getDialect(ersConnectionFactory);
        R2dbcEntityTemplate template = new R2dbcEntityTemplate(
                databaseClient, dialect, converter);
        return template;
    }

    @Primary
    @Bean("ersConnectionFactoryHealthIndicator") // http://localhost:8080/actuator/health 可以被springboot的健康检查识别出来
    public ConnectionFactoryHealthIndicator connectionFactoryHealthIndicator(@Qualifier("ersConnectionFactory") ConnectionFactory connectionFactory) {
        return new ConnectionFactoryHealthIndicator(connectionFactory);
    }

    @Primary
    @Bean("reactiveTransactionManager")
    public ReactiveTransactionManager reactiveTransactionManager(@Qualifier("ersConnectionFactory") ConnectionFactory connectionFactory) {
        return new R2dbcTransactionManager(connectionFactory);
    }


}

end.