public class RPC {

    // serializable Invocation param for each rpc call
    /** A method invocation, including the method name and its parameters. */
    private static class Invocation implements Writable {
        private String methodName;
        private Class[] parameterClasses;
        private Object[] parameters;

        public Invocation() {
        }

        public Invocation(Method method, Object[] parameters) {
            this.methodName = method.getName();
            this.parameterClasses = method.getParameterTypes();
            this.parameters = parameters;
        }

        /** The name of the method invoked. */
        public String getMethodName() {
            return methodName;
        }

        /** The parameter classes. */
        public Class[] getParameterClasses() {
            return parameterClasses;
        }

        /** The parameter instances. */
        public Object[] getParameters() {
            return parameters;
        }

        public void readFields(DataInput in) throws IOException {
            methodName = in.readUTF();
            parameters = new Object[in.readInt()];
            parameterClasses = new Class[parameters.length];
            ObjectWritable objectWritable = new ObjectWritable();
            for (int i = 0; i < parameters.length; i++) {
                parameters[i] = ObjectWritable.readObject(in, objectWritable);
                parameterClasses[i] = objectWritable.getDeclaredClass();
            }
        }

        public void write(DataOutput out) throws IOException {
            out.writeUTF(methodName);
            out.writeInt(parameterClasses.length);
            for (int i = 0; i < parameterClasses.length; i++) {
                ObjectWritable.writeObject(out, parameters[i], parameterClasses[i]);
            }
        }

        public String toString() {
            StringBuffer buffer = new StringBuffer();
            buffer.append(methodName);
            buffer.append("(");
            for (int i = 0; i < parameters.length; i++) {
                if (i != 0)
                    buffer.append(", ");
                buffer.append(parameters[i]);
            }
            buffer.append(")");
            return buffer.toString();
        }
    }

    private static class Invoker implements InvocationHandler {
        private Client client;
        private boolean isClosed = false;

        public Invoker(Class<? extends VersionedProtocol> protocol, InetSocketAddress address, String ticket,
                SocketFactory factory, int rpcTimeout) throws IOException {
//          this.remoteId = Client.ConnectionId.getConnectionId(address, protocol, ticket, rpcTimeout);
//          this.client = CLIENTS.getClient(conf, factory);
        }

        public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {

            ObjectWritable value = (ObjectWritable)
                client.call(new Invocation(method, args), remoteId);
            return value.get();
        }
    }

    public static VersionedProtocol getProxy(Class<? extends VersionedProtocol> protocol, long clientVersion,
            InetSocketAddress addr, String ticket, SocketFactory factory, int rpcTimeout) throws Exception {

        VersionedProtocol proxy = (VersionedProtocol) Proxy.newProxyInstance(protocol.getClassLoader(),
                new Class[] { protocol }, new Invoker(protocol, addr, ticket, factory, rpcTimeout));
//      long serverVersion = proxy.getProtocolVersion(protocol.getName(), clientVersion);
//      if (serverVersion == clientVersion) {
//          return proxy;
//      } else {
//          throw new Exception("version mismatch");
//      }
        return proxy;
    }

    public boolean securityAuth(String ticket) {
        return true;
    }

