协程

Kotlin 协程是Kotlin的轻量级线程，允许以命令方式编写非阻塞代码。在语言方面，挂起函数提供了异步操作的抽象，而在库方面，kotlinx.coroutines提供了诸如async { }和Flow等函数。

Spring Framework在以下范围提供了对协程的支持：

Deferred和Flow返回值在Spring MVC和WebFlux注解的@Controller中的支持
在Spring MVC和WebFlux注解的@Controller中支持挂起函数
WebFlux client和server功能API的扩展。
WebFlux.fn coRouter { } DSL
WebFlux CoWebFilter
RSocket @MessageMapping注解方法中的挂起函数和Flow支持
对RSocketRequester的扩展
Spring AOP

依赖关系

当kotlinx-coroutines-core和kotlinx-coroutines-reactor依赖项在类路径中时，协程支持将被启用：

build.gradle.kts

dependencies {

	implementation("org.jetbrains.kotlinx:kotlinx-coroutines-core:${coroutinesVersion}")
	implementation("org.jetbrains.kotlinx:kotlinx-coroutines-reactor:${coroutinesVersion}")
}

支持版本1.4.0及以上。

响应式如何转换为协程？

对于返回值，从响应式到协程API的转换如下：

fun handler(): Mono<Void>变为suspend fun handler()
fun handler(): Mono<T>变为suspend fun handler(): T或suspend fun handler(): T?取决于Mono是否可以为空（具有更静态类型的优势）
fun handler(): Flux<T>变为fun handler(): Flow<T>

对于输入参数：

如果不需要延迟加载，fun handler(mono: Mono<T>)变为fun handler(value: T)，因为可以调用挂起函数来获取值参数。
如果需要延迟加载，fun handler(mono: Mono<T>)变为fun handler(supplier: suspend () → T)或fun handler(supplier: suspend () → T?)

Flow是协程世界中的Flux等效物，适用于热流或冷流、有限或无限流，主要区别如下：

Flow是推送式的，而Flux是推拉混合式的
背压通过挂起函数实现
Flow只有一个单个挂起的collect方法，操作符作为扩展实现
由于协程的支持，操作符易于实现
扩展允许向Flow添加自定义操作符
收集操作是挂起函数
map操作符支持异步操作（不需要flatMap），因为它接受一个挂起函数参数

阅读这篇关于使用Spring、协程和Kotlin Flow进行响应式编程的博客文章，了解更多细节，包括如何使用协程并发运行代码。

控制器

这是一个协程@RestController的示例。

@RestController
class CoroutinesRestController(client: WebClient, banner: Banner) {

	@GetMapping("/suspend")
	suspend fun suspendingEndpoint(): Banner {
		delay(10)
		return banner
	}

	@GetMapping("/flow")
	fun flowEndpoint() = flow {
		delay(10)
		emit(banner)
		delay(10)
		emit(banner)
	}

	@GetMapping("/deferred")
	fun deferredEndpoint() = GlobalScope.async {
		delay(10)
		banner
	}

	@GetMapping("/sequential")
	suspend fun sequential(): List<Banner> {
		val banner1 = client
				.get()
				.uri("/suspend")
				.accept(MediaType.APPLICATION_JSON)
				.awaitExchange()
				.awaitBody<Banner>()
		val banner2 = client
				.get()
				.uri("/suspend")
				.accept(MediaType.APPLICATION_JSON)
				.awaitExchange()
				.awaitBody<Banner>()
		return listOf(banner1, banner2)
	}

	@GetMapping("/parallel")
	suspend fun parallel(): List<Banner> = coroutineScope {
		val deferredBanner1: Deferred<Banner> = async {
			client
					.get()
					.uri("/suspend")
					.accept(MediaType.APPLICATION_JSON)
					.awaitExchange()
					.awaitBody<Banner>()
		}
		val deferredBanner2: Deferred<Banner> = async {
			client
					.get()
					.uri("/suspend")
					.accept(MediaType.APPLICATION_JSON)
					.awaitExchange()
					.awaitBody<Banner>()
		}
		listOf(deferredBanner1.await(), deferredBanner2.await())
	}

	@GetMapping("/error")
	suspend fun error() {
		throw IllegalStateException()
	}

	@GetMapping("/cancel")
	suspend fun cancel() {
		throw CancellationException()
	}

}

使用@Controller进行视图渲染也受支持。

@Controller
class CoroutinesViewController(banner: Banner) {

	@GetMapping("/")
	suspend fun render(model: Model): String {
		delay(10)
		model["banner"] = banner
		return "index"
	}
}

WebFlux.fn

这是通过coRouter { } DSL定义的协程路由器示例以及相关处理程序。

@Configuration
class RouterConfiguration {

	@Bean
	fun mainRouter(userHandler: UserHandler) = coRouter {
		GET("/", userHandler::listView)
		GET("/api/user", userHandler::listApi)
	}
}

class UserHandler(builder: WebClient.Builder) {

	private val client = builder.baseUrl("...").build()

	suspend fun listView(request: ServerRequest): ServerResponse =
			ServerResponse.ok().renderAndAwait("users", mapOf("users" to
			client.get().uri("...").awaitExchange().awaitBody<User>()))

	suspend fun listApi(request: ServerRequest): ServerResponse =
				ServerResponse.ok().contentType(MediaType.APPLICATION_JSON).bodyAndAwait(
				client.get().uri("...").awaitExchange().awaitBody<User>())
}

事务

协程上的事务通过Spring Framework 5.2提供的响应式事务管理的编程变体得到支持。

对于挂起函数，提供了TransactionalOperator.executeAndAwait扩展。

import org.springframework.transaction.reactive.executeAndAwait

class PersonRepository(private val operator: TransactionalOperator) {

    suspend fun initDatabase() = operator.executeAndAwait {
        insertPerson1()
        insertPerson2()
    }

    private suspend fun insertPerson1() {
        // 插入SQL语句
    }

    private suspend fun insertPerson2() {
        // 插入SQL语句
    }
}

对于Kotlin的Flow，提供了Flow<T>.transactional扩展。

import org.springframework.transaction.reactive.transactional

class PersonRepository(private val operator: TransactionalOperator) {

    fun updatePeople() = findPeople().map(::updatePerson).transactional(operator)

    private fun findPeople(): Flow<Person> {
        // 查询SQL语句
    }

    private suspend fun updatePerson(person: Person): Person {
        // 更新SQL语句
    }
}