java客户端与服务端交互通用处理框架解析

一、综述

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框架，提升发送性能

巴特西

java客户端与服务端交互通用处理框架解析

最新文章

热门文章

巴特西

java客户端与服务端交互通用处理 框架解析

最新文章

热门文章

java客户端与服务端交互通用处理框架解析