Vert.x Kotlin 協程
協程
輕量級線程，相比傳統Java線程，它不會阻塞內核線程，並且能夠讓異步的代碼寫起來和同步代碼同樣舒服。這是官方的文檔 kotlinphp
pom.xml
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>com.xiaoniu.kt</groupId>
<artifactId>vertx-kt</artifactId>
<version>3.5.0</version>
<properties>
<slf4j.version>1.7.21</slf4j.version>
<kotlin.version>1.2.0</kotlin.version>
<maven.compiler.source>1.8</maven.compiler.source>
<maven.compiler.target>1.8</maven.compiler.target>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
<log4j.version>2.8.2</log4j.version>
</properties>
<dependencies>
<dependency>
<groupId>io.vertx</groupId>
<artifactId>vertx-lang-kotlin</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.jetbrains.kotlin</groupId>
<artifactId>kotlin-stdlib</artifactId>
<version>${kotlin.version}</version>
</dependency>
<dependency>
<groupId>io.vertx</groupId>
<artifactId>vertx-lang-kotlin-coroutines</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>io.vertx</groupId>
<artifactId>vertx-web</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>io.vertx</groupId>
<artifactId>vertx-web-client</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>io.vertx</groupId>
<artifactId>vertx-mysql-postgresql-client</artifactId>
<version>${project.version}</version>
</dependency>
<!-- log4j2 日誌配置 start -->
<dependency>
<groupId>org.apache.logging.log4j</groupId>
<artifactId>log4j-api</artifactId>
<version>${log4j.version}</version>
</dependency>
<dependency>
<groupId>org.apache.logging.log4j</groupId>
<artifactId>log4j-core</artifactId>
<version>${log4j.version}</version>
</dependency>
<dependency>
<groupId>org.apache.logging.log4j</groupId>
<artifactId>log4j-slf4j-impl</artifactId>
<version>${log4j.version}</version>
</dependency>
<dependency>
<groupId>org.apache.logging.log4j</groupId>
<artifactId>log4j-jcl</artifactId>
<version>${log4j.version}</version>
</dependency>
<dependency>
<groupId>org.apache.logging.log4j</groupId>
<artifactId>log4j-jul</artifactId>
<version>${log4j.version}</version>
</dependency>
<dependency>
<groupId>com.lmax</groupId>
<artifactId>disruptor</artifactId>
<version>3.3.6</version>
</dependency>
<!-- log4j2 日誌配置 end -->
</dependencies>
<build>
<sourceDirectory>${project.basedir}/src/main/kotlin</sourceDirectory>
<plugins>
<plugin>
<artifactId>kotlin-maven-plugin</artifactId>
<groupId>org.jetbrains.kotlin</groupId>
<version>${kotlin.version}</version>
<configuration>
<jvmTarget>1.8</jvmTarget>
</configuration>
<executions>
<execution>
<id>compile</id>
<goals>
<goal>compile</goal>
</goals>
</execution>
</executions>
</plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-shade-plugin</artifactId>
<version>2.3</version>
<executions>
<!-- Run shade goal on package phase -->
<execution>
<phase>package</phase>
<goals>
<goal>shade</goal>
</goals>
<configuration>
<finalName>kotlin-coroutines-examples</finalName>
<createDependencyReducedPom>false</createDependencyReducedPom>
<transformers>
<transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
<manifestEntries>
<Main-Class>io.vertx.core.Launcher</Main-Class>
<Main-Verticle>com.xiaoniu.kt.RestVerticle</Main-Verticle>
</manifestEntries>
</transformer>
<transformer implementation="org.apache.maven.plugins.shade.resource.AppendingTransformer">
<resource>META-INF/services/io.vertx.core.spi.VerticleFactory</resource>
</transformer>
</transformers>
</configuration>
</execution>
</executions>
</plugin>
</plugins>
</build>
<profiles>
<profile>
<id>staging</id>
<repositories>
<repository>
<id>staging</id>
<url>https://oss.sonatype.org/content/repositories/snapshots/</url>
</repository>
</repositories>
</profile>
</profiles>
</project>
複製代碼
首先來一個普通版的Rest服務
import io.vertx.core.AbstractVerticle
import io.vertx.core.Handler
import io.vertx.core.Vertx
import io.vertx.core.http.HttpMethod
import io.vertx.core.logging.Log4j2LogDelegateFactory
import io.vertx.core.logging.LoggerFactory
import io.vertx.ext.web.Router
import io.vertx.ext.web.RoutingContext
import io.vertx.ext.web.handler.BodyHandler
import io.vertx.ext.web.handler.CorsHandler
import org.apache.logging.log4j.LogManager
import java.time.LocalDateTime
/** * Created by sweet on 2017/12/20. * --------------------------- */
fun main(args: Array<String>) {
System.setProperty("vertx.logger-delegate-factory-class-name",
"io.vertx.core.logging.Log4j2LogDelegateFactory")
Vertx.vertx().deployVerticle(RestVerticle())
}
class RestVerticle : AbstractVerticle() {
val logger = LoggerFactory.getLogger(this.javaClass)
// val log = LogManager.getLogger(this.javaClass)
override fun start() {
var server = vertx.createHttpServer()
var router = Router.router(vertx)
// 綁定父路由
router.mountSubRouter("/api", router)
router.route("/*").handler({
println("能夠在此到處理權限的問題")
it.next()
}).handler(CorsHandler.create("*").allowedMethod(HttpMethod.GET) // 跨域處理
.allowedMethod(HttpMethod.POST)
.allowedMethod(HttpMethod.PUT)
.allowedMethod(HttpMethod.DELETE)
.allowedMethod(HttpMethod.OPTIONS)
.allowedHeader("X-PINGARUNER")
.allowedHeader("Content-Type"))
.failureHandler(errorHandler) // 給 / 開頭的路由綁定 錯誤處理器
// POST 獲取 請求體 必須設置
router.route().handler(BodyHandler.create().setUploadsDirectory("file-cache").setBodyLimit(5000000))
// /api/hello 子路由
router.get("/hello").handler({
it.response().end("Hello Api")
})
// /api/world 子路由
router.get("/world").handler({
1/0
it.response().end("World Api")
})
// /api/post 子路由
router.post("/post").handler({
logger.debug("/post")
var jsonObject = it.bodyAsJson
println("body: ${jsonObject.encode()}")
it.response().end(jsonObject.encodePrettily())
})
// /api/file 子路由 文件上傳
router.post("/file").handler({
it.fileUploads().forEach {
println("fileName: ${it.fileName()}")
println("name: ${it.name()}")
println("size: ${it.size()}")
println("uploadFileName: ${it.uploadedFileName()}\n")
}
it.response().end("OK")
})
server.requestHandler(router::accept).listen(config().getInteger("port", 8080), {
if (it.succeeded()) {
logger.info("Server start ${it.result().actualPort()}")
} else {
it.cause().printStackTrace()
logger.info("Server error ${it.cause().message}")
}
})
}
override fun stop() {
}
// 錯誤處理的兩種寫法, 相比 Java裏面的捕獲異常 在 catch中寫的代碼
fun errorFun(rx : RoutingContext) {
rx.failure().printStackTrace()
println("uri ${rx.request().uri()}")
rx.response().end("ERROR ${LocalDateTime.now()}")
}
val errorHandler = Handler<RoutingContext> {
it.failure().printStackTrace()
println("uri ${it.request().uri()}")
it.response().end("ERROR ${LocalDateTime.now()}")
}
//-------------------------------------------------------
}
複製代碼
協程版 Rest服務
import io.vertx.core.Handler
import io.vertx.core.Vertx
import io.vertx.core.logging.LoggerFactory
import io.vertx.ext.asyncsql.MySQLClient
import io.vertx.ext.sql.ResultSet
import io.vertx.ext.sql.SQLClient
import io.vertx.ext.web.Route
import io.vertx.ext.web.Router
import io.vertx.ext.web.RoutingContext
import io.vertx.ext.web.handler.BodyHandler
import io.vertx.kotlin.core.json.array
import io.vertx.kotlin.core.json.json
import io.vertx.kotlin.core.json.obj
import io.vertx.kotlin.coroutines.CoroutineVerticle
import io.vertx.kotlin.coroutines.awaitResult
import io.vertx.kotlin.coroutines.dispatcher
import kotlinx.coroutines.experimental.launch
import java.time.LocalDateTime
/** * 重點 演示對 MySQL 的操做， 利用 協程 寫出相似同步的代碼 * Created by sweet on 2017/12/21. * --------------------------- */
fun main(args: Array<String>) {
System.setProperty("vertx.logger-delegate-factory-class-name",
"io.vertx.core.logging.Log4j2LogDelegateFactory")
Vertx.vertx().deployVerticle(RestCoroutineVerticle())
}
class RestCoroutineVerticle : CoroutineVerticle() {
private val logger = LoggerFactory.getLogger(this.javaClass)
private lateinit var client: SQLClient
suspend override fun start() {
client = MySQLClient.createShared(vertx, json {
obj(
"host" to "127.0.0.1",
"port" to 3306,
"username" to "root",
"password" to "main",
"database" to "kt-demo"
)
})
val router = Router.router(vertx)
router.route().handler(BodyHandler.create())
.failureHandler(errorHandler)
router.get("/student/:id").handler { getStudent(it) }
router.post("/student").coroutineHandler { createStudent(it) }
router.delete("/student/:id").coroutineHandler { deleteStudent(it) }
router.put("/student/:id").coroutineHandler { updateStudent(it) }
router.get("/teacher/:id").coroutineHandler { getTeacher(it) }
router.get("/teachers").coroutineHandler { getTeachers(it) }
// 演示事務
router.post("/tx").coroutineHandler { tx(it) }
router.get("/test/error").handler({
1 / 0
it.response().end("OK")
})
vertx.createHttpServer()
.requestHandler(router::accept)
.listen(8080, {
if (it.succeeded()) {
logger.info("http server start !!! ${it.result().actualPort()}")
} else {
logger.error("http server error :: " + it.cause().message)
it.cause().printStackTrace()
}
})
}
private suspend fun tx(context: RoutingContext) {
var bodyAsJson = context.bodyAsJson
var school = bodyAsJson.getJsonArray("school")
var teacher = bodyAsJson.getJsonArray("teacher")
var connection = getConnection(client)
connection.use {
beginTx(connection)
try {
var schoolResult = updateWithParams(connection,
"INSERT INTO t_school (name) VALUES (?)",
json { array(school.list.get(0)) }
)
var teacherResult = updateWithParams(connection,
"INSERT INTO t_teacher (name, school_id) VALUES (?,?)",
json { teacher }
)
logger.debug("school result: ${schoolResult.toJson().encodePrettily()}")
logger.debug("teacher result: ${teacherResult.toJson().encodePrettily()}")
commitTx(connection)
context.response().end("ok")
} catch (e: Exception) {
e.printStackTrace()
rollbackTx(connection)
context.response().setStatusCode(500).end()
}
}
}
private suspend fun getTeachers(context: RoutingContext) {
val teacherResultSet = query(client,
"SELECT id, name, school_id FROM t_teacher"
)
var result = teacherResultSet?.getRows()
result.forEach {
var tId = it.getInteger("id")
var studentResultSet = queryWithsParams(client,
"SELECT name, age FROM t_student WHERE teacher_id = ?",
json { array(tId) }
)
var studentResult = studentResultSet?.getRows()
it.put("student", studentResult)
}
logger.debug(result.toString())
context.response().end(result.toString())
}
private suspend fun getTeacher(context: RoutingContext) {
val id = context.pathParam("id")
val teacherResultSet = queryWithsParams(client,
"SELECT id, name, school_id FROM t_teacher WHERE id = ?",
json { array(id) }
)
var result = teacherResultSet?.getRows()
result.forEach {
var tId = it.getInteger("id")
var studentResultSet = queryWithsParams(client,
"SELECT name, age FROM t_student WHERE teacher_id = ?",
json { array(tId) }
)
var studentResult = studentResultSet?.getRows()
it.put("student", studentResult)
}
logger.debug(result.toString())
context.response().end(result.toString())
}
private suspend fun updateStudent(context: RoutingContext) {
val id = context.pathParam("id")
val bodyAsJsonArray = context.bodyAsJsonArray
bodyAsJsonArray.add(id)
val result = updateWithParams(client,
"UPDATE t_student SET name = ?, age = ?, school_id = ?, teacher_id = ? WHERE id = ?",
json { bodyAsJsonArray }
)
context.response().end(result.toJson().encodePrettily())
}
private suspend fun deleteStudent(context: RoutingContext) {
val id = context.pathParam("id")
val result = updateWithParams(client,
"DELETE FROM t_student WHERE id = ?",
json { array(id) }
)
context.response().end(result.toJson().encodePrettily())
}
private suspend fun createStudent(context: RoutingContext) {
val bodyAsJsonArray = context.bodyAsJsonArray
val result = updateWithParams(client,
"INSERT INTO t_student (name, age, school_id, teacher_id) VALUES (?,?,?,?)",
json { bodyAsJsonArray }
)
context.response().end(result.toJson().encodePrettily())
}
private fun getStudent(context: RoutingContext) {
val id = context.pathParam("id")
launch(context.vertx().dispatcher()) {
val result = awaitResult<ResultSet> {
client.queryWithParams("SELECT id, name, age FROM t_student WHERE id = ?",
json { array(id) }, it)
}
if (result.rows.size == 1) {
context.response().end(result.getRows().get(0).encodePrettily())
} else {
context.response().setStatusCode(404).end()
}
}
}
val errorHandler = Handler<RoutingContext> {
it.failure().printStackTrace()
println("uri ${it.request().uri()}")
it.response().end("ERROR ${LocalDateTime.now()}")
}
}
fun Route.coroutineHandler(fn: suspend (RoutingContext) -> Unit) {
handler { ctx ->
launch(ctx.vertx().dispatcher()) {
try {
fn(ctx)
} catch (e: Exception) {
ctx.fail(e)
}
}
}
}
複製代碼
DbHandler.java
這裏面封裝了一些方便操做MySQL的函數， 函數是一等公民的感受真好。java
import io.vertx.core.json.JsonArray
import io.vertx.ext.sql.ResultSet
import io.vertx.ext.sql.SQLClient
import io.vertx.ext.sql.SQLConnection
import io.vertx.ext.sql.UpdateResult
import io.vertx.kotlin.coroutines.awaitResult
/** * Created by sweet on 2017/12/22. * --------------------------- */
suspend fun getConnection(client: SQLClient) : SQLConnection {
return awaitResult {
client.getConnection(it)
}
}
suspend fun queryWithParams(connection: SQLConnection, sql: String, args: JsonArray) : ResultSet {
return awaitResult {
connection.queryWithParams(sql, args, it)
}
}
suspend fun queryWithsParams(client: SQLClient, sql: String, args: JsonArray) : ResultSet {
return awaitResult {
client.queryWithParams(sql, args, it)
}
}
suspend fun query(connection: SQLConnection, sql: String) : ResultSet {
return awaitResult {
connection.query(sql, it)
}
}
suspend fun query(client: SQLClient, sql: String) : ResultSet {
return awaitResult {
client.query(sql, it)
}
}
suspend fun updateWithParams(connection: SQLConnection, sql: String, args: JsonArray) : UpdateResult {
return awaitResult {
connection.updateWithParams(sql, args, it)
}
}
suspend fun updateWithParams(client: SQLClient, sql: String, args: JsonArray) : UpdateResult {
return awaitResult {
client.updateWithParams(sql, args, it)
}
}
suspend fun update(connection: SQLConnection, sql: String) : UpdateResult {
return awaitResult {
connection.update(sql, it)
}
}
suspend fun update(client: SQLClient, sql: String) : UpdateResult {
return awaitResult {
client.update(sql, it)
}
}
suspend fun beginTx(connection: SQLConnection) {
awaitResult<Void> {
connection.setAutoCommit(false, it)
}
}
suspend fun commitTx(connection: SQLConnection) {
awaitResult<Void> {
connection.commit(it)
}
}
suspend fun rollbackTx(connection: SQLConnection) {
awaitResult<Void> {
connection.rollback(it)
}
}
suspend fun executeSQL(connection: SQLConnection, sql: String) {
awaitResult<Void> {
connection.execute(sql, it)
}
}
複製代碼
上面的代碼是對 MySQL的簡單操做，只是演示一下協程給代碼帶來的優點。若是不用協程組裝數據簡直是噩夢，要寫不少 Future，說一下 awaitResult 裏面的泛型，其實就是包在裏面的方法的回調函數裏的類型。多說無心，多寫寫就明白了，其實我看 Kotlin也才兩天，有哪些智障的寫法必定給我指出來，謝謝啦~mysql
SQL
create table t_school
(
id int auto_increment
primary key,
name varchar(60) not null
)
;
create table t_student
(
id int auto_increment
primary key,
name varchar(50) not null,
age int not null,
school_id int not null,
teacher_id int null
)
;
create table t_teacher
(
id int auto_increment
primary key,
name varchar(50) not null,
school_id int not null
)
;
create table t_user
(
id int auto_increment
primary key,
name varchar(50) not null,
age int not null
)
;
複製代碼
表是隨意建的，只是爲了演示一對多關係。web
直接運行 Main方法
打成 Jar 包
mvn clean package
複製代碼
java -Dvertx.logger-delegate-factory-class-name=io.vertx.core.logging.Log4j2LogDelegateFactory -jar xxx.jar就能夠正常運行。sql
相關文章
- 1. 「協程」和「Kotlin 協程」
- 2. 二、Kotlin-協程；
- 3. Kotlin Coroutines(協程)
- 4. Kotlin協程
- 5. Kotlin 之協程
- 6. Kotlin 協程總結
- 7. kotlin協程理解
- 8. kotlin 協程原理
- 9. Kotlin協程簡介
- 10. Kotlin協程嚮導
- 更多相關文章...
- • Kotlin 教程 - Kotlin 教程
- • Lua 協同程序(coroutine) - Lua 教程
- • Kotlin學習（一）基本語法
- • Kotlin學習（二）基本類型
相關標籤/搜索
每日一句
-
每一个你不满意的现在，都有一个你没有努力的曾经。
歡迎關注本站公眾號,獲取更多信息
相關文章
- 1. 「協程」和「Kotlin 協程」
- 2. 二、Kotlin-協程；
- 3. Kotlin Coroutines(協程)
- 4. Kotlin協程
- 5. Kotlin 之協程
- 6. Kotlin 協程總結
- 7. kotlin協程理解
- 8. kotlin 協程原理
- 9. Kotlin協程簡介
- 10. Kotlin協程嚮導