Использование Kotlin Coroutines с Spring

До Spring 5.2 вы можете испытать Kotlin Coroutines благодаря усилиям сообщества, например. весна-котлин-сопрограмма на Гитхабе. В Spring 5.2 появилось несколько функций, помимо стиль функционального программирования, представленный в Spring MVCЕще одной привлекательной особенностью является то, что Kotlin Coroutines наконец-то получили официальную поддержку.

Сопрограммы Kotlin предоставляют альтернативный подход к написанию асинхронных приложений со стеком Spring Reactive, но в императивном стиле кода.

В этом посте я перепишу мой реактивный образец используя Kotlin Coroutines с Spring.

Создайте проект Spring Boot, используя Весенняя инициализация.

Язык: Котлин
Версия Spring Boot: 2.2.0.BUILD-SNAPSHOT
Зависимости: веб-реактивный

Открой пом.xml файл, добавьте некоторые изменения.

Добавьте зависимости, связанные с kotlin-coroutines, в зависимости раздел.

<dependency>
    <groupId>org.jetbrains.kotlinx</groupId>
    <artifactId>kotlinx-coroutines-core</artifactId>
    <version>${kotlinx-coroutines.version}</version>
</dependency>
<dependency>
    <groupId>org.jetbrains.kotlinx</groupId>
    <artifactId>kotlinx-coroutines-reactor</artifactId>
    <version>${kotlinx-coroutines.version}</version>
</dependency>

Определите kotlin-coroutines.версия в свойствах.

<kotlinx-coroutines.version>1.2.0</kotlinx-coroutines.version>

Kotlin coroutines 1.2.0 совместим с Kotlin 1.3.30, определите kotlin.version в файле pom.xml, чтобы использовать эту версию.

<kotlin.version>1.3.30</kotlin.version>

Spring Data заняты добавлением поддержки Kotlin Coroutines. В настоящее время Spring Data R2DBC получил базовую поддержку сопрограмм в своемDatabaseClient. В этом примере мы используем Spring Data R2DBC для операций с данными.

Добавьте зависимости, связанные со Spring Data R2DBC, и используйте PostgresSQL в качестве серверной базы данных.

<dependency>
    <groupId>org.springframework.data</groupId>
    <artifactId>spring-data-r2dbc</artifactId>
    <version>${spring-data-r2dbc.version}</version>
</dependency>

<dependency>
    <groupId>org.postgresql</groupId>
    <artifactId>postgresql</artifactId>
</dependency>

<dependency>
    <groupId>io.r2dbc</groupId>
    <artifactId>r2dbc-postgresql</artifactId>
</dependency>

объявить spring-data-r2dbc.version свойство для использования последних данных Spring Data R2DBC.

 <spring-data-r2dbc.version>1.0.0.BUILD-SNAPSHOT</spring-data-r2dbc.version>

Включает поддержку данных R2dbc путем создания подклассов AbstractR2dbcConfiguration.

@Configuration
@EnableR2dbcRepositories
class DatabaseConfig : AbstractR2dbcConfiguration() {

    override fun connectionFactory(): ConnectionFactory {
        return PostgresqlConnectionFactory(
                PostgresqlConnectionConfiguration.builder()
                        .host("localhost")
                        .database("test")
                        .username("user")
                        .password("password")
                        .build()
        )
    }
}

Создайте класс данных, который сопоставляется с таблицей posts.

@Table("posts")
data class Post(@Id val id: Long? = null,
                @Column("title") val title: String? = null,
                @Column("content") val content: String? = null
)

Следовать Руководство по переводу реактивного стека в Kotlin Coroutines предоставленный в справочной документации Spring, создайте класс репозитория для Post.

@Component
class PostRepository(private val client: DatabaseClient) {

    suspend fun count(): Long =
            client.execute().sql("SELECT COUNT(*) FROM posts")
                    .asType<Long>().fetch().awaitOne()

    fun findAll(): Flow<Post> =
            client.select().from("posts").asType<Post>().fetch().flow()

    suspend fun findOne(id: Long): Post? =
            client.execute()
                    .sql("SELECT * FROM posts WHERE id = \$1")
                    .bind(0, id).asType<Post>()
                    .fetch()
                    .awaitOneOrNull()

    suspend fun deleteAll() =
            client.execute()
                    .sql("DELETE FROM posts")
                    .fetch()
                    .rowsUpdated()
                    .awaitSingle()

    suspend fun save(post: Post) =
            client.insert()
                    .into<Post>()
                    .table("posts")
                    .using(post)
                    .await()

    suspend fun init() {
        save(Post(title = "My first post title", content = "Content of my first post"))
        save(Post(title = "My second post title", content = "Content of my second post"))
    }
}

Создать @RestController класс для Post.

