Spring Data R2DBC und Kotlin Coroutines

Durch die Kombination von Spring Data R2DBC und Kotlin Coroutines erhalten wir eine nahtlose und effiziente Möglichkeit, nicht-blockierende Datenbankoperationen durchzuführen. In diesem Blog-Beitrag untersuchen wir die verschiedenen Optionen für den Umgang mit nicht-blockierenden Datenbankzugriffen mit diesen beiden leistungsstarken Tools innerhalb eines reaktiven Stacks, um eine End-to-End-Funktion mit Spring WebFlux, Kotlin Coroutines und Spring Data R2DBC zu schreiben.

Einerseits ist Spring Data R2DBC ein Tool, mit dem Entwickler über ein reaktives Programmiermodell auf Datenbanken zugreifen können. Dies kann dazu beitragen, die Leistung einer Anwendung zu verbessern, insbesondere bei IO-gebundenen Aufgaben wie Datenbankzugriffen und hoher Gleichzeitigkeit.

Kotlin Coroutines hingegen ist ein Werkzeug zum Schreiben von asynchronem und nicht-blockierendem Code. Sie ermöglichen es Ihnen, Code zu schreiben, der einfach zu lesen und zu verstehen ist, aber dennoch effizient und nicht blockierend ist.

Reaktiv vs. Koroutinen

Reaktive Programmierung ist ein Programmierparadigma, das asynchrone Datenströme und Ereignisse mithilfe von Datenflüssen und der Weitergabe von Änderungen verarbeitet. Im Zusammenhang mit dem Spring-Framework ist Spring WebFlux ein reaktives Webmodul, mit dem Entwickler reaktive Webanwendungen innerhalb des Spring-Ökosystems erstellen können.

Reactor ist die vollständig nicht-blockierende Bibliothek der Wahl für Spring Reactive Web (WebFlux). Sie basiert auf der Spezifikation von Reactive Streams und bietet eine reaktive API mit Typen wie Mono oder Flux, um einen asynchron berechneten Stream von Werten zu kapseln.

Nichtsdestotrotz bietet Kotlin Coroutines ein einfaches und intuitives Programmiermodell für das Schreiben von nicht-blockierendem Code, aber in einem imperativen und sequentiellen Stil. Außerdem arbeiten Coroutines perfekt mit nicht-blockierenden Frameworks wie Spring WebFlux zusammen, indem sie die Reactive API in suspend Funktionen umwandeln und Typen als Flow für das Datenstreaming bereitstellen.

Spring Data R2DBC

Lassen Sie uns nun verstehen, wofür R2DBC steht. R2DBC ist ein Akronym für Reactive Relational Database Connectivity. Es handelt sich um eine Spezifikation, die definiert, wie ein reaktives Programmiermodell auf den Datenbankzugriff angewendet werden kann. Dies ermöglicht es Entwicklern, mit einem nicht-blockierenden Programmiermodell mit Datenbanken zu arbeiten.

Spring Data R2DBC ist eine Implementierung der R2DBC-Spezifikation, die auf Spring Data aufbaut. Sie bietet eine einfache und konsistente API für die Arbeit mit Datenbanken unter Verwendung des Spring-Ökosystems. Das bedeutet, dass Entwickler die gleichen vertrauten Spring Data-Annotationen und -Schnittstellen für die Arbeit mit Datenbanken verwenden können, jedoch mit den zusätzlichen Vorteilen eines reaktiven Stacks.

Erste Schritte

Um mit einer reaktiven Spring-Anwendung zu beginnen, müssen Sie zunächst die entsprechenden Abhängigkeiten zu Ihrem Projekt hinzufügen. Wenn Sie ein neues Projekt von Grund auf neu beginnen, könnte ein guter Ansatz darin bestehen, das Projekt von Spring Initializr aus zu erstellen und Spring Reactive Web (WebFlux), Spring Data R2DBC und einen Treiber, zum Beispiel den PostgreSQL-Treiber, auszuwählen.

Diesmal fügen wir jedoch die benötigten Abhängigkeiten manuell hinzu. Dabei fügen wir auch die erforderlichen Abhängigkeiten für Kotlin Coroutines und Erweiterungen für Reactor hinzu.

dependencies {  
   implementation("org.springframework.boot:spring-boot-starter-webflux")
   implementation("org.springframework.boot:spring-boot-starter-data-r2dbc")
   implementation("org.postgresql:r2dbc-postgresql")
   implementation("io.projectreactor.kotlin:reactor-kotlin-extensions")
   implementation("org.jetbrains.kotlinx:kotlinx-coroutines-reactor")
}

