最近有个项目在压测，TPS有点低。做了一些日志异步批量落地和redis数据预热后，TPS稍微提高了点，但还是没达标。
研究了下项目的系统设计和技术栈。
用的分布式服务架构，其中服务端使用Netty。而客户端为了同步获取响应结果，采用了socket短连接模式。
为了使系统的性能有所提升，决定客户端也改用Netty框架并采用长连接的方式。
Netty是Jboss开源的一款非常优秀的异步通信框架。目前很多主流的开源项目有使用Netty开发的，Dubbo/RocketMQ/Apache Synapse等。
但Netty本身似乎没有提供同步等待响应的接口或方法。
使用过Dubbo的都知道我们平常使用的dubbo发布/订阅端就是同步获取响应报文，虽然它本身是基于Netty开发。
为了知道dubbo怎么实现netty响应结果异步转同步，看了点dubbo的源码。
发现dubbo是使用future+lock+condition实现的。这里就不作展开了，有兴趣可以看一下Dubbo的DefaultFuture这个类。
这里记录一下网上看到的一些异步转同步的方法。
首先定义一个业务操作类，纯粹处理业务。
/**
* 业务操作类
*/
public class TaskService {
public String getNumber() {
return UUID.randomUUID().toString();
}
}
然后写一个线程执行接口，
public interface Executor {
/**
* 异步执行
*/
default void supplyAsync() {}
/**
* 异步执行并回调
* @param callback
*/
default void supplyAsync(Callback callback) {}
/**
* 同步执行
*/
default void supplySync(){}
/**
* 回调接口
*/
@FunctionalInterface
public interface Callback{
void call(Object o);
}
}
接下来，写异步操作。
/**
* 异步任务
*/
public class AsyncExecutor implements Executor {
private TaskService service = new TaskService();
public void supplyAsync() {
this.supplyAsync(null);
}
public void supplyAsync(Callback callback) {
// 另起线程异步执行。
new Thread(() -> {
System.out.println("running async task...");
try {
// 子线程睡眠，主线程不休息。
Thread.sleep(2000);
} catch (InterruptedException e) {
}
String num = service.getNumber();
// 如果有回调，则调用回调函数。
if (callback != null){
callback.call(num);
}
}).start();
}
}
测试代码：
public static void main(String[] args) throws Exception {
System.err.println("
=============== start main thread ===============
");
Executor executor;
/**
* async
*/
executor = new AsyncExecutor();
executor.supplyAsync((o) -> {
System.out.println(String.format("get number[%s] by async.", o));
});
System.err.println("
=============== main thread over ===============
");
}
输出：
Connected to the target VM, address: '127.0.0.1:53015', transport: 'socket' =============== start main thread =============== =============== main thread over =============== running async task... get number[de720331-2ca3-4b33-8dee-4a7d26ede037] by async. Disconnected from the target VM, address: '127.0.0.1:53015', transport: 'socket'
我们可以看到，子线程在睡眠了2秒的时候，主线程并没有等待子线程执行完，而是继续往下执行。
接下来，我们要将异步响应改成同步的。
/**
* 同步执行抽象类
*/
public abstract class AbstractSyncExecutor implements Executor {
// 异步执行器
public AsyncExecutor executor = new AsyncExecutor();
/**
* 异步转同步
*/
public void supplySync() {
executor.supplyAsync(this::process);
await();
}
/**
* 线程等待
*/
protected void await() {}
/**
* 回调
* @param o
*/
protected abstract void process(Object o);
}
第一种：Synchronized + wait + notify
/**
* 采用synchronized配合wait和notify。
*/
public class SynchronizedExecutor extends AbstractSyncExecutor {
@Override
protected void process(Object o) {
System.out.println(String.format("get number[%s] by synchronized.", o));
synchronized (this) {
notify();
}
}
@Override
public void await() {
synchronized (this) {
try {
// 主线程调用wait阻塞等待，直到回调方法调用notify或者notifyAll唤醒。
wait();
} catch (InterruptedException e) {
}
}
}
}
测试代码
public static void main(String[] args) throws Exception {
System.err.println("
=============== start main thread ===============
");
Executor executor;
/**
* synchronized
*/
executor = new SynchronizedExecutor();
executor.supplySync();
System.err.println("
=============== main thread over ===============
");
}
执行结果
Connected to the target VM, address: '127.0.0.1:53235', transport: 'socket' =============== start main thread =============== running async task... get number[05c5db70-ec3b-4411-b080-21e5b4cddf79] by synchronized. =============== main thread over =============== Disconnected from the target VM, address: '127.0.0.1:53235', transport: 'socket'
可以看到内容已经顺序输出了。
第二种：reentrantLock + condition
/**
* 使用lock + condition
*/
public class ReentrantLockExecutor extends AbstractSyncExecutor {
private Lock lock = new ReentrantLock();
private Condition condition;
public ReentrantLockExecutor() {
this.condition = lock.newCondition();
}
@Override
protected void process(Object o) {
System.out.println(String.format("get number[%s] by lockAndCondition.", o));
lock.lock();
try {
condition.signal();
} finally {
lock.unlock();
}
}
@Override
protected void await() {
lock.lock();
try {
// 阻塞等待直到回调函数唤醒
condition.await();
} catch (Exception e) {
} finally {
lock.unlock();
}
}
}
测试代码
public static void main(String[] args) throws Exception {
System.err.println("
=============== start main thread ===============
");
Executor executor;
/**
* reentrantLock
*/
executor = new ReentrantLockExecutor();
executor.supplySync();
System.err.println("
=============== main thread over ===============
");
}
执行结果
Connected to the target VM, address: '127.0.0.1:53254', transport: 'socket' =============== start main thread =============== running async task... get number[6ad17c2c-a3ce-4fc6-a03a-fdb7d84acf7e] by lockAndCondition. =============== main thread over =============== Disconnected from the target VM, address: '127.0.0.1:53254', transport: 'socket'
第三种：countDownLatch
/**
* CountDownLatch
*/
public class CountDownLatchExecutor extends AbstractSyncExecutor {
// 假设每笔调用都创建一个CountDownLatchExecutor，那么从发起到响应只算一次操作，这里设置为1就可以了。
private CountDownLatch latch = new CountDownLatch(1);
@Override
public void process(Object o) {
System.out.println(String.format("get number[%s] by countDownLatch.", o));
// latch count - 1 变成0， 主线程继续执行
latch.countDown();
}
@Override
protected void await() {
try {
// 阻塞直到latch count=0
latch.await();
} catch (InterruptedException e) {
latch.countDown();
}
}
}
测试代码
public static void main(String[] args) throws Exception {
System.err.println("
=============== start main thread ===============
");
Executor executor;
/**
* countDownLatch
*/
executor = new CountDownLatchExecutor();
executor.supplySync();
System.err.println("
=============== main thread over ===============
");
}
执行结果
Connected to the target VM, address: '127.0.0.1:53325', transport: 'socket' =============== start main thread =============== running async task... get number[aa51089e-6f74-4ae7-b463-6a04ca73adcf] by countDownLatch. =============== main thread over =============== Disconnected from the target VM, address: '127.0.0.1:53325', transport: 'socket'
第四种：CyclicBarrier
/**
* CyclicBarrier
*/
public class CyclicBarrierExecutor extends AbstractSyncExecutor {
// 假设每笔调用都创建一个CountDownLatchExecutor
CyclicBarrier barrier = new CyclicBarrier(2);
@Override
protected void process(Object o) {
try {
System.out.println(String.format("get number[%s] by cyclicBarrier.", o));
// await线程数量=2，当前线程被唤醒
barrier.await();
} catch (Exception e) {
e.printStackTrace();
}
}
@Override
protected void await() {
try {
// await线程数为1，等待直至所有线程都到达
barrier.await();
} catch (Exception e) {
}
}
}
与CountDownLatch相反。CyclicBarrier是做加操作。当await线程达到初始parties数时，当前线程就被唤醒。我们需要在主线程await一次，回调线程await一次，然后主线程唤醒。即：CyclicBarrier的栅栏数parties设置为2。
测试代码
public static void main(String[] args) throws Exception {
System.err.println("
=============== start main thread ===============
");
Executor executor;
/**
* CyclicBarrier
*/
executor = new CyclicBarrierExecutor();
executor.supplySync();
System.err.println("
=============== main thread over ===============
");
}
执行结果
Connected to the target VM, address: '127.0.0.1:53527', transport: 'socket' =============== start main thread =============== running async task... get number[2409d6c4-7bb2-4838-a6ef-3ce42e4a18c7] by cyclicBarrier. =============== main thread over =============== Disconnected from the target VM, address: '127.0.0.1:53527', transport: 'socket'
第五种：Future + countDownLatch。
public class SyncFuture<T> implements Future<T> {
private CountDownLatch latch = new CountDownLatch(1);
private T resp;
@Override
public boolean cancel(boolean mayInterruptIfRunning) {
return false;
}
@Override
public boolean isCancelled() {
return false;
}
@Override
public boolean isDone() {
if (this.resp != null) {
return true;
}
return false;
}
@Override
public T get() throws InterruptedException, ExecutionException {
latch.await();
return this.resp;
}
@Override
public T get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
if (latch.await(timeout, unit)) {
return this.resp;
}
return null;
}
public void set(T resp) {
this.resp = resp;
latch.countDown();
}
}
public class FutureExecutor extends AbstractSyncExecutor {
private SyncFuture future;
public FutureExecutor(SyncFuture future) {
this.future = future;
}
@Override
public void process(Object o) {
future.set(o);
}
}
测试代码
public static void main(String[] args) throws Exception {
System.err.println("
=============== start main thread ===============
");
/**
* future + countDownLatch
*/
SyncFuture<String> future = new SyncFuture<>();
new FutureExecutor(future).supplySync();
// Object resp = futureExecutor.get();
// Object resp = futureExecutor.get(1, TimeUnit.SECONDS);
Object resp = future.get(3, TimeUnit.SECONDS);
System.out.println(String.format("get number[%s] by futureAndCountDownLatch.", resp));
System.err.println("
=============== main thread over ===============
");
}
执行结果
Connected to the target VM, address: '127.0.0.1:53590', transport: 'socket' =============== start main thread =============== running async task... get number[942e6c15-1eb2-4572-9a1f-dcb7d262387b] by futureAndCountDownLatch. =============== main thread over =============== Disconnected from the target VM, address: '127.0.0.1:53590', transport: 'socket'
前面四种方式都大同小异。主线程阻塞等待，子线程的回调函数里面唤醒主线程。
只有最后一种不太一样。而Dubbo就是使用最后一种方式，只是把其中的countDownLatch换成了condition+lock。