一、综述
java 客户端与服务端交互过程中，采用NIO通讯是异步的，客户端基本采用同一处理范式，来进行同异步的调用处理。
处理模型有以下几个要素：
1. NIO发送消息后返回的Future
2. 每次发送请求生成的Callback ，回调对象保存有请求数据，获取数据时阻塞线程，服务端返回时唤醒被阻塞的业务线程 并返回数据操作
3. 一个Map 保存有请求id 与 callback实例。 一般 key= reqId， value= callback
4. 一个TimeChecker 超时检测线程， 用户循环检测map里面的请求是否超时，超时的数据之间删除。
以上4个要素基本构成了目前客户端与服务端异步通讯时的处理模式。 目前dubbo、一些mq 框架都采用此模式，理解这个模式对阅读源码非常重要。
二、 处理流程图
流程说明：
1. 业务线程操作
1.1 通过NIO的channel ，write数据，同时返回 future
1.2 将req与future 组成callback实例，放入reqMap
1.3 调用callbak的get() 方法。此时线程会被阻塞一定时间，等待被唤醒（持有callback的锁）。
2. 发送消息后，开始监听返回消息
2.1 网络消息received事件，会触发listenser，根据reqId从reqMap里面获取callback实例，放入线程池执行
2.2 callback实例的 回调方法，标识结果已返回，设置response， 调用notifyAll()方法，唤醒在1.3 被阻塞的线程，返回
3. TimeoutCheckerThread线程
timeoutCheckerThread 负责轮询reqMap，将超时的数据从map里面删除。超时回调删除后，1.3步骤 被阻塞的线程睡眠醒来，就会抛出超时异常
以上几个步骤与流程，就是目前通用的 client 异步操作模式， 3个独立的线程+一个Map 完成整个操作。
在dubbo、各类mq 生产端，都是如此，部分可能有所差异。例如 dubbo的超时检测，用了HashedWheelTimer，比轮询效率更高，但本质不变
三、 实例代码
代码实例在idea 中运行通过，依赖lombock插件， netty3组件，详细代码请看 git：https://github.com/xujianguo1/practise/ 下的nettydemo目录。
对于netty的具体使用不做过多解读，毕竟netty4、5 与netty3 的差异太大。
Invoker接口
package com.luguo.nettydemo.client.handler; import com.luguo.nettydemo.model.RequestMsg; import com.luguo.nettydemo.model.AckMsg; public interface Invoker { /** * 同步调用，直接返回消息 */ public AckMsg invokeSync(RequestMsg request) throws Exception; /** * 异步调用，返回future */ public SimpleFuture invokeAsyc(RequestMsg requestMsg); /** * 收到消息返回时，调用。 */ public void invokeAck(AckMsg ackMsg); }
Invoker 接口默认实现 DefaultInvoker ，单例
package com.luguo.nettydemo.client.handler; import com.luguo.nettydemo.model.AckMsg; import com.luguo.nettydemo.model.RequestMsg; import lombok.extern.slf4j.Slf4j; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.atomic.AtomicLong; @Slf4j public class DefaultInvoker implements Invoker{ private long defaultTimeout = 1000; private static final Invoker invoker = new DefaultInvoker(); private AtomicLong sequencer = new AtomicLong(0);//id序列生成器 private RequestMap reqMap = new RequestMap(); //存放请求Id与回调 private DefaultInvoker(){ Thread timeoutChecker = new Thread(new TimeoutChecker()); timeoutChecker.setName("Timeout-Checker"); timeoutChecker.start(); //启动超时检查 } public static Invoker getInstance(){ return invoker; } public void invokeAck(AckMsg msg){ Long reqId = msg.getRequestId(); SimpleCallback callback = this.reqMap.getCallback(reqId); this.reqMap.remove(reqId); if(callback != null){ callback.setAckMsg(msg); callback.run(); //唤醒等待的线程 } } public void invokeCallback(RequestMsg request,SimpleCallback callback){ //NettyClient SimpleNettyClient client = SimpleNettyClient.getClient("local"); request.setRequestId(sequencer.addAndGet(1)); request.setSendTime(System.currentTimeMillis()); if(callback != null){ reqMap.putData(request.getRequestId(),callback); } client.write(request,callback); } private SimpleFuture invokeFuture(RequestMsg request){ CallbackFuture callbackFuture = new CallbackFuture(); callbackFuture.setRequestMsg(request); invokeCallback(request,callbackFuture); return callbackFuture; } public AckMsg invokeSync(RequestMsg request) throws Exception { SimpleFuture future = invokeFuture(request); return future.get(defaultTimeout); } public SimpleFuture invokeAsyc(RequestMsg requestMsg) { SimpleFuture future=invokeFuture(requestMsg); return future; } /** * 超时检测器 */ private class TimeoutChecker implements Runnable{ public void run(){ while(true){ try{ long now = System.currentTimeMillis(); for(Long reqId:reqMap.requestMap.keySet()){ SimpleCallback callback = reqMap.getCallback(reqId); if(callback.getRequestMsg().getSendTime() +defaultTimeout<now){//已经超时了 reqMap.remove(reqId); //删除超时的数据 log.warn("remove Timeout key="+reqId); } } Thread.sleep(1000); }catch (Exception e){ log.error(e.getMessage(),e); } } } } private class RequestMap{ /** * requestMap key=请求Id， value = 为消息与处理 */ private Map<Long,SimpleCallback> requestMap =new ConcurrentHashMap<Long,SimpleCallback>(); public SimpleCallback getCallback(Long requestId){ return requestMap.get(requestId); } public void putData(Long requestId,SimpleCallback callback){ requestMap.put(requestId,callback); } public void remove(Long requestId){ requestMap.remove(requestId); } } }
CallbackFuture 返回与回调对象
package com.luguo.nettydemo.client.handler; import com.luguo.nettydemo.model.AckMsg; import com.luguo.nettydemo.model.RequestMsg; import lombok.extern.slf4j.Slf4j; import org.jboss.netty.channel.ChannelFuture; @Slf4j public class CallbackFuture implements SimpleCallback,SimpleFuture { private RequestMsg reqMsg; private AckMsg ackMsg; private ChannelFuture future; private boolean isDone = false; public synchronized void run() { //回调方法被执行，表名已经完成了 isDone = true; this.notifyAll(); } public void setRequestMsg(RequestMsg msg){ this.reqMsg = msg; } public RequestMsg getRequestMsg(){ return reqMsg; } public void setAckMsg(AckMsg ack) { this.ackMsg = ack; } public SimpleFuture getFuture(ChannelFuture future) { this.future = future; return this; } public synchronized AckMsg get(long timeout) throws InterruptedException{ long sendTime = this.reqMsg.getSendTime(); while(!isDone){ long leftTime = timeout -(System.currentTimeMillis()-sendTime); if(leftTime <0){//抛出一个超时 throw new RuntimeException("Request timeout ! seqId:"+reqMsg.getRequestId()); }else{ log.info(this.reqMsg.getRequestId()+"需要睡眠时间："+leftTime); this.wait(leftTime); } } return ackMsg; } public boolean isDone() { return false; } }
ClientReceiveHandler： netty 的消息接收处理，会执行回调的run方法，唤醒等待的线程。
package com.luguo.nettydemo.client.handler; import com.luguo.nettydemo.model.AckMsg; import lombok.extern.slf4j.Slf4j; import org.jboss.netty.channel.ChannelHandlerContext; import org.jboss.netty.channel.MessageEvent; import org.jboss.netty.channel.SimpleChannelHandler; import java.util.concurrent.Executor; import java.util.concurrent.Executors; @Slf4j public class ClientReceiveHandler extends SimpleChannelHandler { private static Executor executor = Executors.newCachedThreadPool(); @Override public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) { final AckMsg ackMsg = (AckMsg) e.getMessage(); try { this.executor.execute(new Runnable() { @Override public void run() { //在线程池中执行回调方法 DefaultInvoker.getInstance().invokeAck(ackMsg); } }); } catch (Exception ex) { String msg = "ack callback execute fail \r\n"; log.error(msg + ex.getMessage(), ex); } } }
SimpleNettyClient ：netty的客户端启动，初始化连接服务端线程池。 write方法，将req与callback 放入map
@Slf4j public class SimpleNettyClient { private ClientBootstrap bootstrap; private ChannelPool channelPool; private static Map<String, SimpleNettyClient> clientMap= new ConcurrentHashMap<String, SimpleNettyClient>(); public SimpleNettyClient(){ bootstrap = new ClientBootstrap(new NioClientSocketChannelFactory( Executors.newCachedThreadPool(), Executors.newCachedThreadPool())); bootstrap.setPipelineFactory(new ChannelPipelineFactory() { public ChannelPipeline getPipeline() throws Exception { ChannelPipeline pipeline= Channels.pipeline(); pipeline.addLast("frameDecoder", new LengthFieldBasedFrameDecoder( Integer.MAX_VALUE, 0, 4, 0, 4)); pipeline.addLast("frameEncoder", new LengthFieldPrepender(4)); pipeline.addLast("jsonDecoder", new JsonDecoder(AckMsg.class)); pipeline.addLast("jsonEncoder", new JsonEncoder(RequestMsg.class)); pipeline.addLast("handler", new ClientReceiveHandler()); return pipeline; } }); channelPool = new ChannelPool(Constants.channelPoolSize); } private Channel connect(){ ChannelFuture future = bootstrap.connect(new InetSocketAddress(Constants.host, Constants.port)); // 等待连接创建成功 if (future.awaitUninterruptibly(3000, TimeUnit.MILLISECONDS)) { if (future.isSuccess()) { log.info("Client is conneted to " + Constants.host + ":" + Constants.port); } else { log.warn("Client is not conneted to " + Constants.host + ":" + Constants.port); } } return future.getChannel(); } public static SimpleNettyClient getClient(String clientName){ SimpleNettyClient client = clientMap.get(clientName);//这里可以扩展，进行负载均衡算法选择目标 if(client==null){ synchronized (clientMap){ if( clientMap.get(clientName)==null){//二次检查 client = new SimpleNettyClient(); clientMap.put(clientName,client); return client; } return clientMap.get(clientName); } }else{ return client; } } public SimpleFuture write(RequestMsg requestMsg, SimpleCallback callback){ Channel channel = this.channelPool.get(); if(channel==null){ channel = connect(); } ChannelFuture future = channel.write(requestMsg); this.channelPool.released(channel); // if(requestMsg.getMsgType() ==1){ // future.addListener(new ChannelFutureListener(){ // public void operationComplete(ChannelFuture channelFuture) throws Exception { // if(channelFuture.isSuccess()){ // return; // }else{ // //可以添加 写异常的返回 // } // } // }); // } if(callback != null){ callback.setRequestMsg(requestMsg); return callback.getFuture(future); } return null; } private class ChannelPool { private ArrayBlockingQueue<Channel> channels; public ChannelPool(int poolSize) { this.channels = new ArrayBlockingQueue<Channel>(poolSize); for (int i = 0; i < poolSize; i++) { channels.add(connect()); } } public Channel get(){ try{ return this.channels.take(); }catch (Exception e){ } return null; } /** * 同步释放netty channel */ public void released(Channel ch) { channels.add(ch); } } }
以上即为一个通用的客户端处理流程。 启动代码与req，ack，encoder、decoder 相关代码未贴出，在github上面下载即可。git：https://github.com/xujianguo1/practise/ 下的nettydemo目录。
测试时，先执行StartServer，启动服务端监听程序， 然后启动StartClientTest 客户端程序。
四、总结
网络客户端通讯，处理流程基本相同。
各大开源框架都是在该模式下实现功能的扩充，理解该模式，能很好的阅读各种存在客户端与服务端通讯的开源代码。
例如：1. 接入注册中心，实现server的发现，同时执行路由策略，实现负载均衡。
2. 发送与接收时，执行统计功能。
3. 在不关心返回 的场景（例如：mq的异步发送），上层会接入distruptor框架，提升发送性能
原文：https://www.cnblogs.com/keep-code/p/10769881.html