Hier importieren wir Spring WebFlux, Spring Data R2DBC, R2DBC PostgreSQL Driver und Kotlin Coroutines Bibliotheken. Indem wir außerdem die letzten Abhängigkeiten für Kotlin Coroutines und Reactor hinzufügen, können wir von der Reactor-API in den imperativen Coroutines-Code übersetzen.

Danach benötigen Sie möglicherweise eine Datenbankinstanz. In diesem Beispiel verwenden wir Docker Compose, um einen Docker-Container mit einer PostgreSQL-Datenbank zu erstellen. Zu diesem Zweck erstellen wir eine docker-compose.yml Datei innerhalb des Projekts:

services:  
  postgres:  
    image: postgres:15.0-alpine  
    hostname: postgres  
    ports:  
      - 5432:5432
    restart: always  
    environment:  
      POSTGRES_DB: testDb  
      POSTGRES_USER: someUser  
      POSTGRES_PASSWORD: somePassword

Dann führen wir einfach den Befehl docker-compose up -d im Terminal aus, um ihn im Hintergrund laufen zu lassen.

Datenbank-Konfiguration

Sobald wir die Abhängigkeiten eingerichtet haben und die Datenbank läuft, ist es an der Zeit, die Datenbankkonfiguration zu erstellen. Wir könnten die Spring YAML-Konfiguration verwenden, aber für dieses Beispiel werden wir sie programmatisch erstellen und den Standard-Port von Postgres (5432) verwenden:

@Configuration  
@EnableR2dbcRepositories  
class DatabaseConfig : AbstractR2dbcConfiguration() {  
    @Bean
    override fun connectionFactory(): ConnectionFactory =  
        PostgresqlConnectionFactory(  
            PostgresqlConnectionConfiguration.builder()  
                .host("localhost")  
                .database("testDb")  
                .username("someUser")  
                .password("somePassword")  
                .build()  
        )  
}

Zunächst verwenden wir die @Configuration Annotation von Spring, die es uns ermöglicht, Annotationen für Dependency Injection zu verwenden, während @EnableR2dbcRepositories reaktive relationale Repositories mit R2DBC aktiviert. Die Konfigurationsklasse ist eine Erweiterung von , da wir die Standardklasse mit unserer Datenbankkonfiguration außer Kraft setzen. Schließlich markieren wir sie als @Bean, damit sie später injiziert werden kann.

Innerhalb derselben Klasse DatabaseConfig werden wir auch eine ConnectionFactoryInitializer erstellen, um die Verbindung zu initialisieren, über den Konstruktor die zuvor definierte ConnectionFactory zu injizieren und die Datenbank bei der Initialisierung der Anwendung aufzufüllen:

@Bean  
fun databaseInitializer(connectionFactory: ConnectionFactory): ConnectionFactoryInitializer =  
    ConnectionFactoryInitializer().apply {  
        setConnectionFactory(connectionFactory)  
        setDatabasePopulator(  
            CompositeDatabasePopulator().apply {  
                addPopulators(  
                    ResourceDatabasePopulator(  
                        ClassPathResource("sql/schema.sql"),  
                        ClassPathResource("sql/data.sql")  
                    )  
                )  
            }  
        )  
    }

In einem resources/sql Ordner definieren wir eine schema.sql Datei zur Erstellung der Tabelle:

CREATE TABLE IF NOT EXISTS users (  
     id BIGSERIAL PRIMARY KEY,  
     username VARCHAR NOT NULL,  
     email VARCHAR NOT NULL,  
     CONSTRAINT unique_user UNIQUE (username, email)  
);

Wir definieren eine data.sql mit einigen Standard-Testdaten:

INSERT INTO users (username, email) VALUES ('test', 'user@test.com') ON CONFLICT DO NOTHING;

Wir definieren auch ein User Datentransferobjekt (DTO) und fügen die Anmerkungen für die Abbildung unseres Codes auf die Datenbank hinzu:

@Table("users")
data class User(
    @Id val id: Long,
    @Column("username") val username: String,
    @Column("email") val email: String,
)

Repository

Lassen Sie uns nun unsere verschiedenen Optionen für die nicht-blockierende Arbeit mit der Datenbank mithilfe von Kotlin Coroutines im Repository Layer untersuchen.

DatenbankClient

DatabaseClient ist ein nicht-blockierender, reaktiver Client für die Durchführung von Datenbankaufrufen mit reaktivem Stream-Rückdruck. Spring Data R2DBC bietet dafür Coroutines-Erweiterungen.

Zum Beispiel:

