使用 Kotlin + WebFlux/RxJava 2 实现响应式以及尝试正式版本的协程
在前一篇文章《使用 Kotlin + Spring Boot 进行后端开发》中，曾介绍过尝试使用 Kotlin 来作后端开发。这一次，尝试 WebFlux 以及协程。html
首先，在build.gradle中添加插件和依赖的库。java
plugins {
id 'java'
id 'org.jetbrains.kotlin.jvm' version '1.3.10'
id "org.jetbrains.kotlin.plugin.allopen" version "1.3.10"
}
ext {
libraries = [
rxjava : "2.2.2",
logback : "1.2.3",
spring_boot : "2.1.0.RELEASE",
kotlinx_coroutines_core : "1.0.1"
]
}
group 'com.kotlin.tutorial'
version '1.0-SNAPSHOT'
sourceCompatibility = 1.8
def libs = rootProject.ext.libraries // 库
repositories {
mavenCentral()
}
dependencies {
compile "org.jetbrains.kotlin:kotlin-stdlib-jdk8"
compile "org.jetbrains.kotlin:kotlin-reflect:1.3.10"
testCompile group: 'junit', name: 'junit', version: '4.12'
implementation "io.reactivex.rxjava2:rxjava:${libs.rxjava}"
implementation "org.jetbrains.kotlinx:kotlinx-coroutines-core:${libs.kotlinx_coroutines_core}"
implementation "org.jetbrains.kotlinx:kotlinx-coroutines-rx2:${libs.kotlinx_coroutines_core}"
implementation "ch.qos.logback:logback-classic:${libs.logback}"
implementation "ch.qos.logback:logback-core:${libs.logback}"
implementation "ch.qos.logback:logback-access:${libs.logback}"
implementation "org.springframework.boot:spring-boot-starter-web:${libs.spring_boot}"
implementation "org.springframework.boot:spring-boot-starter-data-mongodb-reactive:${libs.spring_boot}"
}
compileKotlin {
kotlinOptions.jvmTarget = "1.8"
}
compileTestKotlin {
kotlinOptions.jvmTarget = "1.8"
}
复制代码
这次，使用了 allopen 插件。它是官方提供的插件详见：kotlinlang.org/docs/refere…react
Kotlin 的类默认是final的，通常须要使用
open关键字。使用了allopen插件就能够节省open关键字。值得注意的是，须要打开 Intellij 的 Enable annotation processing 选项。git
这样，建立 SpringKotlinApplication 就不须要使用open：github
import org.springframework.boot.SpringApplication
import org.springframework.boot.autoconfigure.SpringBootApplication
/** * Created by tony on 2018/11/13. */
@SpringBootApplication
class SpringKotlinApplication
fun main(args: Array<String>) {
SpringApplication.run(SpringKotlinApplication::class.java, *args)
}
复制代码
另外，不要忘记配置数据库的信息，例子采用的是 MongoDB。web
WebFlux
WebFlux 是 Spring 5 新增的特性，相对于传统 MVC 的同步阻塞IO模型，它采用异步非阻塞的IO模型。spring
WebFlux 的 Flux 取自于 Reactor 中的类 Flux。Reactor 是 Spring 5 响应式开发的基础。mongodb
Reactor 是彻底基于响应式流规范设计和实现的库，Flux 和 Mono 是 Reactor 中的两个基本概念。数据库
Flux 相似 RxJava 的 Observable，它能够触发零到多个事件，并根据实际状况结束处理或触发错误。Mono 最多只触发一个事件，它跟 RxJava 的 Single 和 Maybe 相似，因此能够把 Mono 用于在异步任务完成时发出通知。编程
1.1 建立 Model
首先，建立几个 Model 类。
User 表示用户对象。
import org.springframework.data.annotation.Id
/** * Created by tony on 2018/11/22. */
data class User(@Id val id: String? = null, val name: String, val age: Int, val address: Address) {
constructor() : this(null, "", 0, Address())
constructor(name: String, age: Int, address: Address) : this(null, name = name, age = age, address = address)
}
复制代码
Address 记录用户的地址。
import org.springframework.data.annotation.Id
/** * Created by tony on 2018/11/22. */
data class Address(@Id val id: String? = null, val number: Int, val street: String, val city: String) {
constructor() : this(null, 0, "", "")
constructor(number: Int, street: String, city: String) : this(null, number, street, city)
}
复制代码
Audit 用于记录用户操做的时间。
import org.springframework.data.annotation.Id
import java.time.LocalDateTime
/** * Created by tony on 2018/11/22. */
data class Audit(@Id val id: String? = null, val name: String, val eventDate: LocalDateTime) {
constructor() : this(null, "",LocalDateTime.now())
constructor(name: String, eventDate: LocalDateTime) : this(null, name, eventDate)
}
复制代码
1.2 建立 Repository
建立 UserReactiveRepository 用于 User 对象的查询操做，它实现 ReactiveMongoRepository 接口。
import com.kotlin.tutorial.model.User
import org.springframework.data.mongodb.repository.ReactiveMongoRepository
import org.springframework.stereotype.Repository
import reactor.core.publisher.Flux
/** * Created by tony on 2018/11/22. */
@Repository
interface UserReactiveRepository : ReactiveMongoRepository<User, String> {
fun findUserByAge(age: Int): Flux<User>
fun findUserByAddressCity(city: String): Flux<User>
fun findUserByAgeAndAddressCity(age: Int, city: String): Flux<User>
}
复制代码
建立 AuditRepository 用于查询用户最近一条的操做时间。
import com.kotlin.tutorial.model.Audit
import org.springframework.data.repository.CrudRepository
import org.springframework.stereotype.Repository
/** * Created by tony on 2018/11/22. */
@Repository
interface AuditRepository: CrudRepository<Audit, String> {
fun findFirstByNameOrderByEventDateDesc(name: String): Audit
}
复制代码
1.3 建立 Service
建立 UserReactiveService，经过依赖注入了 userRepository、auditRepository。
import com.kotlin.tutorial.Utils.toLower
import com.kotlin.tutorial.model.Address
import com.kotlin.tutorial.model.Audit
import com.kotlin.tutorial.model.User
import com.kotlin.tutorial.repository.AuditRepository
import com.kotlin.tutorial.repository.UserReactiveRepository
import org.springframework.beans.factory.annotation.Autowired
import org.springframework.stereotype.Component
import reactor.core.publisher.Flux
import java.time.LocalDateTime
/** * Created by tony on 2018/11/22. */
@Component
class UserReactiveService {
@Autowired
lateinit var userRepository: UserReactiveRepository
@Autowired
lateinit var auditRepository: AuditRepository
companion object {
val cities = listOf("Shanghai", "Suzhou", "Hangzhou").toLower()
val streets = listOf("renming road", "zhongshan road").toLower()
}
fun find(age: Int?, rawCity: String?): Flux<User> {
val city = rawCity?.toLowerCase()
return when {
age is Int && city is String -> userRepository.findUserByAgeAndAddressCity(age, city)
city is String -> userRepository.findUserByAddressCity(city)
age is Int -> userRepository.findUserByAge(age)
else -> userRepository.findAll()
}
}
fun generateData(): Flux<User> {
val list = listOf(20, 25, 33, 28, 34).map {
val u = generate(it)
auditRepository.save(Audit(u.name, LocalDateTime.now()))
u
}
return userRepository.deleteAll().thenMany(userRepository.saveAll(list))
}
private fun generate(age: Int): User {
val address = Address(age, streets[age % streets.size], cities[age % cities.size])
return User("Tony$age", age, address)
}
}
复制代码
1.4 建立 Controller
建立 UserController 编写两个 reactive 的接口：
@RestController
@RequestMapping("/user")
class UserController {
@Autowired
lateinit var userReactiveService: UserReactiveService
@GetMapping("/reactive/find")
fun findByReactive(@RequestParam age: Int?, @RequestParam city: String?) = userReactiveService.find(age, city)
@GetMapping("/reactive/generate")
fun genDataByReactive() = userReactiveService.generateData()
......
}
复制代码
建立用户的方式：
curl http://localhost:8080/user/reactive/generate
复制代码
基于城市查询用户的方式：
curl http://localhost:8080/user/reactive/find?city=suzhou
复制代码
RxJava 2
RxJava 库是 JVM 上响应式编程的先驱，也是响应式流规范(Reactive Streams)的基础。
若是对 RxJava 2 不熟悉，也能够购买个人《RxJava 2.x 实战》
2.1 建立 Repository
建立 UserRxJavaRepository 功能跟 UserReactiveRepository 同样，只是多了一个 findUserByName() 方法。
import com.kotlin.tutorial.model.User
import io.reactivex.Flowable
import org.springframework.data.repository.reactive.RxJava2CrudRepository
import org.springframework.stereotype.Repository
/** * Created by tony on 2018/11/22. */
@Repository
interface UserRxJavaRepository : RxJava2CrudRepository<User, String> {
fun findUserByName(name: String): Flowable<User>
fun findUserByAge(age: Int): Flowable<User>
fun findUserByAddressCity(city: String): Flowable<User>
fun findUserByAgeAndAddressCity(age: Int, city: String): Flowable<User>
}
复制代码
2.2 建立 JavaService
建立 UserRxJavaService ，相似于 UserReactiveService。可是，多了两个方法：findByName()、login()。其中，调用 login() 会添加一条审计的记录。
import com.kotlin.tutorial.Utils.toLower
import com.kotlin.tutorial.model.Address
import com.kotlin.tutorial.model.Audit
import com.kotlin.tutorial.model.User
import com.kotlin.tutorial.repository.AuditRepository
import com.kotlin.tutorial.repository.UserRxJavaRepository
import io.reactivex.Flowable
import org.springframework.beans.factory.annotation.Autowired
import org.springframework.stereotype.Component
import java.time.LocalDateTime
/** * Created by tony on 2018/11/22. */
@Component
class UserRxJavaService {
@Autowired
lateinit var userRepository: UserRxJavaRepository
@Autowired
lateinit var auditRepository: AuditRepository
companion object {
val cities = listOf("Shanghai", "Suzhou", "Hangzhou").toLower()
val streets = listOf("renming road", "zhongshan road").toLower()
}
fun findByName(name: String): Flowable<User> = userRepository.findUserByName(name)
fun find(age: Int?, rawCity: String?): Flowable<User> {
val city = rawCity?.toLowerCase()
return when {
age is Int && city is String -> userRepository.findUserByAgeAndAddressCity(age, city)
city is String -> userRepository.findUserByAddressCity(city)
age is Int -> userRepository.findUserByAge(age)
else -> userRepository.findAll()
}
}
fun generateData(): Flowable<User> {
val list = listOf(20, 25, 33, 28, 34).map {
val u = generate(it)
auditRepository.save(Audit(u.name, LocalDateTime.now()))
u
}
return userRepository.deleteAll().andThen(userRepository.saveAll(list))
}
private fun generate(age: Int): User {
val address = Address(age, streets[age % streets.size], cities[age % cities.size])
return User("Tony$age", age, address)
}
fun login(name: String) =
userRepository.findUserByName(name)
.map {
auditRepository.save(Audit(it.name, LocalDateTime.now()))
}
}
复制代码
2.3 建立 Controller
在原有的 UserController 中新增两个 rxjava 的接口：
@RestController
@RequestMapping("/user")
class UserController {
@Autowired
lateinit var userRxJavaService: UserRxJavaService
@GetMapping("/rxjava/find")
fun findByRx(@RequestParam age: Int?, @RequestParam city: String?) = userRxJavaService.find(age, city)
@GetMapping("/rxjava/generate")
fun genDateByRx() = userRxJavaService.generateData()
......
}
复制代码
Kotlin 1.3 的 Coroutines
协程(coroutine)相比于线程更加轻量级，协程又称为微线程。线程和协程的一个显著区别是，线程的阻塞代价是昂贵的，而协程使用了更简单、代价更小的挂起(suspend)来代替阻塞。
Coroutines 是 Kotlin 1.1 增长的实验的功能，到 Kotlin 1.3 已经变成了正式的功能。
先在 UserController 建立一个模拟登录的接口，访问该接口时会添加一条审计的记录
@GetMapping("/rxjava/login")
fun mockLogin(@RequestParam username: String) = userRxJavaService.login(username)
复制代码
而后尝试用传统的 blocking 方式来编写一个获取登录信息的接口：
@GetMapping("/blocking/{username}")
fun getNormalLoginMessage(@PathVariable username: String):String {
val user = userService.findByName(username)
val lastLoginTime = auditService.findByName(user.name).eventDate
return "Hi ${user.name}, you have logged in since $lastLoginTime"
}
复制代码
再尝试用 RxJava 的方式来编写该接口：
@GetMapping("/rxjava/{username}")
fun getRxLoginMessage(@PathVariable username: String)=
userRxJavaService.findByName(username)
.map {
auditService.findByName(it.name).eventDate
}
.map {
"Hi ${username}, you have logged in since $it"
}
复制代码
最后，使用 Coroutines 的方式来编写接口：
@GetMapping("/coroutine/{username}")
fun getLoginMessage(@PathVariable username: String) = runBlocking {
val user = userRxJavaService.findByName(username).awaitSingle()
val lastLoginTime = GlobalScope.async {
auditService.findByName(user.name).eventDate
}.await()
"Hi ${user.name}, you have logged in since $lastLoginTime"
}
复制代码
能够看到，使用协程的方式相似于传统的 blocking 的方式来编写代码。
模拟用户登录：
使用 Coroutines 的方式获取登录信息：
关于协程，更多能够参考以前写的 Coroutines 笔记：
Kotlin Coroutines 笔记 (一）、Kotlin Coroutines 笔记 (二）
虽然 Kotlin 1.3 以后有些变更，可是大致是不变的。以后，也会整理更多 Kotlin Coroutines 笔记。
总结
响应式开发是将来的趋势，不管是服务端开发仍是移动端开发，都会顺应这个趋势。
另外，Kotlin 1.3 以后的协程已是正式版本，Kotlin 在语言级别上支持了协程，它是异步编程的另外一个不错的选择。
本文 demo 的 github 地址：github.com/fengzhizi71…
Java与Android技术栈：每周更新推送原创技术文章，欢迎扫描下方的公众号二维码并关注，期待与您的共同成长和进步。
相关文章
- 1. 使用 Kotlin + WebFlux/RxJava 2 实现响应式以及尝试正式版本的协程
- 2. Kotlin 使用 Spring WebFlux 实现响应式编程
- 3. Spring BOOT ( 基于Kotlin 编程语言) 使用 Spring WebFlux 实现响应式编程
- 4. 使用 media 实现响应式布局
- 5. vue3 的响应式实现
- 6. 最全面的Kotlin协程: Coroutine/Channel/Flow 以及实际应用
- 7. Android 正式版与测试版签名（MD5,SHA1,SHA256）以及忘记keystore的应用正式版签名获取
- 8. Vue 3 响应式原理及实现
- 9. 【Vue2】2. 文本响应式更新（响应式原理）
- 10. Kotlin 1.3正式版发布
- 更多相关文章...
- • MySQL的版本以及版本号 - MySQL教程
- • Mybatis实现映射器的2种方式 - MyBatis教程
- • ☆基于Java Instrument的Agent实现
- • 常用的分布式事务解决方案
相关标签/搜索
每日一句
-
每一个你不满意的现在，都有一个你没有努力的曾经。
最新文章
- 1. 在windows下的虚拟机中,安装华为电脑的deepin操作系统
- 2. 强烈推荐款下载不限速解析神器
- 3. 【区块链技术】孙宇晨：区块链技术带来金融服务的信任变革
- 4. 搜索引起的链接分析-计算网页的重要性
- 5. TiDB x 微众银行 | 耗时降低 58%，分布式架构助力实现普惠金融
- 6. 《数字孪生体技术白皮书》重磅发布（附完整版下载）
- 7. 双十一“避坑”指南：区块链电子合同为电商交易保驾护航！
- 8. 区块链产业，怎样“链”住未来？
- 9. OpenglRipper使用教程
- 10. springcloud请求一次好用一次不好用zuul Name or service not known
欢迎关注本站公众号,获取更多信息
相关文章
- 1. 使用 Kotlin + WebFlux/RxJava 2 实现响应式以及尝试正式版本的协程
- 2. Kotlin 使用 Spring WebFlux 实现响应式编程
- 3. Spring BOOT ( 基于Kotlin 编程语言) 使用 Spring WebFlux 实现响应式编程
- 4. 使用 media 实现响应式布局
- 5. vue3 的响应式实现
- 6. 最全面的Kotlin协程: Coroutine/Channel/Flow 以及实际应用
- 7. Android 正式版与测试版签名（MD5,SHA1,SHA256）以及忘记keystore的应用正式版签名获取
- 8. Vue 3 响应式原理及实现
- 9. 【Vue2】2. 文本响应式更新（响应式原理）
- 10. Kotlin 1.3正式版发布