Giới thiệu về Reactive Relational Database Connectivity (R2DBC)

Bài viết được sự cho phép của tác giả Nguyễn Hữu Khanh

Lúc trước, khi làm việc với các ứng dụng với cơ chế Reactive có sử dụng database, chúng ta thường sẽ sử dụng MongoDB database bởi vì lúc đó rất ít database hỗ trợ cơ chế Reactive ngoại trừ MongoDB. Nhưng hiện tại thì các bạn không cần nhất thiết phải sử dụng MongoDB nữa, Reactive Relational Database Connectivity (R2DBC) sẽ giúp chúng ta làm việc với nhiều relational database system khác theo cơ chế Reactive. R2DBC là một spec định nghĩa cách thức chúng ta sẽ làm việc với các relational database như MySQL, PostgreSQL, … theo cơ chế Reactive là như thế nào. Nó cung cấp cho chúng ta một bộ thư viện Service Provider Interface (SPI) giúp chúng ta có thể implement driver cho từng loại database system tương ứng. Có nhiều driver đã hiện thực bộ SPI này, ví dụ như R2DBC PostgreSQL hiện thực R2DBC cho PostgreSQL database, tương tự chúng ta cũng có R2DBC MySQL hỗ trợ cho MySQL database, … Trong bài viết này, mình sẽ giới thiệu với các bạn về R2DBC với một ví dụ về cách sử dụng R2DBC PostgreSQL driver để thao tác với PostgreSQL database các bạn nhé!

Các bạn có thể xem detail về nội dung specs mà R2DBC định nghĩa ở đây.

Đầu tiên, mình sẽ tạo mới một Maven project:

với R2DBC PostgreSQL driver dependency như sau:

<dependency>
    <groupId>io.r2dbc</groupId>
    <artifactId>r2dbc-postgresql</artifactId>
    <version>0.8.6.RELEASE</version>
</dependency>

Tương tự như khi các bạn làm việc với JDBC, điều đầu tiên chúng ta cần làm việc với một database bất kỳ là setup connection tới nó. Với R2DBC, chúng ta sẽ sử dụng đối tượng ConnectionFactory để làm điều này.

Có 2 cách để chúng ta có thể làm được điều này trong R2DBC đó là:

• Sử dụng connection URL
• Sử dụng trực tiếp code Java

Với connection URL thì các bạn có thể sử dụng URL sau cho PostgreSQL database (các database system khác thì connection URL sẽ khác một xí các bạn nhé!):

r2dbc:postgresql://<username>:<password>@<host>:<port>/<database>

Trong đó:

• host là PostgreSQL server,
• port là port PostgreSQL đang chạy. Port mặc định của PostgreSQL là 5432 nha các bạn!
• database là tên database mà chúng ta sẽ làm việc.
• username là user đăng nhập vào PostgreSQL database
• password là mật khẩu của user

Sau đó thì sử dụng đối tượng ConnectionFactories để get đối tượng ConnectionFactory từ connection URL này:

ConnectionFactory connectionFactory = ConnectionFactories
    .get("r2dbc:postgresql://khanh:123456@localhost:5432/test");

Nếu các bạn sử dụng code Java thì có thể sử dụng đối tượng Builder trong class ConnectionFactoryOptions để làm điều này:

ConnectionFactoryOptions options = ConnectionFactoryOptions.builder()
    .option(ConnectionFactoryOptions.DRIVER, "postgresql")
    .option(ConnectionFactoryOptions.HOST, "localhost")
    .option(ConnectionFactoryOptions.USER, "postgre")
    .option(ConnectionFactoryOptions.PASSWORD, "123456")
    .option(ConnectionFactoryOptions.DATABASE, "test").build();

ConnectionFactory connectionFactory = ConnectionFactories.get(options);

Sau khi đã có đối tượng ConnectionFactory xong, chúng ta có thể tạo một Publisher các connection tới database bằng cách gọi:

Publisher<? extends Connection> publisher = connectionFactory.create();