@Repository  
class UserRepository(private val client: DatabaseClient) {  
    suspend fun findById(id: Long): User? =  
        client  
            .sql("SELECT * FROM users WHERE id = $id")  
            .map { row ->  
                User(  
                    row.get("id") as Long,  
                    row.get("username") as String,  
                    row.get("email") as String,  
                )  
            }  
            .awaitOneOrNull()  
}

In diesem Beispiel definieren wir zunächst eine UserRepository Klasse, um Datenbankoperationen mit DatabaseClient durchzuführen. Die Funktion nimmt einen Parameter entgegen und gibt ein nullbares Objekt zurück. Innerhalb der Methode verwenden wir die , um eine SQL-Abfrage auszuführen, die den Benutzer anhand seiner aus der Datenbank auswählt. Danach verwenden wir die Methode , um die Ergebnismenge auf das Objekt User abzubilden. Schließlich verwenden wir awaitOneOrNull(), um auf das Ergebnis der Operation zu warten und geben null zurück, wenn kein Ergebnis vorliegt.

Wenn Sie mit dem reaktiven Stack von Spring vertraut sind, werden Sie feststellen, dass es nicht notwendig ist, den Wert in ein Mono zu verpacken, sondern dass findByUsername als suspend markiert ist. Das Schlüsselwort suspend wird in Kotlin verwendet, um anzuzeigen, dass es sich um eine Suspend-Funktion handelt, die nur von einer Coroutine oder einer anderen Suspend-Funktion (innerhalb eines Coroutine-Kontexts) aufgerufen werden sollte.

R2dbcEntityTemplate

Es ist auch möglich, das R2dbcEntityTemplate zu verwenden, um Operationen mit Entitäten durchzuführen. Und auch dafür bietet Spring Data R2DBC Coroutines-Erweiterungen.

Zum Beispiel:

@Repository
class UserRepository(private val template: R2dbcEntityTemplate) {
    suspend fun findById(id: Long): User? =
        template.select(User::class.java)
            .matching(
                Query.query(
                    Criteria.where("id").`is`(id)
                )
            )
            .awaitOneOrNull()
}

Dieses Mal definieren wir eine Klasse UserRepository, verwenden aber R2dbcEntityTemplate für die Datenbankoperationen. Die Funktion nimmt einen Parameter entgegen und gibt ein nullbares Objekt zurück. Innerhalb der Methode verwenden wir , um den Benutzer über aus der Datenbank auszuwählen. Die Methode gleicht den Wert des Parameters mit der Abfrage ab. Schließlich verwenden wir awaitOneOrNull(), um auf das Ergebnis des Vorgangs zu warten; sie gibt null zurück, wenn kein Ergebnis vorliegt.

Der Hauptunterschied zum vorherigen Ansatz besteht darin, dass wir diesmal keinen SQL Code schreiben und den Vorteil haben, direkt mit dem User DTO abzugleichen, anstatt jedes einzelne Feld einzeln zuzuordnen.

CoroutineCrudRepository

Das Spring Data R2DBC-Projekt enthält das Coroutine Repository, das den nicht-blockierenden Datenzugriff über die Coroutines von Kotlin ermöglicht. Um es zu verwenden, definieren Sie ein Repository, das CoroutineCrudRepository erweitert.

Zum Beispiel:

interface UserRepository : CoroutineCrudRepository<User, Long> { 
    override suspend fun findById(id: Long): User? 
}

Wir definieren eine Schnittstelle UserRepository, die CoroutineCrudRepository erweitert. Die letztgenannte Schnittstelle bietet mehrere nützliche Methoden zur Durchführung von CRUD-Operationen für User Objekte, wie save, findById und delete.

Für dieses Beispiel brauchen wir die Funktion findById nicht wirklich zu definieren, da sie bereits Teil der Schnittstelle CoroutineCrudRepository ist. Wir können sie jedoch außer Kraft setzen oder neue benutzerdefinierte Methoden definieren, falls wir verschiedene Operationen mit der Datenbank durchführen müssen.

Betrachten Sie das Schlüsselwort suspend; ohne dieses Schlüsselwort oder ohne die Rückgabe eines Typs, der die Kontextfortpflanzung ermöglicht (wie Flow), wäre der Coroutine-Kontext nicht verfügbar.

Die Vorteile dieses Ansatzes sind dieselben wie bei der Verwendung einer Spring Data Repository-Abstraktion. Dadurch wird der Umfang des Boilerplate-Codes reduziert und wir können Abfragen an die Datenbank mit Hilfe von Schlüsselwörtern durchführen, allerdings auf nicht-blockierende Weise mit Coroutines.

Service

