資訊專欄INFORMATION COLUMN
摘要：定時(shí)器例子之前通過調(diào)用定時(shí)器，需要傳一個(gè)回調(diào)，然后所有的代碼邏輯都包在里面。這里定時(shí)器會阻塞在這一行，直到一秒后才會執(zhí)行下面的一行。
之前介紹過quasar，如果你希望在vert.x項(xiàng)目里使用coroutine的話，建議使用vertx-sync。本篇將介紹vertx-sync。
本來打算另起一篇，寫其他方面的東西，但是最近比較忙，就先寫一篇實(shí)踐方面的文章。
vertx-sync是什么上一篇我們已經(jīng)講了 Fiber 相關(guān)的知識，想必大家對Java實(shí)現(xiàn)類似Golang的coroutine已經(jīng)有印象了，既然Java世界里有第三方提供了這么好的庫，那我們就看看怎么跟 vert.x 結(jié)合起來使用。
vert.x官方為了解決異步代碼編寫的困難，使之更加同步化對開發(fā)人員更友好，便基于quasar包裝了一個(gè)的同步庫，vertx-sync，該庫的作者同樣也是vert.x的原作者，所以完成度還是很高的。
vertx-sync 對外只是暴露了幾個(gè)簡單的靜態(tài)API，來完成對vert.x體系內(nèi)一系列的操作包裝，其實(shí)主要也就是三靜態(tài)API而已。
Sync.fiberHandler 如果你希望你的handler里有一些邏輯需要在Fiber里運(yùn)行，則你的handler必須用這個(gè)方法包一下。
Sync.awaitEvent 從Handler里返回一個(gè)事件結(jié)果(同步的)，且 不會阻塞EventLoop
Sync.awaitResult 從Handler里返回一個(gè)異步事件結(jié)果(同步的)，且 不會阻塞EventLoop
直接看介紹可能有點(diǎn)不直觀，下面跑幾個(gè)例子。
使用vertx-sync之前介紹過quasar，如果你希望在項(xiàng)目里使用coroutine的話，需要在JVM里設(shè)置一個(gè)參數(shù)，用于應(yīng)用啟動(dòng)前修改字節(jié)碼(注入一些中斷方法)，從而可以達(dá)到協(xié)程的目的。
具體方法也很簡單。
-javaagent:/path/to/the/quasar-core-0.7.5-jdk8.jar
如果是基于Maven跑單元測試，那只需要引用quasar instrument的插件就可以里
com.vlkan quasar-maven-plugin 0.7.3 true true true instrument co.paralleluniverse quasar-core 0.7.5
上面是一些非常必要的準(zhǔn)備工作，否則你無法使用quasar以及vertx-sync。
之前通過vert.x調(diào)用定時(shí)器，需要傳一個(gè)回調(diào)handler，然后所有的代碼邏輯都包在里面。
vertx.setTimer(1000L, h -> {
System.out.println("time"s up");
});
現(xiàn)在我們來重新塑造一下三觀。
awaitEvent(h -> vertx.setTimer(1000L, h));
System.out.println("time"s up");
這里定時(shí)器會阻塞在awaitEvent這一行，直到一秒后才會執(zhí)行下面的一行。有點(diǎn)類似執(zhí)行 Thread.sleep(1000L)，但是并不會阻塞 EventLoop 因?yàn)閝uasar會在EventLoop基礎(chǔ)之上再開啟一個(gè)fiber。
下面我看個(gè)稍微復(fù)雜點(diǎn)的例子。
我們先用傳統(tǒng)的回調(diào)方式使用vert.x的HttpClient API。
HttpClientRequest httpClientRequest = vertx.createHttpClient().get("leapcloud.cn");
httpClientRequest.handler(response -> {
response.handler(responseBody -> {
System.out.println(responseBody.toString());
});
}).end();
這里有兩層回調(diào)嵌套，一層是得到Http的Response，另一層是從Response里得到詳細(xì)的body。因?yàn)橛衛(wèi)ambda表達(dá)式才使得Java現(xiàn)在看起來并不是那么惡心。但是如果我們需要根據(jù)body的內(nèi)容進(jìn)一步做判斷去繼續(xù)請求其他頁面，則嵌套會變的非常的深。下面嘗試改造成sync方式看看。
HttpClientRequest httpClientRequest = vertx.createHttpClient().get("leapcloud.cn");
HttpClientResponse response = awaitEvent(Sync::fiberHandler);
Buffer body = awaitEvent(response::handler);
System.out.println(body.toString());
額，是不是感覺看著很舒服，無嵌套，直接順序下來，非常的直觀，加上Java8特有的方法引用，會讓代碼更精簡。
寫過vert.x同學(xué)肯定知道其vertx-jdbc-client為了使其兼容異步開發(fā)模型，將JDBC的底層線程池用異步方式包裝了一下，也就是說JDBC層還是通過線程池去訪問數(shù)據(jù)庫的，但是是通過vert.x的context做了層封裝，使其可以將結(jié)果放到對應(yīng)的 EventLoop 里，這樣比較符合vert.x的開發(fā)風(fēng)格。但是帶來的弊端就是嵌套太深。
final JDBCClient client = JDBCClient.createShared(vertx, new JsonObject()
.put("url", "jdbc:hsqldb:mem:test?shutdown=true")
.put("driver_class", "org.hsqldb.jdbcDriver")
.put("max_pool_size", 30));
client.getConnection(conn -> {
if (conn.failed()) {
System.err.println(conn.cause().getMessage());
return;
}
final SQLConnection connection = conn.result();
connection.execute("create table test(id int primary key, name varchar(255))", res -> {
if (res.failed()) {
throw new RuntimeException(res.cause());
}
// insert some test data
connection.execute("insert into test values(1, "Hello")", insert -> {
// query some data
connection.query("select * from test", rs -> {
for (JsonArray line : rs.result().getResults()) {
System.out.println(line.encode());
}
// and close the connection
connection.close(done -> {
if (done.failed()) {
throw new RuntimeException(done.cause());
}
});
});
});
});
});
上面代碼可以是不是有點(diǎn)惡心呢？嘗試改造一下吧。
final JDBCClient client = JDBCClient.createShared(vertx, new JsonObject()
.put("url", "jdbc:hsqldb:mem:test?shutdown=true")
.put("driver_class", "org.hsqldb.jdbcDriver")
.put("max_pool_size", 30));
try (SQLConnection conn = awaitResult(jdbc::getConnection)) {
awaitResult(h -> conn.execute("create table test(id int primary key, name varchar(255))", h));
awaitResult(h -> conn.execute("insert into test values(1, "Hello")", h));
ResultSet query = awaitResult(h -> conn.query("select * from test", h));
for (JsonArray line : query.result.getResults()) {
System.out.println(line.encode());
}
AsyncResult done = awaitResult(h -> conn.close(h));
if (done.failed()) {
throw new RuntimeException(done.cause())
}
} catch (Exception e) {
e.printStackTrace();
}
除了一個(gè)try catch，其他都沒有嵌套，整體邏輯的可讀性非常高，完全是線性的。
你也許會發(fā)現(xiàn)我們似乎一直都沒有用到 fiberHandler 這個(gè)靜態(tài)方法，上面雖然寫了定義，可能大家還是不能夠理解，這里結(jié)合場景也許能更好理解。
我們嘗試實(shí)現(xiàn)一個(gè)操作很耗時(shí)的邏輯然后包到fiber里，避免 EventLoop 被阻塞。這里你也許會很好奇，既然 Fiber 這么廉價(jià)開啟10萬8萬的無所謂啊，恩，這里再提一下quasar的重點(diǎn)部分: fiber可以很廉價(jià)的被創(chuàng)造出來，但是他本質(zhì)上還是跑在一個(gè)線程上面，如果其中一個(gè)fiber執(zhí)行了非常耗時(shí)的操作，則后面的fiber會一直等待，從而造成整個(gè)線程阻塞。 也就是說一個(gè)fiber不能執(zhí)行非常耗時(shí)的操作，比如計(jì)算100萬以內(nèi)的素?cái)?shù)之和，對于這種操作，我們可以通過直接將邏輯放到vert.x的worker線程里多帶帶去跑，然后通過fiber包裝一下就可以了。
