Vetx.x : EventLoop线程不要锁（Synchronized/ReentrantLock）
vert.x 里面有eventloop线程，而且eventloop线程是绝对不能阻塞的，
但是实际使用中，有时候我们对阻塞理解不深刻，所以一些隐藏的阻塞
没有那么好发现。
eventloop 线程和work线程共用相同的锁
下面的代码是一个简单的例子，用户请求localhost:8133/sync的时候
会在work现场锁sb对象，然后在eventloop线程也要获取sb这个对象的锁。
那么，我们发现，这样eventloop就会超时，可以看下日志。
package concurrency.file.vertx;
import io.vertx.core.AbstractVerticle;
import io.vertx.core.Future;
import io.vertx.core.Vertx;
import io.vertx.core.VertxOptions;
import io.vertx.ext.web.Router;
import io.vertx.ext.web.handler.BodyHandler;
/** * work 线程中有锁 * * @author kyle */
public class SynchronizedTest {
public static class Inner extends AbstractVerticle {
StringBuilder sb = new StringBuilder();
public void write(String str) {
synchronized (sb) {
sb.delete(0, sb.length());
sb.append(str);
}
}
@Override
public void start(Future<Void> future) throws Exception {
Router route = Router.router(vertx);
route.route().handler(BodyHandler.create());
route.route("/sync").handler(context1 -> {
vertx.executeBlocking(f -> {
foo();
}, f -> {
});
vertx.setTimer(1, f -> {
write(context1.getBodyAsString());
context1.response().end();
});
});
vertx.createHttpServer()
.requestHandler(route::accept)
.listen(8133, ar -> {
if (ar.failed()) future.fail(ar.cause());
else future.complete();
});
}
private void foo() {
synchronized (sb) {
try {
Thread.sleep(5000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
public static void main(String[] args) {
VertxOptions vertxOptions = new VertxOptions();
vertxOptions.setMaxEventLoopExecuteTime(20_000_000);
Vertx vertx = Vertx.vertx(vertxOptions);
vertx.deployVerticle(new Inner(), ar -> {
for (int i = 0; i < 50; i++) {
vertx.createHttpClient()
.getNow(8133, "localhost", "/sync", result -> {
System.out.println("finish");
});
}
});
}
}
日志如下
这是由于两个线程同时抢夺一个锁，如果eventloop线程拿不到锁，它就会阻塞。
仅eventloop中有锁
下面是上面的代码的修改，去掉了work线程对锁的把持。但是这样写代码依然有问题，如果多个eventloop线程
抢夺一个锁，必定会造成其他的eventloop线程的阻塞。所以也不要使用这种写法
package concurrency.file.vertx;
import io.vertx.core.AbstractVerticle;
import io.vertx.core.Future;
import io.vertx.core.Vertx;
import io.vertx.core.VertxOptions;
import io.vertx.ext.web.Router;
import io.vertx.ext.web.handler.BodyHandler;
/** * work 线程中有锁 * * @author kyle */
public class SynchronizedTest {
public static class Inner extends AbstractVerticle {
StringBuilder sb = new StringBuilder();
public void write(String str) {
synchronized (sb) {
sb.delete(0, sb.length());
sb.append(str);
}
}
@Override
public void start(Future<Void> future) throws Exception {
Router route = Router.router(vertx);
route.route().handler(BodyHandler.create());
route.route("/sync").handler(context1 -> {
write(context1.getBodyAsString());
context1.response().end();
});
vertx.createHttpServer()
.requestHandler(route::accept)
.listen(8133, ar -> {
if (ar.failed()) future.fail(ar.cause());
else future.complete();
});
}
}
public static void main(String[] args) {
VertxOptions vertxOptions = new VertxOptions();
vertxOptions.setMaxEventLoopExecuteTime(20_000_000);
Vertx vertx = Vertx.vertx(vertxOptions);
vertx.deployVerticle(new Inner(), ar -> {
for (int i = 0; i < 50; i++) {
vertx.createHttpClient()
.getNow(8133, "localhost", "/sync", result -> {
System.out.println("finish");
});
}
});
}
}
由于这种现象需要高并发的情况下，才容易观察出问题，所以这种写法，经常会被我们忽略。
看不到的synchronize
看下面的一段代码，是不是觉得没有用到锁，不会阻塞线 。其实不是的，由于Files.write底层仍然使用了synchronized 锁，
所以也会阻塞 eventloop线程，这种情况下，我们很难发现和察觉。
package concurrency.file.vertx;
import io.vertx.core.AbstractVerticle;
import io.vertx.core.Future;
import io.vertx.core.Vertx;
import io.vertx.core.VertxOptions;
import io.vertx.ext.web.Router;
import io.vertx.ext.web.handler.BodyHandler;
/** * work 线程中有锁 * * @author kyle */
public class SynchronizedTest {
public static class Inner extends AbstractVerticle {
public void write(String str) {
Files.write(Path.get("temp"), str.getBytes());
}
@Override
public void start(Future<Void> future) throws Exception {
Router route = Router.router(vertx);
route.route().handler(BodyHandler.create());
route.route("/sync").handler(context1 -> {
write(context1.getBodyAsString());
context1.response().end();
});
vertx.createHttpServer()
.requestHandler(route::accept)
.listen(8133, ar -> {
if (ar.failed()) future.fail(ar.cause());
else future.complete();
});
}
}
public static void main(String[] args) {
VertxOptions vertxOptions = new VertxOptions();
vertxOptions.setMaxEventLoopExecuteTime(20_000_000);
Vertx vertx = Vertx.vertx(vertxOptions);
vertx.deployVerticle(new Inner(), ar -> {
for (int i = 0; i < 50; i++) {
vertx.createHttpClient()
.getNow(8133, "localhost", "/sync", result -> {
System.out.println("finish");
});
}
});
}
}
不用锁，我该怎么办
我个人觉得，在Eventloop线程中是绝对不能用锁的，各种锁都不行，比如synchronize ,reentrantLock, CountdownLatch, CycleBarrier等等。那我们如何处理同步问题呢。我的一个思路是用单线程池，不用加锁，天然保持线程的同步。参考列子，如下。
package concurrency.file.vertx;
import io.vertx.core.AbstractVerticle;
import io.vertx.core.Future;
import io.vertx.core.Vertx;
import io.vertx.core.VertxOptions;
import io.vertx.ext.web.Router;
import io.vertx.ext.web.handler.BodyHandler;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
/** * work 线程中有锁 * * @author kyle */
public class SynchronizedTest {
public static class Inner extends AbstractVerticle {
ExecutorService executorService = Executors.newSingleThreadExecutor();
StringBuilder sb = new StringBuilder();
public Future<Void> write(String str) {
Future<Void> future = Future.future();
executorService.submit(() -> {
try {
Thread.sleep(5000);
} catch (InterruptedException e) {
e.printStackTrace();
}
sb.delete(0, sb.length());
sb.append(str);
});
return future;
}
@Override
public void start(Future<Void> future) throws Exception {
Router route = Router.router(vertx);
route.route().handler(BodyHandler.create());
route.route("/sync").handler(context1 -> {
write(context1.getBodyAsString());
context1.response().end();
});
vertx.createHttpServer()
.requestHandler(route::accept)
.listen(8133, ar -> {
if (ar.failed()) future.fail(ar.cause());
else future.complete();
});
}
}
public static void main(String[] args) {
VertxOptions vertxOptions = new VertxOptions();
vertxOptions.setMaxEventLoopExecuteTime(20_000_000);
Vertx vertx = Vertx.vertx(vertxOptions);
vertx.deployVerticle(new Inner(), ar -> {
for (int i = 0; i < 50; i++) {
vertx.createHttpClient()
.getNow(8133, "localhost", "/sync", result -> {
System.out.println("finish");
});
}
});
}
}