Khởi tạo đối tượng Flux của Project Reactor để subcribe đối tượng Publisher này:

Flux<Connection> flux = Flux.from(publisher);

Bây giờ thì chúng ta có thể sử dụng connection từ đối tượng Mono này để thao tác với database rồi, ví dụ như:

flux.flatMap(connection -> connection.createStatement("<sql_script_here>").execute());

Ở đây, chúng ta sử dụng phương thức createStatement() để truyền vào câu query mà các bạn muốn sử dụng để đối tượng Connection execute và trả về kết quả.

Giờ ví dụ trong database test của PostgreSQL database server của mình có một table tên là student với các column name và address như sau:

CREATE TABLE student (
    name VARCHAR(50),
    address VARCHAR(100)
)

Để insert một record mới vào table này với R2DBC, mình sẽ viết code như sau:

flux.flatMap(connection -> connection.createStatement("INSERT INTO student (name, address) VALUES ($1, $2)")
    .bind("$1", "Khanh")
    .bind("$2", "Ho Chi Minh")
    .execute());

Như các bạn thấy, chúng ta có thể bind giá trị mà chúng ta muốn vào câu query sử dụng phương thức bind() của đối tượng Statement và sau khi đã bind dữ liệu xong, chúng ta có thể gọi phương thức execute() của đối tượng Statement này để chạy câu lệnh SQL.

Xem thêm việc làm React cho SV mới ra trường tại đây

Kết quả sau khi execute câu lệnh SQL sẽ được trả về trong đối tượng Result của R2DBC, chúng ta phải consume đối tượng Result này thì câu lệnh SQL mới fully execute được. Do đó, chúng ta cần viết thêm code như sau:

flux.flatMap(connection -> connection.createStatement("INSERT INTO student (name, address) VALUES ($1, $2)")
        .bind("$1", "Khanh")
        .bind("$2", "Ho Chi Minh")
        .execute());
    .flatMap(Result::getRowsUpdated)

Phương thức getRowsUpdated() sẽ trả về số record được cập nhập sau khi execute câu lệnh SQL. Đối tượng Result còn chứa một phương thức khác tên là map() giúp chúng ta có thể lấy được data trả về từ database, với câu lệnh SELECT chẳng hạn.

Chúng ta cần gọi phương thức subscribe() để subcribe vào Publisher.

Toàn bộ code có thể viết lại như sau:

package com.huongdanjava.r2dbcpostgresql;

import io.r2dbc.spi.ConnectionFactories;
import io.r2dbc.spi.ConnectionFactory;
import io.r2dbc.spi.Result;
import reactor.core.publisher.Flux;

public class Application {

public static void main(String[] args) throws InterruptedException {
ConnectionFactory connectionFactory = ConnectionFactories
     .get("r2dbc:postgresql://khanh:123456@localhost:5432/test");

Flux.from(connectionFactory.create())
.flatMap(connection -> connection.createStatement("INSERT INTO student (name, address) VALUES ($1, $2)")
      .bind("$1", "khanh")
      .bind("$2", "Ho Chi Minh")
      .execute())
.flatMap(Result::getRowsUpdated)
.doOnNext(s -> System.out.println(s))
.subscribe();

Thread.sleep(5000);
}

}

Ở đây, mình có sử dụng thêm phương thức doOnNext() để lấy về thông tin số record được cập nhập trong database. Và vì chúng ta cần đợi cho chương trình chạy hoàn thành rồi mới exit nên mình cũng sử dụng thêm Thread.sleep() 5 giây để làm việc này.

Kết quả khi chạy đoạn code trên như sau:

Check database:

Bài viết gốc được đăng tải tại huongdanjava.com

Có thể bạn quan tâm:

• Giới thiệu về Reactive Programing trong javascript
• Giải thích Javascript Reactivity
• Entity-relationship Model

Xem thêm Việc làm database, tuyển dụng IT hấp dẫn trên TopDev