Реактивный SQL с jOOQ 3.15 и R2DBC

Одна из самых больших новых функций недавно выпущенного jOOQ 3.15 - это поддержка реактивных запросов с помощью R2DBC. Это был очень популярный запрос функциональности, и мы наконец его выполнили.

Вы можете продолжать использовать jOOQ так, как вы привыкли, обеспечивая типобезопасный SQL, встроенный в Java, kotlin или scala, но выполнение ваших запросов больше не блокируется. Вместо этого ваш jOOQ ResultQuery или Query может использоваться как Publisher<R> или Publisher<Integer> в реализации reactive-streams (реактивных потоков) по вашему выбору.

Вместо (или в дополнение к) настройки вашего jOOQ DSLContext с помощью JDBC java.sql.Connection или javax.sql.DataSource просто настройте его с помощью R2DBC io.r2dbc.spi.Connection или io.r2dbc.spi.ConnectionFactory:

ConnectionFactory connectionFactory = ConnectionFactories.get(
    ConnectionFactoryOptions
        .parse("r2dbc:h2:file://localhost/~/r2dbc-test")
        .mutate()
        .option(ConnectionFactoryOptions.USER, "sa")
        .option(ConnectionFactoryOptions.PASSWORD, "")
        .build()
);
 
DSLContext ctx = DSL.using(connectionFactory);

В качестве альтернативы используйте Spring Boot для автоматической настройки jOOQ следующим образом:

Конечно, хороший обзор: pic.twitter.com/tUgNkwzCK4

- Ангел Леонард (@anghelleonard) 15 июля 2021 г.

Используя этот DSLContext, вы можете строить свои запросы как обычно, но вместо того, чтобы вызывать обычные блокирующие методы execute() или fetch(), вы просто оберните запрос в Flux, например. Предположим, что вы запустили генератор кода jOOQ на своей H2 INFORMATION_SCHEMA, тогда вы можете написать:

record Table(String schema, String table) {}
 
Flux.from(ctx
        .select(
            INFORMATION_SCHEMA.TABLES.TABLE_SCHEMA,
            INFORMATION_SCHEMA.TABLES.TABLE_NAME)
        .from(INFORMATION_SCHEMA.TABLES))
 
    // Type safe mapping from Record2<String, String> to Table::new
    .map(Records.mapping(Table::new))
    .doOnNext(System.out::println)
    .subscribe();

jOOQ получит R2DBC Connection у ConnectionFactory и освободит ее после выполнения запроса, что позволит оптимизировать управление ресурсами, что в противном случае несколько сложно с R2DBC и reactor. Другими словами, приведенный выше код соответствует этому написанному вручную запросу:

Flux.usingWhen(
        connectionFactory.create(),
        c -> c.createStatement(
                """
                SELECT table_schema, table_name
                FROM information_schema.tables
                """
             ).execute(),
        c -> c.close()
    )
    .flatMap(it -> it.map((r, m) -> 
         new Table(r.get(0, String.class), r.get(1, String.class))
    ))
    .doOnNext(System.out::println)
    .subscribe();

Оба напечатают что-то вроде следующего:

Обратите внимание, что, если вы используете JDBC, а не R2DBC, вы можете продолжать использовать jOOQ API с библиотеками реактивных потоков в режиме блокировки точно так же, как указано выше, например, если ваша любимая СУБД еще не поддерживает реактивный драйвер R2DBC. В настоящее время список поддерживаемых драйверов согласно r2dbc.io включает:

• oracle-r2dbc

• r2dbc-h2

• r2dbc-postgresql

• r2dbc-mysql

• r2dbc-mssql

• mariadb-connector-r2dbc

Все из них интегрированы с jOOQ 3.15+.

Работоспособный пример

Поиграйте с примером, представленным здесь: https://github.com/jOOQ/jOOQ/tree/main/jOOQ-examples/jOOQ-r2dbc-example

Он использует следующую схему:

CREATE TABLE r2dbc_example.author (
  id INT NOT NULL AUTO_INCREMENT,
  first_name VARCHAR(100) NOT NULL,
  last_name VARCHAR(100) NOT NULL,
   
  CONSTRAINT pk_author PRIMARY KEY (id)
);
 
CREATE TABLE r2dbc_example.book (
  id INT NOT NULL AUTO_INCREMENT,
  author_id INT NOT NULL,
  title VARCHAR(100) NOT NULL,
   
  CONSTRAINT pk_book PRIMARY KEY (id),
  CONSTRAINT fk_book_author FOREIGN KEY (id) 
    REFERENCES r2dbc_example.author
);

и выполняет следующий код

Flux.from(ctx
        .insertInto(AUTHOR)
        .columns(AUTHOR.FIRST_NAME, AUTHOR.LAST_NAME)
        .values("John", "Doe")
        .returningResult(AUTHOR.ID))
    .flatMap(id -> ctx
        .insertInto(BOOK)
        .columns(BOOK.AUTHOR_ID, BOOK.TITLE)
        .values(id.value1(), "Fancy Book"))
    .thenMany(ctx
        .select(
             BOOK.author().FIRST_NAME, 
             BOOK.author().LAST_NAME, 
             BOOK.TITLE)
        .from(BOOK))
    .doOnNext(System.out::println)
    .subscribe();

Чтобы вставить две записи и выбирать запись, получив в результате: