當前位置：首頁 > 编程语言 > java >内容正文

java

[译] RxJava JDBC 简介

發布時間：2025/3/20 java 31 豆豆

生活随笔收集整理的這篇文章主要介紹了 [译] RxJava JDBC 简介小編覺得挺不錯的,現在分享給大家,幫大家做個參考.

www.baeldung.com/rxjava-jdbc
作者：baeldung
譯者：oopsguy.com

1、概述

簡單地說，rxjava-jdbc 是一個用于與關系數據庫交互的 API，其允許以鏈式的方式調用。在此快速教程中，我們將來了解這個類庫，以及如何使用它的一些常用功能。

在閱讀本教程之前，你需要有一定的 RxJava 基礎知識。

2、Maven 依賴

從 Maven 依賴開始，我們需要把依賴添加到 pom.xml 文件中：

<dependency><groupId>com.github.davidmoten</groupId><artifactId>rxjava-jdbc</artifactId><version>0.7.11</version> </dependency>復制代碼

我們可以在 Maven Central 上找到最新版本的 API。

3、主要組件

Database 類是運行所有常見類型數據庫交互的主入口點。要創建一個 Database 對象，我們可以將 ConnectionProvider 接口實現實例傳遞給 from() 靜態方法：

public static ConnectionProvider connectionProvider= new ConnectionProviderFromUrl(DB_CONNECTION, DB_USER, DB_PASSWORD); Database db = Database.from(connectionProvider);復制代碼

ConnectionProvider 有幾個值得注意的實現 — 例如 ConnectionProviderFromContext、ConnectionProviderFromDataSource、ConnectionProviderFromUrl 和 ConnectionProviderPooled。

為了做些基本操作，我們可以使用以下 Database 的 API：

select() — 用于 SQL select 查詢
update() — 用于 DDL 語句，如 create 和 drop，以及 insert、update 和 delete

4、啟動

在下面的快速示例中，我們將展示基本的數據庫操作：

public class BasicQueryTypesTest {Observable<Integer> create,insert1, insert2, insert3, update, delete = null;@Testpublic void whenCreateTableAndInsertRecords_thenCorrect() {create = db.update("CREATE TABLE IF NOT EXISTS EMPLOYEE("+ "id int primary key, name varchar(255))").count();insert1 = db.update("INSERT INTO EMPLOYEE(id, name) VALUES(1, 'John')").dependsOn(create).count();update = db.update("UPDATE EMPLOYEE SET name = 'Alan' WHERE id = 1").dependsOn(create).count();insert2 = db.update("INSERT INTO EMPLOYEE(id, name) VALUES(2, 'Sarah')").dependsOn(create).count();insert3 = db.update("INSERT INTO EMPLOYEE(id, name) VALUES(3, 'Mike')").dependsOn(create).count();delete = db.update("DELETE FROM EMPLOYEE WHERE id = 2").dependsOn(create).count();List<String> names = db.select("select name from EMPLOYEE where id < ?").parameter(3).dependsOn(create).dependsOn(insert1).dependsOn(insert2).dependsOn(insert3).dependsOn(update).dependsOn(delete).getAs(String.class).toList().toBlocking().single();assertEquals(Arrays.asList("Alan"), names);} }復制代碼

這里有一點需要注意 — 我們調用 dependsOn() 來決定查詢的運行順序。

否則，代碼將失敗或產生不可預測的結果，除非我們以一定的順序指定要執行的查詢。

5、自動映射

自動映射功能允許我們將指定的數據庫記錄映射到對象。

我們來看看兩種數據庫記錄的自動映射方法。

5.1、使用接口自動映射

我們可以使用帶注解的接口將數據庫記錄 automap() 到對象。為此，我們可以創建一個帶注解的接口：

public interface Employee {@Column("id")int id();@Column("name")String name();}復制代碼

之后，運行測試：

@Test public void whenSelectFromTableAndAutomap_thenCorrect() {List<Employee> employees = db.select("select id, name from EMPLOYEE").dependsOn(create).dependsOn(insert1).dependsOn(insert2).autoMap(Employee.class).toList().toBlocking().single();assertThat(employees.get(0).id()).isEqualTo(1);assertThat(employees.get(0).name()).isEqualTo("Alan");assertThat(employees.get(1).id()).isEqualTo(2);assertThat(employees.get(1).name()).isEqualTo("Sarah"); }復制代碼

5.2、使用類自動映射

我們還可以使用具體的類來將數據庫記錄自動映射到對象。讓我們看看該類的寫法：

public class Manager {private int id;private String name;// standard constructors, getters, and setters }復制代碼

運行測試：

@Test public void whenSelectManagersAndAutomap_thenCorrect() {List<Manager> managers = db.select("select id, name from MANAGER").dependsOn(create).dependsOn(insert1).dependsOn(insert2).autoMap(Manager.class).toList().toBlocking().single();assertThat(managers.get(0).getId()).isEqualTo(1);assertThat(managers.get(0).getName()).isEqualTo("Alan");assertThat(managers.get(1).getId()).isEqualTo(2);assertThat(managers.get(1).getName()).isEqualTo("Sarah"); }復制代碼

這里有幾個要點：

create、insert1 和 insert2 引用了在創建 Manager 表并將記錄插入其中返回的 Observables
我們查詢中所 select 的列的數量必須與 Manager 類構造方法中的參數數量相匹配
列必須是可以自動映射到構造方法中的類型

有關自動映射的更多信息，請訪問 GitHub 上的 rxjava-jdbc 倉庫

6、使用大對象

API 支持使用大對象（如 CLOB 和 BLOBS）。在接下來的小節中，我們將介紹如何利用這一功能。

6.1、CLOB

我們來如何 insert 和 select 一個 CLOB：

@Before public void setup() throws IOException {create = db.update("CREATE TABLE IF NOT EXISTS " + "SERVERLOG (id int primary key, document CLOB)").count();InputStream actualInputStream= new FileInputStream("src/test/resources/actual_clob");actualDocument = getStringFromInputStream(actualInputStream);InputStream expectedInputStream = new FileInputStream("src/test/resources/expected_clob");expectedDocument = getStringFromInputStream(expectedInputStream);insert = db.update("insert into SERVERLOG(id,document) values(?,?)").parameter(1).parameter(Database.toSentinelIfNull(actualDocument)).dependsOn(create).count(); }@Test public void whenSelectCLOB_thenCorrect() throws IOException {db.select("select document from SERVERLOG where id = 1").dependsOn(create).dependsOn(insert).getAs(String.class).toList().toBlocking().single();assertEquals(expectedDocument, actualDocument); }復制代碼

請注意，getStringFromInputStream() 是將 InputStream 的內容轉換為 String。

6.2、BLOB

我們可以使用 API 以類似的方式使用 BLOB。唯一的區別是，我們不必傳遞一個 String 到 toSentinelIfNull() 方法，而是傳遞一個字節數組。

我們可以這樣做：

@Before public void setup() throws IOException {create = db.update("CREATE TABLE IF NOT EXISTS "+ "SERVERLOG (id int primary key, document BLOB)").count();InputStream actualInputStream= new FileInputStream("src/test/resources/actual_clob");actualDocument = getStringFromInputStream(actualInputStream);byte[] bytes = this.actualDocument.getBytes(StandardCharsets.UTF_8);InputStream expectedInputStream = new FileInputStream("src/test/resources/expected_clob");expectedDocument = getStringFromInputStream(expectedInputStream);insert = db.update("insert into SERVERLOG(id,document) values(?,?)").parameter(1).parameter(Database.toSentinelIfNull(bytes)).dependsOn(create).count(); }復制代碼

之后，我們可以在之前的例子中服用同樣的測試。

7、事務

接下來，我們來看看事務支持。

事務管理允許我們處理在單個事務中分組多個數據庫操作的事務，以便它們都能被提交 — 永久保存到數據庫中，或者完全回滾。

我們來看一個快速示例：

@Test public void whenCommitTransaction_thenRecordUpdated() {Observable<Boolean> begin = db.beginTransaction();Observable<Integer> createStatement = db.update("CREATE TABLE IF NOT EXISTS EMPLOYEE(id int primary key, name varchar(255))").dependsOn(begin).count();Observable<Integer> insertStatement = db.update("INSERT INTO EMPLOYEE(id, name) VALUES(1, 'John')").dependsOn(createStatement).count();Observable<Integer> updateStatement = db.update("UPDATE EMPLOYEE SET name = 'Tom' WHERE id = 1").dependsOn(insertStatement).count();Observable<Boolean> commit = db.commit(updateStatement);String name = db.select("select name from EMPLOYEE WHERE id = 1").dependsOn(commit).getAs(String.class).toBlocking().single();assertEquals("Tom", name); }復制代碼

我們調用 beginTransaction() 方法來開始一個事務。調用此方法后，每個數據庫操作都將在同一個事務中運行，直到調用 commit() 或 rollback() 方法為止。

我們可以使用 rollback() 方法捕獲 Exception 來回滾整個事務，以防代碼由于某些原因而失敗。我們可以為所有 Exception 或指定 Exception 進行此操作。

8、返回生成的 key

如果我們在處理的表中設置了 auto_increment 字段，我們可能需要檢索生成的值。我們可以通過調用 returnGeneratedKeys() 方法來實現。