Java R2DBC 响应式编程数据库的使用

响应式API操作关系型数据库

引入

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-webflux</artifactId>
</dependency>
​
<!-- 整合Spring Data -->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-r2dbc</artifactId>
</dependency>
​
<!-- 驱动 -->
<!-- https://mvnrepository.com/artifact/io.asyncer/r2dbc-mysql -->
<dependency>
    <groupId>io.asyncer</groupId>
    <artifactId>r2dbc-mysql</artifactId>
    <version>1.1.2</version>
</dependency>
​
<!-- 连接池 -->
<!-- https://mvnrepository.com/artifact/io.r2dbc/r2dbc-pool -->
<dependency>
    <groupId>io.r2dbc</groupId>
    <artifactId>r2dbc-pool</artifactId>
    <version>1.0.1.RELEASE</version>
</dependency>

原生R2DBC

package com.tlovo.r2dbc;
​
import com.tlovo.r2dbc.entity.TAuthor;
import io.asyncer.r2dbc.mysql.MySqlConnectionConfiguration;
import io.asyncer.r2dbc.mysql.MySqlConnectionFactory;
import reactor.core.publisher.Mono;
​
import java.util.Scanner;
​
public class R2DBCTestDemo {
    public static void main(String[] args) {
        MySqlConnectionConfiguration configuration = MySqlConnectionConfiguration.builder()
                .host("192.168.10.133")
                .port(3306)
                .username("root")
                .password("123")
                .database("r2dbc_test")
                .build();
​
        // 1.获取连接工厂
        MySqlConnectionFactory factory = MySqlConnectionFactory.from(configuration);
​
        // 2.获取到连接，发送sql
        Mono.from(factory.create())
                // 获取到连接对象，执行sql并获取结果流
                .flatMapMany(conn ->
                        conn.createStatement("SELECT * FROM t_author where id=?")
                                .bind(0, 1L) // sql语句参数绑定
                                .execute() // 发送并执行sql语句
                )
                // 对结果流进行处理，转化为对应流（此处将每个结果转化为TAuthor对象）
                .flatMap(result -> {
                    return result.map(readable -> {
                        Long id = readable.get("id", Long.class);
                        String name = readable.get("name", String.class);
                        return new TAuthor(id, name);
                    });
                })
                // 订阅者消费数据流
                .subscribe(System.out::println);
​
        new Scanner(System.in).next();
    }
}

整合Spring Data

配置类

package com.tlovo.r2dbc.config;
​
import com.tlovo.r2dbc.converter.BookConverter;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.r2dbc.convert.R2dbcCustomConversions;
import org.springframework.data.r2dbc.dialect.MySqlDialect;
import org.springframework.data.r2dbc.repository.config.EnableR2dbcRepositories;
​
@Configuration
@EnableR2dbcRepositories // 开启R2DBC仓库功能
public class R2DbcConfiguration {
​
    /**
     * 自定义转换器配置
     */
    @Bean
    @ConditionalOnMissingBean
    public R2dbcCustomConversions r2dbcCustomConversions(){
        // 新增自定义转换器
        return R2dbcCustomConversions.of(MySqlDialect.INSTANCE, new BookConverter());
    }
}

CRUD接口

package com.tlovo.r2dbc.repositories;
​
import com.tlovo.r2dbc.entity.TBook;
import org.springframework.data.r2dbc.repository.Query;
import org.springframework.data.r2dbc.repository.R2dbcRepository;
import org.springframework.stereotype.Repository;
import reactor.core.publisher.Mono;
​
// R2dbcRepository<表实体类, 表主键类型>
@Repository
public interface BookRepository extends R2dbcRepository<TBook, Long> {
    // 继承 R2dbcRepository 后默认继承了一堆CRUD方法（类似于mybatis-plus）
    
    // 该方法实现由Springboot自动生成，仅限单表查询
    // where id In() and name like ?
    Flux<TAuthor> findAllByIdInAndNameLike(Collection<Long> id, String name);
    
    // 多表复杂查询 - 1对1
    @Query("select book.*,author.name as name from t_book book" +
            " left join t_author author on book.author_id = author.id" +
            " where book.id = :bookId")  // 自定义Query注解，指定sql语句
    Mono<TBook> findBookAndAuthor(Long bookId);
}

转换器

package com.tlovo.r2dbc.converter;
​
import com.tlovo.r2dbc.entity.TAuthor;
import com.tlovo.r2dbc.entity.TBook;
import io.r2dbc.spi.Row;
import org.springframework.core.convert.converter.Converter;
import org.springframework.data.convert.ReadingConverter;
import org.springframework.lang.NonNull;
​
import java.time.Instant;
​
/**
 * 自定义数据库实体转换器
 * 以后所有的Book查询结果封装都会调用该转换器
 */
@ReadingConverter // 读取数据库数据时，将Row转为TBook
public class BookConverter implements Converter<Row, TBook> {
    @Override
    public TBook convert(@NonNull Row source) {
        TBook book = new TBook();
        book.setId(source.get("id", Long.class));
        book.setTitle(source.get("title", String.class));
​
        Long authorId = source.get("author_id", Long.class);
        book.setAuthorId(authorId);
        book.setPublishTime(source.get("publish_time", Instant.class));
​
        // 让转换器兼容更多的表结构
        if (source.getMetadata().contains("name")) {
            TAuthor author = new TAuthor();
            author.setId(authorId);
            author.setName(source.get("name", String.class));
​
            book.setAuthor(author);
        }
​
        return book;
    }
}

数据库实体

package com.tlovo.r2dbc.entity;
​
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import lombok.ToString;
import org.springframework.data.annotation.Id;
import org.springframework.data.relational.core.mapping.Table;
​
@Data
@Table("t_author")
@ToString
@NoArgsConstructor
@AllArgsConstructor
public class TAuthor {
    @Id
    private Long id;
    private String name;
}

package com.tlovo.r2dbc.entity;
​
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import lombok.ToString;
import org.springframework.data.annotation.Id;
import org.springframework.data.relational.core.mapping.Table;
​
import java.time.Instant;
​
@Table("t_book")
@Data
@ToString
@NoArgsConstructor
@AllArgsConstructor
public class TBook {
    @Id
    private Long id;
    private String title;
    private Long authorId;
    private Instant publishTime; // 响应式中日期的映射用 Instant 或者 LocalXxx
​
    private TAuthor author; // 每个图书有唯一作者
}

Controller

package com.tlovo.r2dbc.controller;
​
import com.tlovo.r2dbc.entity.TBook;
import com.tlovo.r2dbc.repositories.BookRepository;
import lombok.RequiredArgsConstructor;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
​
@RestController
@RequiredArgsConstructor
public class BookController {
    private final BookRepository bookRepository;
​
    @GetMapping("/book/1")
    public Flux<TBook> book1() {
        return bookRepository.findAll();
    }
​
    @GetMapping("/book/2")
    public Mono<TBook> book2() {
        return bookRepository.findBookAndAuthor(1L);
    }
}