@RestController
@RequestMapping("/posts")
class PostController(
        private val postRepository: PostRepository
) {

    @GetMapping("")
    fun findAll(): Flow<Post> =
            postRepository.findAll()

    @GetMapping("{id}")
    suspend fun findOne(@PathVariable id: Long): Post? =
            postRepository.findOne(id) ?: throw PostNotFoundException(id)

    @PostMapping("")
    suspend fun save(@RequestBody post: Post) =
            postRepository.save(post)

    @GetMapping("{id}/comments")
    fun findCommentsByPostId(@PathVariable id: Long): Flow<Comment> =
            commentRepository.findByPostId(id)

}

Вы также можете инициализировать данные в CommandLineRunner боб или послушать @ApplicationReadyEventиспользовать runBlocking для переноса задач сопрограмм.

runBlocking {
     val deleted = postRepository.deleteAll()
     println(" $deleted posts was cleared.")
     postRepository.init()
}

Для успешного запуска приложения убедитесь, что сервер PostgreSQL работает. я подготовил файл для создания докеров чтобы просто запустить сервер PostgresSQL и инициализировать схему базы данных в док-контейнере.

docker-compose up

Запустите приложение сейчас, оно должно работать хорошо, как предыдущие Реактивные примеры.

В дополнение к аннотированным контроллерам Kotlin Coroutines также поддерживается в функциональном RouterFunction DSL с использованием coRouter для определения ваших правил маршрутизации.

@Configuration
class RouterConfiguration {

    @Bean
    fun routes(postHandler: PostHandler) = coRouter {
        "/posts".nest {
            GET("", postHandler::all)
            GET("/{id}", postHandler::get)
            POST("", postHandler::create)
            PUT("/{id}", postHandler::update)
            DELETE("/{id}", postHandler::delete)
        }
    }
}

Как и изменения в контроллере, PostHandler написано в императивном стиле.

@Component
class PostHandler(private val posts: PostRepository) {

    suspend fun all(req: ServerRequest): ServerResponse {
        return ok().bodyAndAwait(this.posts.findAll())
    }

    suspend fun create(req: ServerRequest): ServerResponse {
        val body = req.awaitBody<Post>()
        val createdPost = this.posts.save(body)
        return created(URI.create("/posts/$createdPost")).buildAndAwait()
    }

    suspend fun get(req: ServerRequest): ServerResponse {
        println("path variable::${req.pathVariable("id")}")
        val foundPost = this.posts.findOne(req.pathVariable("id").toLong())
        println("found post:$foundPost")
        return when {
            foundPost != null -> ok().bodyAndAwait(foundPost)
            else -> notFound().buildAndAwait()
        }
    }

    suspend fun update(req: ServerRequest): ServerResponse {
        val foundPost = this.posts.findOne(req.pathVariable("id").toLong())
        val body = req.awaitBody<Post>()
        return when {
            foundPost != null -> {
                this.posts.update(foundPost.copy(title = body.title, content = body.content))
                noContent().buildAndAwait()
            }
            else -> notFound().buildAndAwait()
        }
    }

    suspend fun delete(req: ServerRequest): ServerResponse {
        val deletedCount = this.posts.deleteById(req.pathVariable("id").toLong())
        println("$deletedCount posts was deleted")
        return notFound().buildAndAwait()
    }
}

Помимо аннотированных контроллеров и функционального маршрутизатора DSL, Spring WebClient также используйте Kotlin Coroutines.

@RestController
@RequestMapping("/posts")
class PostController(private val client: WebClient) {
    @GetMapping("")
    suspend fun findAll() =
            client.get()
                    .uri("/posts")
                    .accept(MediaType.APPLICATION_JSON)
                    .awaitExchange()
                    .awaitBody<Any>()


/*
    @GetMapping("")
    suspend fun findAll(): Flow<Post> =
            client.get()
                    .uri("/posts")
                    .accept(MediaType.APPLICATION_JSON)
                    .awaitExchange()
                    .awaitBody()
*/

    @GetMapping("/{id}")
    suspend fun findOne(@PathVariable id: Long): PostDetails = withDetails(id)


    private suspend fun withDetails(id: Long): PostDetails {
        val post =
                client.get().uri("/posts/$id")
                        .accept(APPLICATION_JSON)
                        .awaitExchange().awaitBody<Post>()

        val count =
                client.get().uri("/posts/$id/comments/count")
                        .accept(APPLICATION_JSON)
                        .awaitExchange().awaitBody<Long>()

        return PostDetails(post, count)
    }

}

в withDetails метод, отправка и подсчет вызовов удаленных API один за другим в последовательности.

Если вы хотите выполнять сопрограммы параллельно, используйте async контекст, чтобы обернуть все вызовы и поместить все задачи в coroutineScope контекст. В PostDetailsчтобы построить результаты, используйте await дождаться завершения удаленных вызовов.

private suspend fun withDetails(id: Long): PostDetails = coroutineScope {
        val post = async {
            client.get().uri("/posts/$id")
                    .accept(APPLICATION_JSON)
                    .awaitExchange().awaitBody<Post>()
        }
        val count = async {
            client.get().uri("/posts/$id/comments/count")
                    .accept(APPLICATION_JSON)
                    .awaitExchange().awaitBody<Long>()
        }
        PostDetails(post.await(), count.await())
}

Проверьте коды с Гитхаба.