    public static void main(String[] args) {
        try {
            PingProtocol pingProto = (PingProtocol) RPC.getProxy(PingProtocol.class, PingProtocol.versionID, null, null,
                    null, 0);
            NodeID nodeId = new NodeID("localhost", 1);

            // get name representing the running Java virtual machine.
            String name = ManagementFactory.getRuntimeMXBean().getName();
            System.out.println(name);
            // get pid
            String pid = name.split("@")[0];
            System.out.println("Pid is:" + pid);

            NodeContext nodeContext = new NodeContext(pid);
            pingProto.ping(nodeId, nodeContext);

        } catch (Exception e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
    }
}

# jsonrpcclient.py
import json

class RPCProxy:
    def __init__(self, connection):
        self._connection = connection
    def __getattr__(self, name):
        def do_rpc(*args, **kwargs):
            self._connection.send(json.dumps((name, args, kwargs)))
            result = json.loads(self._connection.recv())
            return result
        return do_rpc
>>> from multiprocessing.connection import Client
>>> c = Client(('localhost', 17000), authkey=b'peekaboo')
>>> proxy = RPCProxy(c)
>>> proxy.add(2, 3)

5
>>> proxy.sub(2, 3)
-1

# jsonrpcserver.py
import json

class RPCHandler:
    def __init__(self):
        self._functions = { }
    def register_function(self, func):
        self._functions[func.__name__] = func
    def handle_connection(self, connection):
        try:
            while True:
                # Receive a message
                func_name, args, kwargs = json.loads(connection.recv())
                # Run the RPC and send a response
                try:
                    r = self._functions[func_name](*args,**kwargs)
                    connection.send(json.dumps(r))
                except Exception as e:
                    connection.send(json.dumps(str(e)))
        except EOFError:
             pass

from multiprocessing.connection import Listener
from threading import Thread

def rpc_server(handler, address, authkey):
    sock = Listener(address, authkey=authkey)
    while True:
        client = sock.accept()
        t = Thread(target=handler.handle_connection, args=(client,))
        t.daemon = True
        t.start()

# Some remote functions
def add(x, y):
    return x + y

def sub(x, y):
    return x - y

# Register with a handler
handler = RPCHandler()
handler.register_function(add)
handler.register_function(sub)

# Run the server
rpc_server(handler, ('localhost', 17000), authkey=b'peekaboo')

public void run() {
    SERVER.set(Server.this);// ThreadLocal
    // 经典 Selector 循环事件监听模型
    while (running) {
        SelectionKey key = null;
        try {
            selector.select();
            Iterator<SelectionKey> iter = selector.selectedKeys().iterator();
            while (iter.hasNext()) {
                key = iter.next();
                iter.remove();
                try {
                    if (key.isValid()) {
                        if (key.isAcceptable())
                            doAccept(key);
                    }
                } catch (IOException e) {
                }
                key = null;
            }
    }
    //清除和停止各种资源和连接...
}

void doAccept(SelectionKey key) throws IOException,  OutOfMemoryError {
    Connection c = null;
    ServerSocketChannel server = (ServerSocketChannel) key.channel();
    SocketChannel channel;
    // 监听到 ACCEPT 事件以后，通过循环accept获取SocketChannel,可能多个新连接
    while ((channel = server.accept()) != null) {
        channel.configureBlocking(false);
        channel.socket().setTcpNoDelay(tcpNoDelay);
        Reader reader = getReader(); //循环获取一个reader
        try {
            reader.startAdd();// 若reader在select中阻塞，唤醒select方法，并置Reader到wait状态
            // 注册SocketChannel到reader.readSelector, 监听READ事件
            SelectionKey readKey = reader.registerChannel(channel);
            c = new Connection(readKey, channel, System.currentTimeMillis());
            readKey.attach(c);
            synchronized (connectionList) {
                connectionList.add(numConnections, c);
                numConnections++;
            }      
        } finally {
            reader.finishAdd(); // 将Reader从wait状态唤醒
        }
    }
}

void doRead(SelectionKey key) throws InterruptedException {
    int count = 0;
    Connection c = (Connection)key.attachment();
    if (c == null) {
        return;  
    }
    c.setLastContact(System.currentTimeMillis());

    count = c.readAndProcess();
}

public void run() {
    LOG.info("Starting SocketReader");
    synchronized (this) {
        while (running) {
            SelectionKey key = null;
            readSelector.select();
            // 被唤醒以后，需要等待注册事件完成，必须释放锁this，
            // 这样才能让其他线程调用registerChannel，并通过finishAdd唤醒等待或者超时
            while (adding) {
                this.wait(1000);
            }              
            // selector经典循环模式
            Iterator<SelectionKey> iter = readSelector.selectedKeys().iterator();
            while (iter.hasNext()) {
                key = iter.next();
                iter.remove();
                if (key.isValid()) {
                    if (key.isReadable()) {
                        doRead(key);
                    }
                }
                key = null;
            }
        }
    }
}

/**
* 让reader进入等待状态，该状态可以允许新连接注册到readSelector
* 如果reader正在select()上阻塞，则立即唤醒返回，否则，select()
* 的调用会立即返回,无论当前是否有read事件; 
* 该方法需要和finishAdd结合使用,用以唤醒reader从该等待状态退出
*/
public void startAdd() {
    adding = true;
    readSelector.wakeup();
}

public synchronized SelectionKey registerChannel(SocketChannel channel)
                                                    throws IOException {
    return channel.register(readSelector, SelectionKey.OP_READ);
}

public synchronized void finishAdd() {
    adding = false;
    this.notify();        
}

public void run() {
    SERVER.set(Server.this);
    ByteArrayOutputStream buf = 
        new ByteArrayOutputStream(INITIAL_RESP_BUF_SIZE);
    while (running) {
        final Call call = callQueue.take(); // pop the queue; maybe blocked here
        Writable value = null;
        CurCall.set(call);
        // Make the call as the user via Subject.doAs, thus associating
        // the call with the Subject
        if (call.connection.user == null) {
            value = call(call.connection.protocol, call.param, 
                        call.timestamp);
        } else {
            ...
        }
        CurCall.set(null);

        synchronized (call.connection.responseQueue) {
        // setupResponse() needs to be sync'ed together with 
        // responder.doResponse() since setupResponse may use
        // SASL to encrypt response data and SASL enforces
        // its own message ordering.
            setupResponse(buf, call, 
                        (error == null) ? Status.SUCCESS : Status.ERROR, 
                        value, errorClass, error);
            // Discard the large buf and reset it back to 
            // smaller size to freeup heap
            if (buf.size() > maxRespSize) {
                buf = new ByteArrayOutputStream(INITIAL_RESP_BUF_SIZE);
            }
            responder.doRespond(call);
        }
    }
}

    public void run() {
      SERVER.set(Server.this);
      long lastPurgeTime = 0;   // last check for old calls.
      while (running) {
        try {
          waitPending();     // 一个Channel正在注册的时候，必须等待
          writeSelector.select(PURGE_INTERVAL);
          Iterator<SelectionKey> iter = writeSelector.selectedKeys().iterator();
          while (iter.hasNext()) {
            SelectionKey key = iter.next();
            iter.remove();
            if (key.isValid() && key.isWritable()) {
                doAsyncWrite(key);
            }
          }
          long now = System.currentTimeMillis();
          if (now < lastPurgeTime + PURGE_INTERVAL) {
            continue;
          }
          lastPurgeTime = now;
          //长时间无法发送出去的Call，进行清除操作
          ArrayList<Call> calls;
          // get the list of channels from list of keys.
          synchronized (writeSelector.keys()) {
            calls = new ArrayList<Call>(writeSelector.keys().size());
            iter = writeSelector.keys().iterator();
            while (iter.hasNext()) {
              SelectionKey key = iter.next();
              Call call = (Call)key.attachment();
              if (call != null && key.channel() == call.connection.channel) { 
                calls.add(call);
              }
            }
          }

          for(Call call : calls) {
            doPurge(call, now);
          }
        } catch (OutOfMemoryError e) {
          // 这里不断的复制新Call，当由太多线程注册了事件的时候，新建数据量过大可能导致内存耗尽
          // 睡眠1分钟，给其他线程一点时间处理掉相关call
          try { Thread.sleep(60000); } catch (Exception ie) {}
        } catch (Exception e) {}
      }
    }

    private void doAsyncWrite(SelectionKey key) throws IOException {
      Call call = (Call)key.attachment();
      if (call == null) {
        return;
      }
      if (key.channel() != call.connection.channel) {
        throw new IOException("doAsyncWrite: bad channel");
      }

      synchronized(call.connection.responseQueue) {
        if (processResponse(call.connection.responseQueue, false)) {
          try {
            key.interestOps(0);
          } catch (CancelledKeyException e) {
            /* The Listener/reader might have closed the socket.
             * We don't explicitly cancel the key, so not sure if this will
             * ever fire.
             * This warning could be removed.
             */
          }
        }
      }
    }
    // Remove calls that have been pending in the responseQueue 
    // for a long time.
    private void doPurge(Call call, long now) throws IOException {
      LinkedList<Call> responseQueue = call.connection.responseQueue;
      synchronized (responseQueue) {
        Iterator<Call> iter = responseQueue.listIterator(0);
        while (iter.hasNext()) {
          call = iter.next();
          if (now > call.timestamp + PURGE_INTERVAL) {
            closeConnection(call.connection);
            break;
          }
        }
      }
    }

    private boolean processResponse(LinkedList<Call> responseQueue,
                                    boolean inHandler) throws IOException {
      boolean error = true;
      boolean done = false;       // there is more data for this channel.
      int numElements = 0;
      Call call = null;
      try {
        synchronized (responseQueue) {
          // If there are no items for this channel, then we are done
          numElements = responseQueue.size();
          if (numElements == 0) {
            error = false;
            return true;              // no more data for this channel.
          }
          // Extract the first call
          call = responseQueue.removeFirst();
          SocketChannel channel = call.connection.channel;
          // Send as much data as we can in the non-blocking fashion
          int numBytes = channelWrite(channel, call.response);
          if (numBytes < 0) {
            return true;
          }
          if (!call.response.hasRemaining()) {
            call.connection.decRpcCount();
            if (numElements == 1) {    // last call fully processes.
              done = true;             // no more data for this channel.
            } else {
              done = false;            // more calls pending to be sent.
            }
          } else {
            // 无法一次写完，则需要再次加入到表头，下次首先执行该Call的后面部分写操作
            call.connection.responseQueue.addFirst(call);
            // 如果是在Handler线程中调用的，则需要进行同步，
            // 让Responder线程进入等待状态，并唤醒writeSelector注册channel
            if (inHandler) {
              // set the serve time when the response has to be sent later
              call.timestamp = System.currentTimeMillis();   
              incPending();
              try {
                // 唤醒writeSelector, 这样 channel.register() 才能完成注册操作
                writeSelector.wakeup();
                channel.register(writeSelector, SelectionKey.OP_WRITE, call);
              } catch (ClosedChannelException e) {
                //Its ok. channel might be closed else where.
                done = true;
              } finally {
                decPending(); //唤醒Responder等待，此时Responder会继续执行
              }
            }
          }
          error = false;              // everything went off well
        }
      } finally {
        if (error && call != null) {
          done = true;               // error. no more data for this channel.
          closeConnection(call.connection);
        }
      }
      return done;
    }

    // 由Handler线程调用
    void doRespond(Call call) throws IOException {
      synchronized (call.connection.responseQueue) {
        call.connection.responseQueue.addLast(call);
        if (call.connection.responseQueue.size() == 1) {
          processResponse(call.connection.responseQueue, true);
        }
      }
    }

    private synchronized void incPending() {   // call waiting to be enqueued.
      pending++;
    }

    private synchronized void decPending() { // call done enqueueing.
      pending--;
      notify();
    }

    private synchronized void waitPending() throws InterruptedException {
      while (pending > 0) {
        wait();
      }
    }

/**
 * A polymorphic Writable that writes an instance with it's class name. Handles
 * arrays, strings and primitive types without a Writable wrapper.
 */
public class ObjectWritable implements Writable {

    private Class declaredClass;
    private Object instance;

    private static final Map<String, Class<?>> PRIMITIVE_NAMES = new HashMap<String, Class<?>>();
    static {
        PRIMITIVE_NAMES.put("boolean", Boolean.TYPE);
        PRIMITIVE_NAMES.put("byte", Byte.TYPE);
        PRIMITIVE_NAMES.put("char", Character.TYPE);
        PRIMITIVE_NAMES.put("short", Short.TYPE);
        PRIMITIVE_NAMES.put("int", Integer.TYPE);
        PRIMITIVE_NAMES.put("long", Long.TYPE);
        PRIMITIVE_NAMES.put("float", Float.TYPE);
        PRIMITIVE_NAMES.put("double", Double.TYPE);
        PRIMITIVE_NAMES.put("void", Void.TYPE);
    }

    private static class NullInstance implements Writable {
        private Class<?> declaredClass;

        public NullInstance() {
            ;
        }

        public NullInstance(Class declaredClass) {
            this.declaredClass = declaredClass;
        }

        public void readFields(DataInput in) throws IOException {
            String className = in.readUTF();
            declaredClass = PRIMITIVE_NAMES.get(className);
            if (declaredClass == null) {
                try {
                    declaredClass = Class.forName(className);
                } catch (ClassNotFoundException e) {
                    throw new RuntimeException(e.toString());
                }
            }
        }

        public void write(DataOutput out) throws IOException {
            out.writeUTF(declaredClass.getName());
        }
    }

    public static void writeObject(DataOutput out, Object instance, Class declaredClass) throws IOException {

        if (instance == null) { // null
            instance = new NullInstance(declaredClass);
            declaredClass = Writable.class;
        }

        out.writeUTF(declaredClass.getName()); // always write declared

        if (declaredClass.isArray()) { // array
            int length = Array.getLength(instance);
            out.writeInt(length);
            for (int i = 0; i < length; i++) {
                writeObject(out, Array.get(instance, i), declaredClass.getComponentType());
            }

        } else if (declaredClass == String.class) { // String
            out.writeUTF((String) instance);

        } else if (declaredClass.isPrimitive()) { // primitive type

            if (declaredClass == Boolean.TYPE) { // boolean
                out.writeBoolean(((Boolean) instance).booleanValue());
            } else if (declaredClass == Character.TYPE) { // char
                out.writeChar(((Character) instance).charValue());
            } else if (declaredClass == Byte.TYPE) { // byte
                out.writeByte(((Byte) instance).byteValue());
            } else if (declaredClass == Short.TYPE) { // short
                out.writeShort(((Short) instance).shortValue());
            } else if (declaredClass == Integer.TYPE) { // int
                out.writeInt(((Integer) instance).intValue());
            } else if (declaredClass == Long.TYPE) { // long
                out.writeLong(((Long) instance).longValue());
            } else if (declaredClass == Float.TYPE) { // float
                out.writeFloat(((Float) instance).floatValue());
            } else if (declaredClass == Double.TYPE) { // double
                out.writeDouble(((Double) instance).doubleValue());
            } else if (declaredClass == Void.TYPE) { // void
            } else {
                throw new IllegalArgumentException("Not a primitive: " + declaredClass);
            }
        } else if (declaredClass.isEnum()) { // enum
            out.writeUTF(((Enum) instance).name());
        } else if (Writable.class.isAssignableFrom(declaredClass)) { // Writable
            out.writeUTF(instance.getClass().getName());
            ((Writable) instance).write(out);

        } else {
            throw new IOException("Can't write: " + instance + " as " + declaredClass);
        }
    }

    public static Object readObject(DataInput in) throws IOException {
        return readObject(in, null);
    }

    public static Object readObject(DataInput in, ObjectWritable objectWritable) throws IOException {
        String className = in.readUTF();
        Class<?> declaredClass = PRIMITIVE_NAMES.get(className);
        if (declaredClass == null) {
            try {
                declaredClass = Class.forName(className);
            } catch (ClassNotFoundException e) {
                throw new RuntimeException("readObject can't find class " + className, e);
            }
        }

        Object instance;

        if (declaredClass.isPrimitive()) { // primitive types

            if (declaredClass == Boolean.TYPE) { // boolean
                instance = Boolean.valueOf(in.readBoolean());
            } else if (declaredClass == Character.TYPE) { // char
                instance = Character.valueOf(in.readChar());
            } else if (declaredClass == Byte.TYPE) { // byte
                instance = Byte.valueOf(in.readByte());
            } else if (declaredClass == Short.TYPE) { // short
                instance = Short.valueOf(in.readShort());
            } else if (declaredClass == Integer.TYPE) { // int
                instance = Integer.valueOf(in.readInt());
            } else if (declaredClass == Long.TYPE) { // long
                instance = Long.valueOf(in.readLong());
            } else if (declaredClass == Float.TYPE) { // float
                instance = Float.valueOf(in.readFloat());
            } else if (declaredClass == Double.TYPE) { // double
                instance = Double.valueOf(in.readDouble());
            } else if (declaredClass == Void.TYPE) { // void
                instance = null;
            } else {
                throw new IllegalArgumentException("Not a primitive: " + declaredClass);
            }

        } else if (declaredClass.isArray()) { // array
            int length = in.readInt();
            instance = Array.newInstance(declaredClass.getComponentType(), length);
            for (int i = 0; i < length; i++) {
                Array.set(instance, i, readObject(in));
            }

        } else if (declaredClass == String.class) { // String
            instance = in.readUTF();
        } else if (declaredClass.isEnum()) { // enum
            instance = Enum.valueOf((Class<? extends Enum>) declaredClass, in.readUTF());
        } else { // Writable
            Class instanceClass = null;
            String str = "";
            try {
                str = in.readUTF();
                instanceClass = Class.forName(str);
            } catch (ClassNotFoundException e) {
                throw new RuntimeException("readObject can't find class " + str, e);
            }

            Writable writable = (Writable) newInstance(instanceClass);
            writable.readFields(in);
            instance = writable;

            if (instanceClass == NullInstance.class) { // null
                declaredClass = ((NullInstance) instance).declaredClass;
                instance = null;
            }
        }

        if (objectWritable != null) { // store values
            objectWritable.declaredClass = declaredClass;
            objectWritable.instance = instance;
        }

        return instance;

    }
    private static final Map<Class<?>, Constructor<?>> CONSTRUCTOR_CACHE = 
            new ConcurrentHashMap<Class<?>, Constructor<?>>();


    public static <T> T newInstance(Class<T> theClass) {
        T result;
        try {
            Constructor<T> meth = (Constructor<T>) CONSTRUCTOR_CACHE.get(theClass);
            if (meth == null) {
                meth = theClass.getDeclaredConstructor();
                meth.setAccessible(true);
                CONSTRUCTOR_CACHE.put(theClass, meth);
            }
            result = meth.newInstance();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
        return result;
    }

}

  /**
   * When the read or write buffer size is larger than this limit, i/o will be 
   * done in chunks of this size. Most RPC requests and responses would be
   * be smaller.
   */
  private static int NIO_BUFFER_LIMIT = 8*1024; //should not be more than 64KB.

  /**
   * This is a wrapper around {@link WritableByteChannel#write(ByteBuffer)}.
   * If the amount of data is large, it writes to channel in smaller chunks. 
   * This is to avoid jdk from creating many direct buffers as the size of 
   * buffer increases. This also minimizes extra copies in NIO layer
   * as a result of multiple write operations required to write a large 
   * buffer.  
   *
   * @see WritableByteChannel#write(ByteBuffer)
   */
  private int channelWrite(WritableByteChannel channel, 
                           ByteBuffer buffer) throws IOException {

    int count =  (buffer.remaining() <= NIO_BUFFER_LIMIT) ?
                 channel.write(buffer) : channelIO(null, channel, buffer);
    if (count > 0) {
      rpcMetrics.incrSentBytes(count);
    }
    return count;
  }


  /**
   * This is a wrapper around {@link ReadableByteChannel#read(ByteBuffer)}.
   * If the amount of data is large, it writes to channel in smaller chunks. 
   * This is to avoid jdk from creating many direct buffers as the size of 
   * ByteBuffer increases. There should not be any performance degredation.
   * 
   * @see ReadableByteChannel#read(ByteBuffer)
   */
  private int channelRead(ReadableByteChannel channel, 
                          ByteBuffer buffer) throws IOException {

    int count = (buffer.remaining() <= NIO_BUFFER_LIMIT) ?
                channel.read(buffer) : channelIO(channel, null, buffer);
    if (count > 0) {
      rpcMetrics.incrReceivedBytes(count);
    }
    return count;
  }

  private static int channelIO(ReadableByteChannel readCh, 
                               WritableByteChannel writeCh,
                               ByteBuffer buf) throws IOException {

    int originalLimit = buf.limit();
    int initialRemaining = buf.remaining();
    int ret = 0;

    while (buf.remaining() > 0) {
      try {
        int ioSize = Math.min(buf.remaining(), NIO_BUFFER_LIMIT);
        buf.limit(buf.position() + ioSize);

        ret = (readCh == null) ? writeCh.write(buf) : readCh.read(buf); 

        if (ret < ioSize) {
          break;
        }

      } finally {
        buf.limit(originalLimit);        
      }
    }

    int nBytes = initialRemaining - buf.remaining(); 
    return (nBytes > 0) ? nBytes : ret;
  }      
}

// return the RpcEngine configured to handle a protocol
static synchronized RpcEngine getProtocolEngine(Class<?> protocol,
    Configuration conf) {
    RpcEngine engine = PROTOCOL_ENGINES.get(protocol);
    if (engine == null) {
    Class<?> impl = conf.getClass(ENGINE_PROP+"."+protocol.getName(),
                                    WritableRpcEngine.class);
    engine = (RpcEngine)ReflectionUtils.newInstance(impl, conf);
    PROTOCOL_ENGINES.put(protocol, engine);
    }
    return engine;
}

public static <T> T getProxy(Class<T> protocol,
                            long clientVersion,
                            InetSocketAddress addr,
                            UserGroupInformation ticket,
                            Configuration conf,
                            SocketFactory factory,
                            int rpcTimeout) throws IOException {
    // ProtocolProxy 只是对真 Proxy 的封装，还需要再次调用 ProtocolProxy.getProxy()
    return getProtocolProxy(protocol, clientVersion, addr, ticket,
         conf, factory, rpcTimeout, null).getProxy();
}

public static <T> ProtocolProxy<T> getProtocolProxy(Class<T> protocol,
                            long clientVersion,
                            InetSocketAddress addr,
                            UserGroupInformation ticket,
                            Configuration conf,
                            SocketFactory factory,
                            int rpcTimeout,
                            RetryPolicy connectionRetryPolicy,
                            AtomicBoolean fallbackToSimpleAuth)
   throws IOException {
    if (UserGroupInformation.isSecurityEnabled()) {
    SaslRpcServer.init(conf);
    }
    return getProtocolEngine(protocol, conf).getProxy(protocol, clientVersion,
        addr, ticket, conf, factory, rpcTimeout, connectionRetryPolicy,
        fallbackToSimpleAuth);
}

public static class Call implements Schedulable {
    private final int callId;             // the client's call id
    private final int retryCount;        // the retry count of the call
    private final Writable rpcRequest;    // Serialized Rpc request from client
    private final Connection connection;  // connection to client
    private long timestamp;               // time received when response is null
                                          // time served when response is not null
    private ByteBuffer rpcResponse;       // the response for this call
    private AtomicInteger responseWaitCount = new AtomicInteger(1);
    private final RPC.RpcKind rpcKind;
    private final byte[] clientId;
    private final Span traceSpan; // the tracing span on the server side

    private Call(Call call) {
      this(call.callId, call.retryCount, call.rpcRequest, call.connection,
          call.rpcKind, call.clientId, call.traceSpan);
    }
}

// RpcClientFactoryPBImpl
public Object getClient(Class<?> protocol, long clientVersion,
      InetSocketAddress addr, Configuration conf) {
    Constructor<?> constructor = cache.get(protocol);
    if (constructor == null) {
        Class<?> pbClazz = null;
        pbClazz = localConf.getClassByName(getPBImplClassName(protocol));
        constructor = pbClazz.getConstructor(Long.TYPE, InetSocketAddress.class, Configuration.class);
        constructor.setAccessible(true);
        cache.putIfAbsent(protocol, constructor);
    }
    Object retObject = constructor.newInstance(clientVersion, addr, conf);
}

// ContainerManagementProtocolPBClientImpl
public ContainerManagementProtocolPBClientImpl(long clientVersion,
      InetSocketAddress addr, Configuration conf) throws IOException {
    RPC.setProtocolEngine(conf, ContainerManagementProtocolPB.class,
      ProtobufRpcEngine.class);
    UserGroupInformation ugi = UserGroupInformation.getCurrentUser();

    int expireIntvl = conf.getInt(NM_COMMAND_TIMEOUT, DEFAULT_COMMAND_TIMEOUT);
    proxy =
        (ContainerManagementProtocolPB) RPC.getProxy(ContainerManagementProtocolPB.class,
          clientVersion, addr, ugi, conf,
          NetUtils.getDefaultSocketFactory(conf), expireIntvl);
  }

// RPCServerFactoryImpl
private Server createServer(Class<?> pbProtocol, InetSocketAddress addr, Configuration conf, 
      SecretManager<? extends TokenIdentifier> secretManager, int numHandlers, 
      BlockingService blockingService, String portRangeConfig) throws IOException {
    RPC.setProtocolEngine(conf, pbProtocol, ProtobufRpcEngine.class);
    RPC.Server server = new RPC.Builder(conf).setProtocol(pbProtocol)
        .setInstance(blockingService).setBindAddress(addr.getHostName())
        .setPort(addr.getPort()).setNumHandlers(numHandlers).setVerbose(false)
        .setSecretManager(secretManager).setPortRangeConfig(portRangeConfig)
        .build();
    LOG.info("Adding protocol "+pbProtocol.getCanonicalName()+" to the server");
    server.addProtocol(RPC.RpcKind.RPC_PROTOCOL_BUFFER, pbProtocol, blockingService);
    return server;
  }