In der Dienstebene müssen wir eine Klasse definieren, die mit @Service annotiert ist, das Repository injizieren und es von einer aussetzenden Funktion aus aufrufen:

@Service  
class UserService(val repository: CoroutineUserRepository) {  
   suspend fun findUserById(id: Long): User? = 
       repository.findById(id)
}

Da wir jedes Beispiel mit der gleichen Funktionssignatur definiert haben, können wir jedes beliebige Beispiel auswählen, das von diesem Dienst aus aufgerufen werden soll.

Controller

In der Controller-Schicht müssen wir eine Klasse definieren, die mit einer @Controller -Annotation versehen ist. In unserem Fall verwenden wir die spezialisierte Version . Im Konstruktor injizieren wir den Dienst und rufen die zuvor definierte Methode von einer aussetzenden Funktion aus auf, wenn der Endpunkt /users/{id} aufgerufen wird:

@RestController
class UserController(private val userService: UserService) {
    @GetMapping("/users/{id}")
    suspend fun getUserById(@PathVariable id: String): ResponseEntity<User> {
        val user = userService.findUserById(id)
        return if (user != null) ResponseEntity.ok(user)  
        else ResponseEntity.notFound().build()
    }
}

Wenn Sie die Funktion getUserById mit dem Schlüsselwort suspend innerhalb von @RestController kommentieren, wird WebFlux angewiesen, beim Aufruf des Endpunkts Coroutines statt der Reactive API zu verwenden.

Ausführen und Testen des Beispiels

Bei der Verwendung von Spring WebFlux konfiguriert Spring automatisch Netty als Standardserver, ein nicht blockierendes I/O-Client-Server-Framework.

Wenn wir die Beispielanwendung ausführen, sollten wir in der Konsole etwa so etwas erhalten, was uns mitteilt, dass die Anwendung läuft und funktioniert:

2023-01-25T15:26:57.692+01:00  INFO 69644 --- [           main] .s.d.r.c.RepositoryConfigurationDelegate : Finished Spring Data repository scanning in 65 ms. Found 1 R2DBC repository interfaces.
2023-01-25T15:26:58.399+01:00  INFO 69644 --- [           main] o.s.b.web.embedded.netty.NettyWebServer  : Netty started on port 8080
2023-01-25T15:26:58.403+01:00  INFO 69644 --- [           main] c.e.r2dbc.ExampleR2DBCApplicationKt      : Started ExampleR2DBCApplicationKt in 1.642 seconds (process running for 2.034)

Wenn wir nun den Endpunkt aufrufen, sollten wir etwa so etwas erhalten:

GET http://localhost:8080/users/1

HTTP/1.1 200 OK
Content-Type: application/json
Content-Length: 58

{
  "id": 1,
  "username": "testuser",
  "email": "testuser@test.com"
}

Schlussfolgerungen

Sowohl Spring Data R2DBC als auch Kotlin Coroutines sind leistungsstarke Werkzeuge für die Entwicklung asynchroner, nicht blockierender Anwendungen. Spring Data R2DBC bietet zwar eine bequeme Möglichkeit, mit relationalen Datenbanken unter Verwendung reaktiver Programmierung zu arbeiten, erfordert aber auch eine bestimmte Denkweise und Herangehensweise, wie z.B. das Einhalten reaktiver Ketten und das Abonnieren von Werten, was eine gewisse Eingewöhnungszeit erfordern kann.

Kotlin Coroutines hingegen bieten einen vertrauteren und intuitiveren Ansatz für das Schreiben von asynchronem Code. Sie ermöglichen es Entwicklern, typische Programmiermuster zu verwenden und Code zu schreiben, der so aussieht, als würde er sequentiell ausgeführt werden.

Wir haben nur einige Grundlagen dieser Konzepte erforscht und die verschiedenen Möglichkeiten aufgezeigt, wie wir mit Spring Data R2DBC und Coroutines mit der Datenbank interagieren können. Dabei haben wir gezeigt, dass es relativ einfach ist, diese in ein Spring-Projekt zu integrieren, da Bibliotheken wie kotlinx-coroutines-reactive und kotlinx-coroutines-reactor es uns ermöglichen, von der Reactive API zur Coroutines API zu konvertieren.

In Zukunft könnten wir auch andere Möglichkeiten untersuchen, die Vorteile von Coroutines im restlichen Spring-Ökosystem oder sogar in anderen Frameworks zu nutzen.

Wenn Sie Ihr Wissen in diesem Bereich erweitern möchten, empfehle ich Ihnen, die Dokumentation aller in diesem Beitrag erwähnten Tools zu lesen, um weitere Einzelheiten zu erfahren.